Merge branch 'main' into fix/issue_948

.github/workflows/code-testing.yml

@@ -119,6 +119,23 @@ jobs:
         run: pip install tox tox-gh-actions
       - name: "Run pytest via tox for ${{ matrix.python }}"
         run: tox
+  test-python-windows:
+    name: Pytest on 3.12 for windows
+    runs-on: windows-2022
+    needs: [lint-python, type-python]
+    env:
+      # Required to prevent asyncssh to fail.
+      USERNAME: WindowsUser
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.12
+      - name: Install dependencies
+        run: pip install tox tox-gh-actions
+      - name: Run pytest via tox for 3.12 on Windows
+        run: tox
   test-documentation:
     name: Build offline documentation for testing
     runs-on: ubuntu-20.04

anta/reporter/csv_reporter.py

@@ -107,7 +107,7 @@ def generate(cls, results: ResultManager, csv_filename: pathlib.Path) -> None:
         ]
 
         try:
-            with csv_filename.open(mode="w", encoding="utf-8") as csvfile:
+            with csv_filename.open(mode="w", encoding="utf-8", newline="") as csvfile:
                 csvwriter = csv.writer(
                     csvfile,
                     delimiter=",",

anta/runner.py

@@ -8,7 +8,7 @@
 import asyncio
 import logging
 import os
-import resource
+import sys
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any
 
@@ -26,35 +26,38 @@
     from anta.result_manager import ResultManager
     from anta.result_manager.models import TestResult
 
-logger = logging.getLogger(__name__)
+if os.name == "posix":
+    import resource
 
-DEFAULT_NOFILE = 16384
+    DEFAULT_NOFILE = 16384
 
+    def adjust_rlimit_nofile() -> tuple[int, int]:
+        """Adjust the maximum number of open file descriptors for the ANTA process.
 
-def adjust_rlimit_nofile() -> tuple[int, int]:
-    """Adjust the maximum number of open file descriptors for the ANTA process.
+        The limit is set to the lower of the current hard limit and the value of the ANTA_NOFILE environment variable.
 
-    The limit is set to the lower of the current hard limit and the value of the ANTA_NOFILE environment variable.
+        If the `ANTA_NOFILE` environment variable is not set or is invalid, `DEFAULT_NOFILE` is used.
 
-    If the `ANTA_NOFILE` environment variable is not set or is invalid, `DEFAULT_NOFILE` is used.
+        Returns
+        -------
+        tuple[int, int]
+            The new soft and hard limits for open file descriptors.
+        """
+        try:
+            nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE))
+        except ValueError as exception:
+            logger.warning("The ANTA_NOFILE environment variable value is invalid: %s\nDefault to %s.", exc_to_str(exception), DEFAULT_NOFILE)
+            nofile = DEFAULT_NOFILE
+
+        limits = resource.getrlimit(resource.RLIMIT_NOFILE)
+        logger.debug("Initial limit numbers for open file descriptors for the current ANTA process: Soft Limit: %s | Hard Limit: %s", limits[0], limits[1])
+        nofile = min(limits[1], nofile)
+        logger.debug("Setting soft limit for open file descriptors for the current ANTA process to %s", nofile)
+        resource.setrlimit(resource.RLIMIT_NOFILE, (nofile, limits[1]))
+        return resource.getrlimit(resource.RLIMIT_NOFILE)
 
-    Returns
-    -------
-    tuple[int, int]
-        The new soft and hard limits for open file descriptors.
-    """
-    try:
-        nofile = int(os.environ.get("ANTA_NOFILE", DEFAULT_NOFILE))
-    except ValueError as exception:
-        logger.warning("The ANTA_NOFILE environment variable value is invalid: %s\nDefault to %s.", exc_to_str(exception), DEFAULT_NOFILE)
-        nofile = DEFAULT_NOFILE
 
-    limits = resource.getrlimit(resource.RLIMIT_NOFILE)
-    logger.debug("Initial limit numbers for open file descriptors for the current ANTA process: Soft Limit: %s | Hard Limit: %s", limits[0], limits[1])
-    nofile = min(limits[1], nofile)
-    logger.debug("Setting soft limit for open file descriptors for the current ANTA process to %s", nofile)
-    resource.setrlimit(resource.RLIMIT_NOFILE, (nofile, limits[1]))
-    return resource.getrlimit(resource.RLIMIT_NOFILE)
+logger = logging.getLogger(__name__)
 
 
 def log_cache_statistics(devices: list[AntaDevice]) -> None:
@@ -167,7 +170,8 @@ def prepare_tests(
 
     if total_test_count == 0:
         msg = (
-            f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current test catalog and device inventory, please verify your inputs."
+            f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current "
+            "test catalog and device inventory, please verify your inputs."
         )
         logger.warning(msg)
         return None
@@ -247,9 +251,6 @@ async def main(
     dry_run
         Build the list of coroutine to run and stop before test execution.
     """
-    # Adjust the maximum number of open file descriptors for the ANTA process
-    limits = adjust_rlimit_nofile()
-
     if not catalog.tests:
         logger.info("The list of tests is empty, exiting")
         return
@@ -270,10 +271,19 @@ async def main(
             "--- ANTA NRFU Run Information ---\n"
             f"Number of devices: {len(inventory)} ({len(selected_inventory)} established)\n"
             f"Total number of selected tests: {final_tests_count}\n"
-            f"Maximum number of open file descriptors for the current ANTA process: {limits[0]}\n"
-            "---------------------------------"
         )
 
+        if os.name == "posix":
+            # Adjust the maximum number of open file descriptors for the ANTA process
+            limits = adjust_rlimit_nofile()
+            run_info += f"Maximum number of open file descriptors for the current ANTA process: {limits[0]}\n"
+        else:
+            # Running on non-Posix system, cannot manage the resource.
+            limits = (sys.maxsize, sys.maxsize)
+            run_info += "Running on a non-POSIX system, cannot adjust the maximum number of file descriptors.\n"
+
+        run_info += "---------------------------------"
+
         logger.info(run_info)
 
         if final_tests_count > limits[0]:

docs/faq.md

@@ -110,6 +110,17 @@ anta_title: Frequently Asked Questions (FAQ)
         pip install -U pyopenssl>22.0
         ```
 
+## Caveat running on non-POSIX platforms (e.g. Windows)
+
+???+ faq "Caveat running on non-POSIX platforms (e.g. Windows)"
+
+    While ANTA should in general work on non-POSIX platforms (e.g. Windows),
+    there are some known limitations:
+
+    - On non-Posix platforms, ANTA is not able to check and/or adjust the system limit of file descriptors.
+
+    ANTA test suite is being run in the CI on a Windows runner.
+
 ## `__NSCFConstantString initialize` error on OSX
 
 ???+ faq "`__NSCFConstantString initialize` error on OSX"

pyproject.toml

@@ -23,13 +23,13 @@ dependencies = [
   "asyncssh>=2.16",
   "cvprac>=1.3.1",
   "eval-type-backport>=0.1.3",  # Support newer typing features in older Python versions (required until Python 3.9 support is removed)
+  "httpx>=0.27.0",
   "Jinja2>=3.1.2",
   "pydantic>=2.7",
   "pydantic-extra-types>=2.3.0",
   "PyYAML>=6.0",
   "requests>=2.31.0",
-  "rich>=13.5.2,<14",
-  "httpx>=0.27.0"
+  "rich>=13.5.2,<14"
 ]
 keywords = ["test", "anta", "Arista", "network", "automation", "networking", "devops", "netdevops"]
 classifiers = [
@@ -259,6 +259,9 @@ extras =
 # tox -e <env> -- path/to/my/test::test
 commands =
    pytest {posargs}
+# To test on non-POSIX system
+# https://github.com/tox-dev/tox/issues/1455
+passenv = USERNAME
 
 [testenv:lint]
 description = Check the code style
@@ -467,7 +470,8 @@ disable = [ # Any rule listed here can be disabled: https://github.com/astral-sh
     "unnecessary-lambda",
     "abstract-class-instantiated", # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-instantiation-of-abstract-classes-abstract
     "unexpected-keyword-arg", # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-arguments-in-calls-call-arg and other rules
-    "no-value-for-parameter" # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-arguments-in-calls-call-arg
+    "no-value-for-parameter", # Overlap with https://mypy.readthedocs.io/en/stable/error_code_list.html#check-arguments-in-calls-call-arg
+    "import-outside-toplevel"
 ]
 max-statements=61
 max-returns=8

tests/units/reporter/test__init__.py

@@ -188,5 +188,5 @@ class TestReportJinja:
 
     def test_fail__init__file_not_found(self) -> None:
         """Test __init__ failure if file is not found."""
-        with pytest.raises(FileNotFoundError, match="template file is not found: /gnu/terry/pratchett"):
+        with pytest.raises(FileNotFoundError, match=r"template file is not found: [/|\\]gnu[/|\\]terry[/|\\]pratchett"):
             ReportJinja(Path("/gnu/terry/pratchett"))

tests/units/reporter/test_csv.py

@@ -8,6 +8,7 @@
 import csv
 import pathlib
 from typing import Any, Callable
+from unittest.mock import patch
 
 import pytest
 
@@ -49,8 +50,8 @@ def test_report_csv_generate(
         # Generate the CSV report
         ReportCsv.generate(result_manager, csv_filename)
 
-        # Read the generated CSV file
-        with pathlib.Path.open(csv_filename, encoding="utf-8") as csvfile:
+        # Read the generated CSV file - newline required on Windows..
+        with pathlib.Path.open(csv_filename, encoding="utf-8", newline="") as csvfile:
             reader = csv.reader(csvfile, delimiter=",")
             rows = list(reader)
 
@@ -82,11 +83,9 @@ def test_report_csv_generate_os_error(
         max_test_entries = 10
         result_manager = result_manager_factory(max_test_entries)
 
-        # Create a temporary CSV file path and make tmp_path read_only
-        tmp_path.chmod(0o400)
         csv_filename = tmp_path / "read_only.csv"
 
-        with pytest.raises(OSError, match="Permission denied"):
+        with patch("pathlib.Path.open", side_effect=OSError("Any OSError")), pytest.raises(OSError, match="Any OSError"):
             # Generate the CSV report
             ReportCsv.generate(result_manager, csv_filename)
 

tests/units/test_runner.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-import resource
+import os
 import sys
 from pathlib import Path
 from unittest.mock import patch
@@ -16,10 +16,16 @@
 from anta.catalog import AntaCatalog
 from anta.inventory import AntaInventory
 from anta.result_manager import ResultManager
-from anta.runner import adjust_rlimit_nofile, main, prepare_tests
+from anta.runner import main, prepare_tests
 
 from .test_models import FakeTest, FakeTestWithMissingTest
 
+if os.name == "posix":
+    # The function is not defined on non-POSIX system
+    import resource
+
+    from anta.runner import adjust_rlimit_nofile
+
 DATA_DIR: Path = Path(__file__).parent.parent.resolve() / "data"
 FAKE_CATALOG: AntaCatalog = AntaCatalog.from_list([(FakeTest, None)])
 
@@ -65,8 +71,10 @@ async def test_no_selected_device(caplog: pytest.LogCaptureFixture, inventory: A
     assert msg in caplog.messages
 
 
+@pytest.mark.skipif(os.name != "posix", reason="Cannot run this test on Windows")
 def test_adjust_rlimit_nofile_valid_env(caplog: pytest.LogCaptureFixture) -> None:
     """Test adjust_rlimit_nofile with valid environment variables."""
+    # pylint: disable=E0606
     with (
         caplog.at_level(logging.DEBUG),
         patch.dict("os.environ", {"ANTA_NOFILE": "20480"}),
@@ -96,6 +104,7 @@ def side_effect_setrlimit(resource_id: int, limits: tuple[int, int]) -> None:
         setrlimit_mock.assert_called_once_with(resource.RLIMIT_NOFILE, (20480, 1048576))
 
 
+@pytest.mark.skipif(os.name != "posix", reason="Cannot run this test on Windows")
 def test_adjust_rlimit_nofile_invalid_env(caplog: pytest.LogCaptureFixture) -> None:
     """Test adjust_rlimit_nofile with valid environment variables."""
     with (
@@ -129,6 +138,31 @@ def side_effect_setrlimit(resource_id: int, limits: tuple[int, int]) -> None:
         setrlimit_mock.assert_called_once_with(resource.RLIMIT_NOFILE, (16384, 1048576))
 
 
+@pytest.mark.skipif(os.name == "posix", reason="Run this test on Windows only")
+async def test_check_runner_log_for_windows(caplog: pytest.LogCaptureFixture, inventory: AntaInventory) -> None:
+    """Test log output for Windows host regarding rlimit."""
+    caplog.set_level(logging.INFO)
+    manager = ResultManager()
+    # Using dry-run to shorten the test
+    await main(manager, inventory, FAKE_CATALOG, dry_run=True)
+    assert "Running on a non-POSIX system, cannot adjust the maximum number of file descriptors." in caplog.records[-3].message
+
+
+# We could instead merge multiple coverage report together but that requires more work than just this.
+@pytest.mark.skipif(os.name != "posix", reason="Fake non-posix for coverage")
+async def test_check_runner_log_for_windows_fake(caplog: pytest.LogCaptureFixture, inventory: AntaInventory) -> None:
+    """Test log output for Windows host regarding rlimit."""
+    with patch("os.name", new="win32"):
+        del sys.modules["anta.runner"]
+        from anta.runner import main  # pylint: disable=W0621
+
+        caplog.set_level(logging.INFO)
+        manager = ResultManager()
+        # Using dry-run to shorten the test
+        await main(manager, inventory, FAKE_CATALOG, dry_run=True)
+        assert "Running on a non-POSIX system, cannot adjust the maximum number of file descriptors." in caplog.records[-3].message
+
+
 @pytest.mark.parametrize(
     ("inventory", "tags", "tests", "devices_count", "tests_count"),
     [