Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 4 additions & 7 deletions dev-docs/PRD_AND_ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -5960,14 +5960,14 @@ Example: `IOSXE_USERNAME` and `IOSXE_PASSWORD` could be used for:
│ IOSXE_PASSWORD=device-pass │
│ │
│ Detection Result: │
│ detect_controller_type() → "sdwan" │
│ resolve_controller() → ControllerContext(controller_type="SDWAN")│
│ → TestTypeResolver uses BASE_CLASS_MAPPING["SDWANTestBase"] │
│ → Device SSH uses IOSXE_USERNAME/IOSXE_PASSWORD │
└────────────────────────────────────────────────────────────────────┘
```

**What happens during D2D test execution:**
1. Framework detects `CONTROLLER_TYPE=SDWAN` from controller credentials
1. Framework detects `controller_context.controller_type == "SDWAN"` from controller credentials
2. Framework loads `SDWANDeviceResolver` for device inventory resolution
3. Tests connect to devices via SSH using `IOSXE_*` credentials
4. Controller credentials are NOT used for connection (D2D tests bypass controller)
Expand All @@ -5978,11 +5978,8 @@ The detected controller type informs test categorization:

```python
# In orchestrator.py
controller_type = detect_controller_type()
if controller_type:
logger.info(f"Detected controller type: {controller_type}")
else:
logger.warning("Could not auto-detect controller type from environment")
controller_context = resolve_controller()
logger.info(f"Detected controller type: {controller_context.controller_type}")
```

#### Usage Examples
Expand Down
142 changes: 12 additions & 130 deletions nac_test/core/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

import logging
import os
import warnings
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
Expand Down Expand Up @@ -272,10 +271,6 @@ class ControllerConfig:
),
}

# Module-level cache for the credential set that was matched during detection.
# Populated by detect_controller_type(), consumed by get_matched_credential_set().
_matched_credential_sets: dict[str, CredentialSet] = {}


class ResolutionError(NacTestError):
"""Base for controller resolution failures."""
Expand Down Expand Up @@ -313,10 +308,6 @@ def resolve_controller() -> ControllerContext:
:class:`ResolutionError` subclass on failure. The caller decides
how to handle failures — this function never calls ``sys.exit()``.

Side-effects:
* Populates ``_matched_credential_sets`` (same as the legacy
``detect_controller_type()``).

Returns:
ControllerContext with ``controller_type`` and ``auth_method``.

Expand All @@ -340,7 +331,6 @@ def resolve_controller() -> ControllerContext:
# Exactly one complete set — success
controller_type = next(iter(complete))
matched_cred_set = complete[controller_type]
_matched_credential_sets[controller_type] = matched_cred_set

ctx = ControllerContext(
controller_type=controller_type,
Expand All @@ -364,42 +354,30 @@ def get_controller_context() -> ControllerContext:
``PyATSOrchestrator``, which serializes it to ``NAC_TEST_CONTROLLER_CONTEXT``
before launching subprocesses.

**Primary path (subprocess):** Deserializes from ``NAC_TEST_CONTROLLER_CONTEXT``
environment variable set by ``PyATSOrchestrator``.

**Fallback (transitional):** If the env var is absent, falls back to
``detect_controller_type()`` for backwards compatibility. This fallback
will be removed in Phase 3 once all consumers are migrated.
Deserializes from ``NAC_TEST_CONTROLLER_CONTEXT`` environment variable
set by ``PyATSOrchestrator``.

Returns:
ControllerContext with controller_type and auth_method.

Raises:
ValueError: If no controller credentials are found (via fallback path).
ValueError: If ``NAC_TEST_CONTROLLER_CONTEXT`` is not set or invalid.
"""
raw = os.environ.get(ENV_CONTROLLER_CONTEXT)
if raw:
return ControllerContext.from_json(raw)

# --- Transitional fallback (remove in Phase 3) -----------------------
logging.getLogger(__name__).info(
"NAC_TEST_CONTROLLER_CONTEXT not set — falling back to "
"detect_controller_type(). This fallback will be removed in a "
"future release."
)

controller_type = detect_controller_type()
return ControllerContext(
controller_type=controller_type,
auth_method=_infer_auth_method(controller_type),
)
if not raw:
raise ValueError(
f"Environment variable {ENV_CONTROLLER_CONTEXT} is not set. "
"Controller context must be resolved by the orchestrator via "
"resolve_controller() and passed to subprocesses."
)
return ControllerContext.from_json(raw)


def format_resolution_error(error: ResolutionError) -> str:
"""Format a :class:`ResolutionError` into a user-facing message.

Re-uses the existing detailed error formatters so that CLI output
stays identical to the legacy ``detect_controller_type()`` path.
Uses the detailed error formatters for multiple, incomplete, or missing
credentials.
"""
if isinstance(error, MultipleControllersFound):
return _format_multiple_credentials_error(error.controllers)
Expand Down Expand Up @@ -679,85 +657,6 @@ def _set_var_count(cs: CredentialSet) -> int:
return values


def detect_controller_type() -> ControllerTypeKey:
"""Detect the controller type based on environment variables.

.. deprecated::
This function is retained for backwards compatibility with external
packages (e.g., ``nac-test-pyats-common``) that have not yet migrated
to :func:`resolve_controller`. New code should use ``resolve_controller()``
directly and handle :class:`ResolutionError` subtypes. This function
will be removed once all consumers have migrated.

This function examines environment variables to determine which network controller
architecture is being targeted. It ensures exactly one controller type has credentials
configured to prevent ambiguous test contexts.

Controller credentials are required for ALL test types:
- API tests: Use credentials directly for controller authentication
- D2D tests: Use controller type to determine device resolution logic

Returns:
The detected controller type (e.g., "ACI", "SDWAN", "CC", "MERAKI", "FMC", "ISE").

Raises:
ValueError: If no controller credentials are found, multiple controllers are
configured, or credentials are incomplete.

Example:
>>> os.environ.update({"ACI_URL": "https://apic.local",
... "ACI_USERNAME": "admin",
... "ACI_PASSWORD": "pass"})
>>> controller = detect_controller_type()
>>> print(controller)
"ACI"

Note:
This function delegates to :func:`resolve_controller` and converts typed
exceptions to ``ValueError`` for backwards compatibility with existing callers.
"""
warnings.warn(
"detect_controller_type() is deprecated; use resolve_controller() instead.",
DeprecationWarning,
stacklevel=2,
)
try:
ctx = resolve_controller()
return ctx.controller_type
except ResolutionError as e:
raise ValueError(format_resolution_error(e)) from e


def get_matched_credential_set(controller_type: str) -> CredentialSet | None:
"""Get the credential set that was matched during controller detection.

.. deprecated::
This function is a transitional API for ``nac-test-pyats-common`` auth
adapters. It will be removed in Phase 3 once auth adapters migrate to
using ``get_controller_context().auth_method`` directly. New code should
not use this function.

Returns the CredentialSet that satisfied detection for the given controller
type. This is populated by detect_controller_type() / resolve_controller()
and is intended for use by auth adapters in nac-test-pyats-common to
determine which authentication mechanism to use.

Args:
controller_type: The controller type key (e.g., "SDWAN", "ACI").

Returns:
The matched CredentialSet, or None if detection has not been called
or the controller type was not detected.
"""
warnings.warn(
"get_matched_credential_set() is deprecated; use "
"get_controller_context().auth_method instead.",
DeprecationWarning,
stacklevel=2,
)
return _matched_credential_sets.get(controller_type)


def _find_credential_sets() -> tuple[
dict[ControllerTypeKey, CredentialSet],
list[ControllerTypeKey],
Expand Down Expand Up @@ -804,23 +703,6 @@ def _find_credential_sets() -> tuple[
return complete, partial


def _infer_auth_method(controller_type: str) -> AuthMethod:
"""Infer auth_method by scanning env vars for a controller type.

Used only in the transitional fallback path of
``get_controller_context()`` when ``NAC_TEST_CONTROLLER_CONTEXT``
is absent. Mirrors the logic of ``_find_credential_sets()`` but
returns only the auth_method string.
"""
config = CONTROLLER_REGISTRY.get(controller_type)
if config is None:
return AuthMethod.SESSION
for cred_set in config.credential_sets:
if all(is_env_var_set(v) for v in cred_set.env_vars):
return cred_set.auth_method
return AuthMethod.SESSION


def _format_incomplete_credentials_error(partial_controllers: Sequence[str]) -> str:
"""Format error message for incomplete controller credentials.

Expand Down
6 changes: 6 additions & 0 deletions nac_test/core/controller_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
"""

import logging
import os
import sys
from collections.abc import Callable
from dataclasses import dataclass
Expand All @@ -19,6 +20,7 @@
from nac_test.core.auth_cache import AuthCache
from nac_test.core.controller import (
CONTROLLER_REGISTRY,
ENV_CONTROLLER_CONTEXT,
get_controller_url,
get_display_name,
)
Expand Down Expand Up @@ -142,6 +144,10 @@ def preflight_auth_check(ctx: ControllerContext) -> AuthCheckResult:
detail="Pre-flight check skipped (no auth adapter available)",
)

# Ensure controller context env var is set for auth adapters (e.g., SDWANManagerAuth)
# that inspect get_controller_context() during get_auth().
os.environ[ENV_CONTROLLER_CONTEXT] = ctx.to_json()

# Invalidate any stale cached token so we validate the current credentials.
# Best-effort: a failure here must never block test execution.
config = CONTROLLER_REGISTRY.get(controller_type)
Expand Down
7 changes: 3 additions & 4 deletions nac_test/pyats_core/common/base_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,10 +172,9 @@ def setup(self) -> None:
self.data_model = self.load_data_model()

# Get controller context from environment
# In normal operation, CombinedOrchestrator resolves the controller and
# passes it via NAC_TEST_CONTROLLER_CONTEXT env var. The accessor
# get_controller_context() reads this, with a fallback to env var scan
# for direct pyats invocation or legacy compatibility.
# CombinedOrchestrator resolves the controller and passes it via
# the NAC_TEST_CONTROLLER_CONTEXT env var. The accessor
# get_controller_context() deserializes this context.
try:
ctx = get_controller_context()
except (ValueError, KeyError) as e:
Expand Down
18 changes: 0 additions & 18 deletions nac_test/utils/controller.py

This file was deleted.

25 changes: 19 additions & 6 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import pytest

from nac_test.core.constants import ENV_CONTROLLER_CONTEXT
from nac_test.core.controller import CONTROLLER_REGISTRY
from nac_test.core.controller import CONTROLLER_REGISTRY, resolve_controller
from nac_test.core.types import AuthMethod, ControllerContext
from tests.e2e.mocks.mock_server import MockAPIServer

Expand Down Expand Up @@ -68,11 +68,6 @@ def clean_controller_env(monkeypatch: pytest.MonkeyPatch) -> None:
# Clear serialized controller context from previous tests
monkeypatch.delenv(ENV_CONTROLLER_CONTEXT, raising=False)

# Clear module-level credential cache to prevent cross-test pollution
from nac_test.core import controller

controller._matched_credential_sets.clear()


@pytest.fixture(scope="session", autouse=True)
def bypass_proxy_for_localhost() -> Generator[None, None, None]:
Expand Down Expand Up @@ -193,3 +188,21 @@ def cc_context() -> ControllerContext:
def iosxe_context() -> ControllerContext:
"""Pre-built ControllerContext for IOS-XE with session auth."""
return ControllerContext(controller_type="IOSXE", auth_method=AuthMethod.SESSION)


# =============================================================================
# Context injection test helpers
# =============================================================================


def resolve_and_inject_context(monkeypatch: pytest.MonkeyPatch) -> ControllerContext:
"""Resolve controller from current environment and inject into ENV_CONTROLLER_CONTEXT.
Designed for happy-path tests to avoid DRY repetition."""
ctx = resolve_controller()
monkeypatch.setenv(ENV_CONTROLLER_CONTEXT, ctx.to_json())
return ctx


def inject_context(monkeypatch: pytest.MonkeyPatch, ctx: ControllerContext) -> None:
"""Inject a pre-built ControllerContext into ENV_CONTROLLER_CONTEXT."""
monkeypatch.setenv(ENV_CONTROLLER_CONTEXT, ctx.to_json())
8 changes: 4 additions & 4 deletions tests/integration/test_cli_aci_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,12 +77,12 @@ def test_cli_validation_triggers_when_aci_url_set_and_no_defaults(
def test_cli_validation_passes_when_aci_url_not_set(
self,
minimal_test_env: dict[str, Path],
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""CLI should skip ACI validation when ACI_URL is not set."""
# Ensure ACI_URL is not set
monkeypatch.delenv("ACI_URL", raising=False)
"""CLI should skip ACI validation when ACI_URL is not set.

Relies on the global autouse ``clean_controller_env`` fixture in
``tests/conftest.py`` ensuring ACI_URL is unset.
"""
# We need to mock the DataMerger and orchestrator since we don't have full environment
with (
patch("nac_test.cli.main.DataMerger") as mock_merger,
Expand Down
8 changes: 4 additions & 4 deletions tests/integration/test_controller_detection_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import pytest

from nac_test.core.controller import detect_controller_type
from nac_test.core.controller import resolve_controller
from nac_test.pyats_core.orchestrator import PyATSOrchestrator


Expand Down Expand Up @@ -65,7 +65,7 @@ def test_controller_switch_scenario(self, monkeypatch: pytest.MonkeyPatch) -> No
monkeypatch.setenv("ACI_USERNAME", "admin")
monkeypatch.setenv("ACI_PASSWORD", "password")

assert detect_controller_type() == "ACI"
assert resolve_controller().controller_type == "ACI"

# Clear ACI and switch to FMC
monkeypatch.delenv("ACI_URL")
Expand All @@ -76,7 +76,7 @@ def test_controller_switch_scenario(self, monkeypatch: pytest.MonkeyPatch) -> No
monkeypatch.setenv("FMC_USERNAME", "admin")
monkeypatch.setenv("FMC_PASSWORD", "password")

assert detect_controller_type() == "FMC"
assert resolve_controller().controller_type == "FMC"

# Clear FMC and switch to ISE
monkeypatch.delenv("FMC_URL")
Expand All @@ -87,4 +87,4 @@ def test_controller_switch_scenario(self, monkeypatch: pytest.MonkeyPatch) -> No
monkeypatch.setenv("ISE_USERNAME", "admin")
monkeypatch.setenv("ISE_PASSWORD", "password")

assert detect_controller_type() == "ISE"
assert resolve_controller().controller_type == "ISE"
Loading
Loading