Merge branch 'main' into issue_822

.pre-commit-config.yaml

@@ -43,7 +43,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.7
+    rev: v0.6.8
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -52,7 +52,7 @@ repos:
       name: Run Ruff formatter
 
   - repo: https://github.com/pycqa/pylint
-    rev: "v3.3.0"
+    rev: "v3.3.1"
     hooks:
       - id: pylint
         name: Check code style with pylint

anta/catalog.py

@@ -296,11 +296,16 @@ def __init__(
             else:
                 self._filename = Path(filename)
 
-        # Default indexes for faster access
-        self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]] = defaultdict(set)
-        self.tests_without_tags: set[AntaTestDefinition] = set()
-        self.indexes_built: bool = False
-        self.final_tests_count: int = 0
+        self.indexes_built: bool
+        self.tag_to_tests: defaultdict[str | None, set[AntaTestDefinition]]
+        self._tests_without_tags: set[AntaTestDefinition]
+        self._init_indexes()
+
+    def _init_indexes(self) -> None:
+        """Init indexes related variables."""
+        self.tag_to_tests = defaultdict(set)
+        self._tests_without_tags = set()
+        self.indexes_built = False
 
     @property
     def filename(self) -> Path | None:
@@ -485,7 +490,7 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
 
         - tag_to_tests: A dictionary mapping each tag to a set of tests that contain it.
 
-        - tests_without_tags: A set of tests that do not have any tags.
+        - _tests_without_tags: A set of tests that do not have any tags.
 
         Once the indexes are built, the `indexes_built` attribute is set to True.
         """
@@ -499,11 +504,15 @@ def build_indexes(self, filtered_tests: set[str] | None = None) -> None:
                 for tag in test_tags:
                     self.tag_to_tests[tag].add(test)
             else:
-                self.tests_without_tags.add(test)
+                self._tests_without_tags.add(test)
 
-        self.tag_to_tests[None] = self.tests_without_tags
+        self.tag_to_tests[None] = self._tests_without_tags
         self.indexes_built = True
 
+    def clear_indexes(self) -> None:
+        """Clear this AntaCatalog instance indexes."""
+        self._init_indexes()
+
     def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> set[AntaTestDefinition]:
         """Return all tests that match a given set of tags, according to the specified strictness.
 

anta/runner.py

