From e8e018a6dea008ceda54126de6ba258609e93f1c Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Mon, 29 Jun 2026 09:56:47 -0400 Subject: [PATCH 1/6] feat(learn): add --learn mode framework for expected state capture Implements the framework for a two-phase operational test workflow: 1. Learn: `nac-test --learn` captures live state as baseline YAML 2. Verify: `nac-test -d ./learned_state` compares against baseline New components: - --learn CLI flag (propagated via NAC_TEST_LEARN env var) - LearningModeMixin: contract for tests that support learning mode (override capture_learned_state() to define what to capture) - learned_state utilities: save/load YAML files compatible with the existing -d merge mechanism (nac_yaml.merge_dict) - _run_learning_mode() in NACTestBase: branches on env var in run_verification_async(), calls capture method, writes output Tests opt in by inheriting LearningModeMixin and implementing capture_learned_state(). Tests without it are skipped gracefully in learn mode. Default (--learn not passed) behavior is unchanged. Co-Authored-By: Claude Opus 4.6 (1M context) --- nac_test/cli/main.py | 12 ++ nac_test/combined_orchestrator.py | 3 + nac_test/pyats_core/common/base_test.py | 84 ++++++++ .../pyats_core/common/learning_mode_mixin.py | 101 ++++++++++ nac_test/pyats_core/orchestrator.py | 19 +- nac_test/utils/learned_state.py | 80 ++++++++ .../test_combined_orchestrator_controller.py | 2 + tests/unit/test_learning_mode.py | 179 ++++++++++++++++++ 8 files changed, 479 insertions(+), 1 deletion(-) create mode 100644 nac_test/pyats_core/common/learning_mode_mixin.py create mode 100644 nac_test/utils/learned_state.py create mode 100644 tests/unit/test_learning_mode.py diff --git a/nac_test/cli/main.py b/nac_test/cli/main.py index 539b4bfa..af5f4565 100644 --- a/nac_test/cli/main.py +++ b/nac_test/cli/main.py @@ -279,6 +279,16 @@ def version_callback(value: bool) -> None: ] +Learn = Annotated[ + bool, + typer.Option( + "--learn", + help="Run in learning mode: capture live operational state as baseline instead of verifying. Output is written to {output}/learned_state/ and can be loaded as a -d path for verification.", + envvar="NAC_TEST_LEARN", + ), +] + + Testbed = Annotated[ Path | None, typer.Option( @@ -304,6 +314,7 @@ def main( exclude: Exclude = None, render_only: RenderOnly = False, dry_run: DryRun = False, + learn: Learn = False, processes: Processes = None, pyats: PyATS = False, robot: Robot = False, @@ -413,6 +424,7 @@ def main( exclude_tags=exclude, render_only=render_only, dry_run=dry_run, + learn=learn, processes=processes, extra_args=validated_robot_args, max_parallel_devices=max_parallel_devices, diff --git a/nac_test/combined_orchestrator.py b/nac_test/combined_orchestrator.py index d39e0572..ecc35cfe 100644 --- a/nac_test/combined_orchestrator.py +++ b/nac_test/combined_orchestrator.py @@ -82,6 +82,7 @@ def __init__( exclude_tags: list[str] | None = None, render_only: bool = False, dry_run: bool = False, + learn: bool = False, max_parallel_devices: int | None = None, minimal_reports: bool = False, custom_testbed_path: Path | None = None, @@ -129,6 +130,7 @@ def __init__( self.exclude_tags = exclude_tags or [] self.render_only = render_only self.dry_run = dry_run + self.learn = learn self.processes = processes self.extra_args = extra_args @@ -220,6 +222,7 @@ def run_tests(self) -> CombinedResults: custom_testbed_path=self.custom_testbed_path, controller_type=self.controller_type, dry_run=self.dry_run, + learn=self.learn, verbose=self.verbose, loglevel=self.loglevel, include_tags=self.include_tags, diff --git a/nac_test/pyats_core/common/base_test.py b/nac_test/pyats_core/common/base_test.py index c5080e44..e81baf3b 100644 --- a/nac_test/pyats_core/common/base_test.py +++ b/nac_test/pyats_core/common/base_test.py @@ -1956,6 +1956,12 @@ async def run_verification_async(self) -> list[VerificationResult]: } ] + # Learning mode: capture state instead of verifying + if getattr(self, "SUPPORTS_LEARNING", False) and os.environ.get( + "NAC_TEST_LEARN" + ): + return await self._run_learning_mode(items_to_verify) + # Detect verification pattern based on return type if isinstance(items_to_verify, dict): # Grouped verification: {group_key: [contexts]} @@ -2068,6 +2074,84 @@ async def _run_grouped_verification( return flattened_results + async def _run_learning_mode( + self, items: list[dict[str, Any]] | dict[str, list[dict[str, Any]]] + ) -> list["VerificationResult"]: + """Execute in learning mode — capture state instead of asserting. + + Calls the test's capture_learned_state() method to query live state, + then writes the captured data to the learned_state output directory. + + Args: + items: Items from get_items_to_verify() (list or dict depending on pattern). + + Returns: + Single-element list with a PASSED result indicating successful capture. + """ + from nac_test.utils.learned_state import save_learned_state + + # Flatten grouped items to a list for the capture method + if isinstance(items, dict): + flat_items = [ctx for group in items.values() for ctx in group] + else: + flat_items = items + + from nac_test.pyats_core.constants import DEFAULT_API_CONCURRENCY + + semaphore = asyncio.Semaphore(DEFAULT_API_CONCURRENCY) + client = getattr(self, "client", None) + + if not hasattr(self, "capture_learned_state"): + msg = ( + f"{self.__class__.__name__} has SUPPORTS_LEARNING=True but no " + f"capture_learned_state() method. Inherit LearningModeMixin." + ) + self.logger.warning(msg) + return [ + { + "status": ResultStatus.SKIPPED, + "context": {"action": "learn"}, + "reason": msg, + "api_duration": 0, + } + ] + + try: + learned_data = await self.capture_learned_state( + semaphore, client, flat_items + ) + except NotImplementedError as e: + self.logger.warning(str(e)) + return [ + { + "status": ResultStatus.SKIPPED, + "context": {"action": "learn"}, + "reason": str(e), + "api_duration": 0, + } + ] + + # Write captured state to output directory + hostname = getattr(self, "hostname", None) + test_name = self.__class__.__name__ + learned_state_dir = Path( + os.environ.get("NAC_TEST_LEARNED_STATE_DIR", "learned_state") + ) + output_path = save_learned_state( + learned_data, learned_state_dir, test_name, hostname + ) + + self.logger.info(f"Learned state saved to {output_path}") + + return [ + { + "status": ResultStatus.PASSED, + "context": {"action": "learn", "output_path": str(output_path)}, + "reason": f"Successfully captured baseline state ({len(flat_items)} items) → {output_path.name}", + "api_duration": 0, + } + ] + async def _run_item_verification( self, items: list[dict[str, Any]] ) -> list[VerificationResult]: diff --git a/nac_test/pyats_core/common/learning_mode_mixin.py b/nac_test/pyats_core/common/learning_mode_mixin.py new file mode 100644 index 00000000..d2540303 --- /dev/null +++ b/nac_test/pyats_core/common/learning_mode_mixin.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: MPL-2.0 +# Copyright (c) 2025 Daniel Schmidt + +"""Learning mode mixin for operational test classes. + +Provides the contract for tests that support a two-phase workflow: +1. Learn mode (--learn): capture live state and save as baseline +2. Verify mode (default): compare live state against captured baseline + +Tests opt in by inheriting LearningModeMixin and implementing +capture_learned_state(). The base class orchestration checks for +SUPPORTS_LEARNING and the NAC_TEST_LEARN env var to route execution. + +Usage: + class VerifyBGPRoutes(LearningModeMixin, IOSXETestBase): + SUPPORTS_LEARNING = True + + async def capture_learned_state(self, semaphore, client, items): + # Query live state and return structured data + return {"sdwan": {"sites": [...]}} + + def get_items_to_verify(self): + # Same as normal — extract what to check + ... + + async def verify_item(self, semaphore, client, context): + # Normal verification against data model (which now includes learned state) + ... +""" + +import os +from pathlib import Path +from typing import Any + + +class LearningModeMixin: + """Mixin adding learning mode support to operational test classes. + + Tests that support learning inherit this mixin and override + capture_learned_state(). The framework detects learn mode via + the NAC_TEST_LEARN environment variable and calls the capture + method instead of the normal verify loop. + + Attributes: + SUPPORTS_LEARNING: Class-level flag indicating this test supports + the --learn mode. Set to True in subclasses that implement + capture_learned_state(). + """ + + SUPPORTS_LEARNING: bool = True + + @property + def is_learn_mode(self) -> bool: + """Check if running in learning mode. + + Returns: + True if NAC_TEST_LEARN environment variable is set and truthy. + """ + return bool(os.environ.get("NAC_TEST_LEARN")) + + @property + def learned_state_dir(self) -> Path: + """Get the output directory for learned state files. + + Returns: + Path from NAC_TEST_LEARNED_STATE_DIR env var, or 'learned_state' + as fallback. + """ + return Path(os.environ.get("NAC_TEST_LEARNED_STATE_DIR", "learned_state")) + + async def capture_learned_state( + self, + semaphore: Any, + client: Any, + items: list[dict[str, Any]], + ) -> dict[str, Any]: + """Capture live state for all items. Override in subclass. + + Called in learning mode instead of the normal verify_item() loop. + The implementation should make the same queries as verify_item() but + return the raw captured state rather than a pass/fail verdict. + + The returned dict should be structured so it merges cleanly into the + data model when loaded via -d (using nac_yaml's merge_dict logic). + + Args: + semaphore: Asyncio semaphore for concurrency control. + client: HTTP client or SSH connection (same as verify_item receives). + items: List of context dicts from get_items_to_verify(). + + Returns: + Dictionary containing captured state, structured for data model merge. + + Raises: + NotImplementedError: If subclass doesn't override this method. + """ + raise NotImplementedError( + f"{self.__class__.__name__} has SUPPORTS_LEARNING=True but does not " + f"implement capture_learned_state(). Override this method to define " + f"what state to capture in learning mode." + ) diff --git a/nac_test/pyats_core/orchestrator.py b/nac_test/pyats_core/orchestrator.py index 6465d5be..b3898b79 100644 --- a/nac_test/pyats_core/orchestrator.py +++ b/nac_test/pyats_core/orchestrator.py @@ -71,6 +71,7 @@ def __init__( custom_testbed_path: Path | None = None, controller_type: str | None = None, dry_run: bool = False, + learn: bool = False, verbose: bool = False, loglevel: LogLevel = DEFAULT_LOGLEVEL, include_tags: list[str] | None = None, @@ -108,6 +109,7 @@ def __init__( self.minimal_reports = minimal_reports self.custom_testbed_path = custom_testbed_path self.dry_run = dry_run + self.learn = learn self.verbose = verbose self.loglevel = loglevel self.include_tags = include_tags @@ -278,6 +280,12 @@ async def _execute_api_tests_standard(self, test_files: list[Path]) -> Path | No env["NAC_TEST_TYPE"] = "api" # Pass test_dir so plugin can compute relative test names env[ENV_TEST_DIR] = str(self.test_dir) + # Learning mode: propagate flag and set output directory + if self.learn: + env["NAC_TEST_LEARN"] = "1" + learned_state_dir = self.base_output_dir / "learned_state" + learned_state_dir.mkdir(parents=True, exist_ok=True) + env["NAC_TEST_LEARNED_STATE_DIR"] = str(learned_state_dir) # Execute and return the archive path assert self.subprocess_runner is not None # Should be initialized by now @@ -365,6 +373,13 @@ async def _execute_ssh_tests_device_centric( # Set environment variable for test subprocesses to find broker os.environ["NAC_TEST_BROKER_SOCKET"] = str(broker.socket_path) + # Learning mode: propagate to D2D subprocesses via os.environ + if self.learn: + os.environ["NAC_TEST_LEARN"] = "1" + learned_state_dir = self.base_output_dir / "learned_state" + learned_state_dir.mkdir(parents=True, exist_ok=True) + os.environ["NAC_TEST_LEARNED_STATE_DIR"] = str(learned_state_dir) + # Execute device tests with broker running return await self._execute_device_tests_with_broker(test_files, devices) @@ -374,8 +389,10 @@ async def _execute_ssh_tests_device_centric( ) return None finally: - # Clean up environment variable + # Clean up environment variables os.environ.pop("NAC_TEST_BROKER_SOCKET", None) + os.environ.pop("NAC_TEST_LEARN", None) + os.environ.pop("NAC_TEST_LEARNED_STATE_DIR", None) async def _execute_device_tests_with_broker( self, test_files: list[Path], devices: list[dict[str, Any]] diff --git a/nac_test/utils/learned_state.py b/nac_test/utils/learned_state.py new file mode 100644 index 00000000..d657de16 --- /dev/null +++ b/nac_test/utils/learned_state.py @@ -0,0 +1,80 @@ +# SPDX-License-Identifier: MPL-2.0 +# Copyright (c) 2025 Daniel Schmidt + +"""Utilities for reading and writing learned state files. + +Learned state files capture live operational state from network devices, +enabling a two-phase test workflow: +1. Learn: Capture live state and write to YAML files +2. Verify: Load captured state via the standard -d merge mechanism + +Files are written in a structure compatible with nac_yaml.merge_dict(), +so they can be passed as an additional -d path during verification. +""" + +import logging +from pathlib import Path +from typing import Any + +from nac_yaml import yaml + +logger = logging.getLogger(__name__) + + +def save_learned_state( + data: dict[str, Any], + output_dir: Path, + test_name: str, + hostname: str | None = None, +) -> Path: + """Write learned state to a YAML file. + + The output file is named by test class and optionally hostname, + allowing per-device learned state for D2D tests. + + Args: + data: The captured state dictionary to persist. Should be structured + to merge cleanly with the data model when loaded via -d. + output_dir: Directory where learned state files are written. + test_name: Test class name (used in filename). + hostname: Optional device hostname for D2D tests (included in filename). + + Returns: + Path to the written file. + """ + output_dir.mkdir(parents=True, exist_ok=True) + + if hostname: + safe_hostname = hostname.replace("/", "_").replace("\\", "_") + filename = f"{test_name}_{safe_hostname}.yaml" + else: + filename = f"{test_name}.yaml" + + output_path = output_dir / filename + + logger.info("Writing learned state to %s", output_path) + yaml.write_yaml_file(data, output_path) + + return output_path + + +def load_learned_state(file_path: Path) -> dict[str, Any]: + """Load learned state from a YAML file. + + Args: + file_path: Path to the YAML file containing learned state. + + Returns: + Dictionary containing the learned state data, + or empty dict if the file doesn't exist or can't be loaded. + """ + if not file_path.exists(): + logger.warning("Learned state file not found: %s", file_path) + return {} + + try: + data = yaml.load_yaml_files([file_path]) + return data if isinstance(data, dict) else {} + except Exception as e: + logger.error("Failed to load learned state from %s: %s", file_path, e) + return {} diff --git a/tests/unit/test_combined_orchestrator_controller.py b/tests/unit/test_combined_orchestrator_controller.py index bd08a5de..d8a2d965 100644 --- a/tests/unit/test_combined_orchestrator_controller.py +++ b/tests/unit/test_combined_orchestrator_controller.py @@ -235,6 +235,7 @@ def test_combined_orchestrator_passes_controller_to_pyats( custom_testbed_path=None, controller_type="SDWAN", dry_run=False, + learn=False, verbose=False, loglevel=DEFAULT_LOGLEVEL, include_tags=[], @@ -400,6 +401,7 @@ def test_combined_orchestrator_production_mode_passes_controller( custom_testbed_path=None, controller_type="CC", dry_run=False, + learn=False, verbose=False, loglevel=DEFAULT_LOGLEVEL, include_tags=[], diff --git a/tests/unit/test_learning_mode.py b/tests/unit/test_learning_mode.py new file mode 100644 index 00000000..2ddcf457 --- /dev/null +++ b/tests/unit/test_learning_mode.py @@ -0,0 +1,179 @@ +# SPDX-License-Identifier: MPL-2.0 +# Copyright (c) 2025 Daniel Schmidt + +"""Unit tests for the learning mode framework. + +Tests cover: +- LearningModeMixin behavior (mode detection, property access) +- Learned state file utilities (save/load) +- CLI flag recognition +- Environment variable propagation +""" + +import os +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +from nac_test.pyats_core.common.learning_mode_mixin import LearningModeMixin +from nac_test.utils.learned_state import load_learned_state, save_learned_state + + +class TestLearningModeMixin: + """Tests for the LearningModeMixin class.""" + + def _make_instance(self) -> LearningModeMixin: + """Create a bare mixin instance for testing.""" + return LearningModeMixin() + + def test_supports_learning_class_attribute(self) -> None: + """SUPPORTS_LEARNING is True by default on the mixin.""" + instance = self._make_instance() + assert instance.SUPPORTS_LEARNING is True + + def test_is_learn_mode_false_by_default(self) -> None: + """is_learn_mode returns False when env var not set.""" + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("NAC_TEST_LEARN", None) + instance = self._make_instance() + assert instance.is_learn_mode is False + + def test_is_learn_mode_true_when_set(self) -> None: + """is_learn_mode returns True when NAC_TEST_LEARN is set.""" + with patch.dict(os.environ, {"NAC_TEST_LEARN": "1"}): + instance = self._make_instance() + assert instance.is_learn_mode is True + + def test_is_learn_mode_false_for_empty_string(self) -> None: + """is_learn_mode returns False for empty string.""" + with patch.dict(os.environ, {"NAC_TEST_LEARN": ""}): + instance = self._make_instance() + assert instance.is_learn_mode is False + + def test_learned_state_dir_default(self) -> None: + """learned_state_dir returns default path when env var not set.""" + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("NAC_TEST_LEARNED_STATE_DIR", None) + instance = self._make_instance() + assert instance.learned_state_dir == Path("learned_state") + + def test_learned_state_dir_from_env(self) -> None: + """learned_state_dir reads from NAC_TEST_LEARNED_STATE_DIR.""" + with patch.dict(os.environ, {"NAC_TEST_LEARNED_STATE_DIR": "/tmp/my_learned"}): + instance = self._make_instance() + assert instance.learned_state_dir == Path("/tmp/my_learned") + + def test_capture_learned_state_raises_not_implemented(self) -> None: + """Default capture_learned_state raises NotImplementedError.""" + import asyncio + + instance = self._make_instance() + with pytest.raises(NotImplementedError, match="SUPPORTS_LEARNING=True"): + loop = asyncio.new_event_loop() + try: + loop.run_until_complete( + instance.capture_learned_state( + asyncio.Semaphore(1), None, [{"item": "test"}] + ) + ) + finally: + loop.close() + + +class TestLearnedStateUtilities: + """Tests for the learned_state save/load utility functions.""" + + def test_save_learned_state_creates_file(self, tmp_path: Path) -> None: + """save_learned_state writes a YAML file to the output directory.""" + data: dict[str, Any] = {"sdwan": {"sites": [{"id": 100}]}} + result = save_learned_state(data, tmp_path, "VerifyBGPPeers") + + assert result.exists() + assert result.name == "VerifyBGPPeers.yaml" + assert result.parent == tmp_path + + def test_save_learned_state_with_hostname(self, tmp_path: Path) -> None: + """save_learned_state includes hostname in filename for D2D tests.""" + data: dict[str, Any] = {"state": "captured"} + result = save_learned_state( + data, tmp_path, "VerifyBGPPeers", hostname="router-01" + ) + + assert result.name == "VerifyBGPPeers_router-01.yaml" + + def test_save_learned_state_creates_directory(self, tmp_path: Path) -> None: + """save_learned_state creates output directory if it doesn't exist.""" + nested_dir = tmp_path / "deep" / "nested" / "dir" + data: dict[str, Any] = {"test": True} + result = save_learned_state(data, nested_dir, "TestCapture") + + assert result.exists() + assert nested_dir.exists() + + def test_save_learned_state_sanitizes_hostname(self, tmp_path: Path) -> None: + """save_learned_state handles special characters in hostname.""" + data: dict[str, Any] = {"test": True} + result = save_learned_state( + data, tmp_path, "Test", hostname="router/with/slashes" + ) + + assert "/" not in result.name + assert result.name == "Test_router_with_slashes.yaml" + + def test_load_learned_state_returns_empty_for_missing_file(self) -> None: + """load_learned_state returns empty dict for non-existent file.""" + result = load_learned_state(Path("/nonexistent/path.yaml")) + assert result == {} + + def test_save_and_load_roundtrip(self, tmp_path: Path) -> None: + """Data survives a save → load roundtrip.""" + data: dict[str, Any] = { + "sdwan": { + "sites": [ + { + "id": 100, + "routers": [ + { + "device_variables": { + "system_ip": "10.0.0.1", + "learned_state": { + "bgp_neighbors": [ + { + "peer_addr": "10.1.1.1", + "state": "established", + } + ] + }, + } + } + ], + } + ] + } + } + output_path = save_learned_state(data, tmp_path, "TestRoundtrip") + loaded = load_learned_state(output_path) + + assert loaded["sdwan"]["sites"][0]["id"] == 100 + neighbors = loaded["sdwan"]["sites"][0]["routers"][0]["device_variables"][ + "learned_state" + ]["bgp_neighbors"] + assert neighbors[0]["peer_addr"] == "10.1.1.1" + assert neighbors[0]["state"] == "established" + + +class TestCLILearnFlag: + """Tests for the --learn CLI flag integration.""" + + def test_learn_flag_accepted(self) -> None: + """CLI accepts --learn flag without error.""" + from typer.testing import CliRunner + + from nac_test.cli.main import app + + runner = CliRunner() + result = runner.invoke(app, ["--help"]) + assert result.exit_code == 0 + assert "--learn" in result.output From 6638678c90a96aab4b5cbe56011b0086d3c77974 Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Mon, 29 Jun 2026 10:42:19 -0400 Subject: [PATCH 2/6] fix(learn): inherit aetest.Testcase in LearningModeMixin for TestableMeta compat PyATS's TestableMeta metaclass expects all methods on test classes to have a .source attribute set during class __init__. Without inheriting from aetest.Testcase, the mixin's methods lack this attribute, causing AttributeError at class definition time. Co-Authored-By: Claude Opus 4.6 (1M context) --- nac_test/pyats_core/common/learning_mode_mixin.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/nac_test/pyats_core/common/learning_mode_mixin.py b/nac_test/pyats_core/common/learning_mode_mixin.py index d2540303..2ab9e207 100644 --- a/nac_test/pyats_core/common/learning_mode_mixin.py +++ b/nac_test/pyats_core/common/learning_mode_mixin.py @@ -32,15 +32,27 @@ async def verify_item(self, semaphore, client, context): from pathlib import Path from typing import Any +from pyats import aetest -class LearningModeMixin: + +class LearningModeMixin(aetest.Testcase): # type: ignore[misc] """Mixin adding learning mode support to operational test classes. + Inherits from aetest.Testcase so that PyATS's TestableMeta metaclass + processes this class correctly (methods get the required .source attribute). + Python's MRO ensures aetest.Testcase appears only once when combined with + other base classes that also inherit from it. + Tests that support learning inherit this mixin and override capture_learned_state(). The framework detects learn mode via the NAC_TEST_LEARN environment variable and calls the capture method instead of the normal verify loop. + Usage: + class MyTest(LearningModeMixin, SDWANManagerTestBase): + async def capture_learned_state(self, semaphore, client, items): + ... + Attributes: SUPPORTS_LEARNING: Class-level flag indicating this test supports the --learn mode. Set to True in subclasses that implement From c749a6c781f51a416ece62d2cdd5bfb4290bc640 Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Mon, 29 Jun 2026 12:21:36 -0400 Subject: [PATCH 3/6] feat(learn): write marker file with reason when capture returns empty Instead of writing `--- {}` (confusing), empty captures now produce a file with `_learned_state_empty: "reason"`. This distinguishes "learning ran but found nothing" from "learning was never run for this device." Co-Authored-By: Claude Opus 4.6 (1M context) --- nac_test/pyats_core/common/base_test.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/nac_test/pyats_core/common/base_test.py b/nac_test/pyats_core/common/base_test.py index e81baf3b..44f444f7 100644 --- a/nac_test/pyats_core/common/base_test.py +++ b/nac_test/pyats_core/common/base_test.py @@ -2137,6 +2137,16 @@ async def _run_learning_mode( learned_state_dir = Path( os.environ.get("NAC_TEST_LEARNED_STATE_DIR", "learned_state") ) + + # If capture returned empty data, write a marker file with the reason + # so users know learning ran but found nothing (vs. never ran at all) + if not learned_data: + config = getattr(self, "TEST_CONFIG", {}) + command = config.get("api_endpoint", "unknown") + learned_data = { + "_learned_state_empty": f"No data returned by parser for: {command}" + } + output_path = save_learned_state( learned_data, learned_state_dir, test_name, hostname ) From 0635efa969ea303396ecb9a81040d0087229a257 Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Mon, 29 Jun 2026 14:29:57 -0400 Subject: [PATCH 4/6] fix(learn): write learned_state/ to repo root instead of output/ Learned state files are meant to be version-controlled, not ephemeral. Changed default output from {output_dir}/learned_state/ to {cwd}/learned_state/ (repo root). Users pass -d learned_state/ during verify mode to merge the captured baseline into the data model. Co-Authored-By: Claude Opus 4.6 (1M context) --- nac_test/pyats_core/orchestrator.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nac_test/pyats_core/orchestrator.py b/nac_test/pyats_core/orchestrator.py index b3898b79..511e9e1b 100644 --- a/nac_test/pyats_core/orchestrator.py +++ b/nac_test/pyats_core/orchestrator.py @@ -283,7 +283,7 @@ async def _execute_api_tests_standard(self, test_files: list[Path]) -> Path | No # Learning mode: propagate flag and set output directory if self.learn: env["NAC_TEST_LEARN"] = "1" - learned_state_dir = self.base_output_dir / "learned_state" + learned_state_dir = Path.cwd() / "learned_state" learned_state_dir.mkdir(parents=True, exist_ok=True) env["NAC_TEST_LEARNED_STATE_DIR"] = str(learned_state_dir) @@ -376,7 +376,7 @@ async def _execute_ssh_tests_device_centric( # Learning mode: propagate to D2D subprocesses via os.environ if self.learn: os.environ["NAC_TEST_LEARN"] = "1" - learned_state_dir = self.base_output_dir / "learned_state" + learned_state_dir = Path.cwd() / "learned_state" learned_state_dir.mkdir(parents=True, exist_ok=True) os.environ["NAC_TEST_LEARNED_STATE_DIR"] = str(learned_state_dir) From dd57d9fac644e1c993a9694113114faee430abea Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Mon, 29 Jun 2026 20:52:43 -0400 Subject: [PATCH 5/6] feat(learn): add LEARNED_STATE_KEY class variable to LearningModeMixin Provides a consistent, testable namespace identifier for each learning test's output within the merged data model. Convention: "{type}.{Class}" (e.g., "api.VerifyOMPRoutes", "d2d.VerifyBGPLearnedRoutes"). Co-Authored-By: Claude Opus 4.6 (1M context) --- nac_test/pyats_core/common/base_test.py | 3 +-- nac_test/pyats_core/common/learning_mode_mixin.py | 6 ++++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/nac_test/pyats_core/common/base_test.py b/nac_test/pyats_core/common/base_test.py index 44f444f7..8b99a733 100644 --- a/nac_test/pyats_core/common/base_test.py +++ b/nac_test/pyats_core/common/base_test.py @@ -50,6 +50,7 @@ get_defaults_prefix, ) from nac_test.utils.formatting import format_file_timestamp_ms +from nac_test.utils.learned_state import save_learned_state from nac_test.utils.yaml import safe_load T = TypeVar("T") @@ -2088,8 +2089,6 @@ async def _run_learning_mode( Returns: Single-element list with a PASSED result indicating successful capture. """ - from nac_test.utils.learned_state import save_learned_state - # Flatten grouped items to a list for the capture method if isinstance(items, dict): flat_items = [ctx for group in items.values() for ctx in group] diff --git a/nac_test/pyats_core/common/learning_mode_mixin.py b/nac_test/pyats_core/common/learning_mode_mixin.py index 2ab9e207..286cf472 100644 --- a/nac_test/pyats_core/common/learning_mode_mixin.py +++ b/nac_test/pyats_core/common/learning_mode_mixin.py @@ -57,9 +57,15 @@ async def capture_learned_state(self, semaphore, client, items): SUPPORTS_LEARNING: Class-level flag indicating this test supports the --learn mode. Set to True in subclasses that implement capture_learned_state(). + LEARNED_STATE_KEY: Unique string identifying this test's namespace + within the learned_state data structure. Must be unique across + all tests that support learning. Used as the key under + {architecture}.learned_state.{LEARNED_STATE_KEY} in the merged + data model. Convention: use the class name. """ SUPPORTS_LEARNING: bool = True + LEARNED_STATE_KEY: str = "" @property def is_learn_mode(self) -> bool: From b4f158ed4fc6ff9a50d1b318b8c66e3d68512cde Mon Sep 17 00:00:00 2001 From: Christopher Hart Date: Sat, 15 Aug 2026 19:35:27 -0400 Subject: [PATCH 6/6] fix(test): strip ANSI codes in CLI help assertion Rich/Typer adds ANSI escape codes to --help output on Linux but not Windows, causing the plain string assertion to fail in CI. Strip escape codes before checking for --learn flag presence. Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/unit/test_learning_mode.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_learning_mode.py b/tests/unit/test_learning_mode.py index 2ddcf457..fdd6f344 100644 --- a/tests/unit/test_learning_mode.py +++ b/tests/unit/test_learning_mode.py @@ -169,6 +169,8 @@ class TestCLILearnFlag: def test_learn_flag_accepted(self) -> None: """CLI accepts --learn flag without error.""" + import re + from typer.testing import CliRunner from nac_test.cli.main import app @@ -176,4 +178,6 @@ def test_learn_flag_accepted(self) -> None: runner = CliRunner() result = runner.invoke(app, ["--help"]) assert result.exit_code == 0 - assert "--learn" in result.output + # Strip ANSI escape codes before checking (Rich adds them on Linux) + plain_output = re.sub(r"\x1b\[[0-9;]*m", "", result.output) + assert "--learn" in plain_output