@@ -147,6 +147,7 @@ def prepare_tests(
     device_to_tests: defaultdict[AntaDevice, set[AntaTestDefinition]] = defaultdict(set)
 
     # Create AntaTestRunner tuples from the tags
+    final_tests_count = 0
     for device in inventory.devices:
         if tags:
             if not any(tag in device.tags for tag in tags):
@@ -159,9 +160,9 @@ def prepare_tests(
         # Add the tests with matching tags from device tags
         device_to_tests[device].update(catalog.get_tests_by_tags(device.tags))
 
-        catalog.final_tests_count += len(device_to_tests[device])
+        final_tests_count += len(device_to_tests[device])
 
-    if catalog.final_tests_count == 0:
+    if len(device_to_tests.values()) == 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."
         )
@@ -171,13 +172,15 @@ def prepare_tests(
     return device_to_tests
 
 
-def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]]) -> list[Coroutine[Any, Any, TestResult]]:
+def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinition]], manager: ResultManager) -> list[Coroutine[Any, Any, TestResult]]:
     """Get the coroutines for the ANTA run.
 
     Parameters
     ----------
     selected_tests
         A mapping of devices to the tests to run. The selected tests are generated by the `prepare_tests` function.
+    manager
+        A ResultManager
 
     Returns
     -------
@@ -189,6 +192,7 @@ def get_coroutines(selected_tests: defaultdict[AntaDevice, set[AntaTestDefinitio
         for test in test_definitions:
             try:
                 test_instance = test.test(device=device, inputs=test.inputs)
+                manager.add(test_instance.result)
                 coros.append(test_instance.test())
             except Exception as e:  # noqa: PERF203, BLE001
                 # An AntaTest instance is potentially user-defined code.
@@ -256,25 +260,26 @@ async def main(  # noqa: PLR0913
             selected_tests = prepare_tests(selected_inventory, catalog, tests, tags)
             if selected_tests is None:
                 return
+            final_tests_count = sum(len(tests) for tests in selected_tests.values())
 
         run_info = (
             "--- ANTA NRFU Run Information ---\n"
             f"Number of devices: {len(inventory)} ({len(selected_inventory)} established)\n"
-            f"Total number of selected tests: {catalog.final_tests_count}\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"
             "---------------------------------"
         )
 
         logger.info(run_info)
 
-        if catalog.final_tests_count > limits[0]:
+        if final_tests_count > limits[0]:
             logger.warning(
                 "The number of concurrent tests is higher than the open file descriptors limit for this ANTA process.\n"
                 "Errors may occur while running the tests.\n"
                 "Please consult the ANTA FAQ."
             )
 
-        coroutines = get_coroutines(selected_tests)
+        coroutines = get_coroutines(selected_tests, manager)
 
     if dry_run:
         logger.info("Dry-run mode, exiting before running the tests.")
@@ -286,8 +291,6 @@ async def main(  # noqa: PLR0913
         AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=len(coroutines))
 
     with Catchtime(logger=logger, message="Running ANTA tests"):
-        test_results = await asyncio.gather(*coroutines)
-        for r in test_results:
-            manager.add(r)
+        await asyncio.gather(*coroutines)
 
     log_cache_statistics(selected_inventory.devices)

tests/benchmark/test_anta.py

@@ -38,19 +38,16 @@ def test_anta_dry_run(benchmark: BenchmarkFixture, catalog: AntaCatalog, invento
     def bench() -> ResultManager:
         """Need to wrap the ANTA Runner to instantiate a new ResultManger for each benchmark run."""
         manager = ResultManager()
+        catalog.clear_indexes()
         asyncio.run(main(manager, inventory, catalog, dry_run=True))
         return manager
 
     manager = benchmark(bench)
 
     logging.disable(logging.NOTSET)
-    if len(manager.results) != 0:
-        pytest.fail("ANTA Dry-Run mode should not return any result", pytrace=False)
-    if catalog.final_tests_count != len(inventory) * len(catalog.tests):
-        pytest.fail(f"Expected {len(inventory) * len(catalog.tests)} selected tests but got {catalog.final_tests_count}", pytrace=False)
-    bench_info = (
-        "\n--- ANTA NRFU Dry-Run Benchmark Information ---\n" f"Selected tests: {catalog.final_tests_count}\n" "-----------------------------------------------"
-    )
+    if len(manager.results) != len(inventory) * len(catalog.tests):
+        pytest.fail(f"Expected {len(inventory) * len(catalog.tests)} tests but got {len(manager.results)}", pytrace=False)
+    bench_info = "\n--- ANTA NRFU Dry-Run Benchmark Information ---\n" f"Test count: {len(manager.results)}\n" "-----------------------------------------------"
     logger.info(bench_info)
 
 
@@ -73,6 +70,7 @@ def test_anta(benchmark: BenchmarkFixture, catalog: AntaCatalog, inventory: Anta
     def bench() -> ResultManager:
         """Need to wrap the ANTA Runner to instantiate a new ResultManger for each benchmark run."""
         manager = ResultManager()
+        catalog.clear_indexes()
         asyncio.run(main(manager, inventory, catalog))
         return manager
 
@@ -94,7 +92,7 @@ def bench() -> ResultManager:
             for test in dupes:
                 msg = f"Found duplicate in test catalog: {test}"
                 logger.error(msg)
-        pytest.fail(f"Expected {len(catalog.tests) * len(inventory)} test results but got {len(manager.results)}", pytrace=False)
+        pytest.fail(f"Expected {len(catalog.tests) * len(inventory)} tests but got {len(manager.results)}", pytrace=False)
     bench_info = (
         "\n--- ANTA NRFU Benchmark Information ---\n"
         f"Test results: {len(manager.results)}\n"

tests/conftest.py

@@ -17,7 +17,7 @@
 DATA_DIR: Path = Path(__file__).parent.resolve() / "data"
 
 
-@pytest.fixture(params=[{"count": 1}])
+@pytest.fixture(params=[{"count": 1}], ids=["1-reachable-device-without-cache"])
 def inventory(request: pytest.FixtureRequest) -> Iterator[AntaInventory]:
     """Generate an ANTA inventory."""
     user = "admin"

tests/units/test_catalog.py

@@ -260,10 +260,10 @@ def test_build_indexes_all(self) -> None:
         """Test AntaCatalog.build_indexes()."""
         catalog: AntaCatalog = AntaCatalog.parse(DATA_DIR / "test_catalog_with_tags.yml")
         catalog.build_indexes()
-        assert len(catalog.tests_without_tags) == 6
+        assert len(catalog._tests_without_tags) == 6
         assert "leaf" in catalog.tag_to_tests
         assert len(catalog.tag_to_tests["leaf"]) == 3
-        all_unique_tests = catalog.tests_without_tags
+        all_unique_tests = catalog._tests_without_tags
         for tests in catalog.tag_to_tests.values():
             all_unique_tests.update(tests)
         assert len(all_unique_tests) == 11
@@ -275,8 +275,8 @@ def test_build_indexes_filtered(self) -> None:
         catalog.build_indexes({"VerifyUptime", "VerifyCoredump", "VerifyL3MTU"})
         assert "leaf" in catalog.tag_to_tests
         assert len(catalog.tag_to_tests["leaf"]) == 1
-        assert len(catalog.tests_without_tags) == 1
-        all_unique_tests = catalog.tests_without_tags
+        assert len(catalog._tests_without_tags) == 1
+        all_unique_tests = catalog._tests_without_tags
         for tests in catalog.tag_to_tests.values():
             all_unique_tests.update(tests)
         assert len(all_unique_tests) == 4