Ansible ND 4.X | ND Manage Switches Module + Pydantic Models + Smart Endpoints

[AKDRG]

This PR introduces the initial implementation of ND Manage switch lifecycle support in the cisco.nd collection.
It adds a new switch management resource layer, endpoint wrappers, Pydantic model hierarchy, utility helpers.

What’s Included

1. New switch lifecycle resource implementation

Added a full resource engine in plugins/module_utils/manage_switches/nd_switch_resources.py
Implements state handling for:

• gathered
• merged
• replaced
• overridden
• deleted
• POAP flows (bootstrap, pre-provision, serial swap)
• RMA flow

Adds structured operation sequencing:

• discovery
• add/remove
• role updates
• credentials save
• wait-for-manageability
• save/deploy finalize

2. ND Manage endpoint layer

Added endpoint models under plugins/module_utils/endpoints/v1/:

• manage_credentials_switches.py
• manage_fabrics_actions.py
• manage_fabrics_bootstrap.py
• manage_fabrics_inventory.py
• manage_fabrics_switchactions.py
• manage_fabrics_switches.py
• manage_fabrics.py

4. Pydantic model framework and nd_manage_switches models

• Added nd_manage_switches model package and exports:
• bootstrap, config, discovery, preprovision, rma, switch_actions, switch_data, enums, validators
• package-level init re-exports for consistent imports

5. Utilities for lifecycle orchestration
   Added helpers in plugins/module_utils/manage_switches:

• utils.py

6. Unit Tests for the Endpoints and Integration Tests for the States supported by the Module.

These utilities centralize grouping, payload shaping, wait/retry logic, and fabric operations.

[mikewiebe]

We need to support separate save and deploy options

Use the following property structures

config_actions:
  save: true
  deploy: true
  type: switch [Enum Choices: switch and global]

If config_actions.save is false and config_actions.deploy is true we should fail

[allenrobel]

Added several comments. Mainly around removing all legacy type annotations and removing unneeded __metaclass = type.

Added comment regarding divorcing as much code as possible from Ansible dependencies. So, split the action plugin file into two files; one containing Ansible dependencies and the other free of Ansible dependencies.

Also, possible bug with conditional in SwitchesValidate elif conditional.

[allenrobel]

Remove. No longer needed.

Comment applies for all files.

[allenrobel]

Remove. No longer needed.

Comment applies for all files.

[allenrobel]

Remove imports for Dict, List, Optional, Union and use modern type annotations.

Easiest way to do this if you're using Claude Code is to ask Claude to update his memory, or Claude.md file, to always use modern type annotations for all new code. You can also ask him to migrate all existing legacy type annotations to modern annotations.

Comment applies for all files in this PR.

[allenrobel]

Doesn't the or ignore_fields["role"] at the end of this conditional mean that in role-only mode, this branch fires for every unmatched nd_item in the inner loop, so would record spurious role mismatches against arbitrary pairings? Shouldn't role-only mode not be producing role mismatch diagnostics at all? Or, if it should, maybe the logic needs rethinking?

[allenrobel]

nd_data_copy is never mutated, right? So the copy() is not needed and nd_data_list could be iterated directly.

[allenrobel]

Can you separate this out into a separate file that has no Ansible dependencies? Below is a pure python implementation that uses modern type annotations and fixes some of the issues raised by my comments.

Diagnostics (previously surfaced with Display) are now collected in self.missing_ips and self.role_mismatches. The caller decides how to surface them.

This is also now trivially testable since we don't have to mock ActionBash and capture Display output to test the same logic e.g.:

def test_missing_seed_ip():
    v = SwitchesValidate(
        config_data=[{"seed_ip": "10.0.0.1", "role": "leaf"}],
        nd_data=[{"fabricManagementIp": "10.0.0.2", "switchRole": "leaf"}],
    )
    assert v.response is False
    assert v.missing_ips == ["10.0.0.1"]
    assert v.role_mismatches == {}

New SwitchesValidate:

# -*- coding: utf-8 -*-

# Copyright: (c) 2026, Cisco Systems
# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)

"""Pure-Python validator for ND switch inventory data.

Validates that every entry in a user-supplied config list has a matching
switch in the ND API response. No Ansible dependencies — reusable from
any Python context.

Matching modes are expressed via ``ignore_fields``:
  - ``{"seed_ip": 0, "role": 0}`` (default): match by seed_ip AND role.
  - ``{"seed_ip": 0, "role": 1}``: match by seed_ip only.
  - ``{"seed_ip": 1, "role": 0}``: match by role only.
"""

from __future__ import annotations

from typing import Any

from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import (
    BaseModel,
    Field,
    ValidationError,
    field_validator,
    model_validator,
)
from ansible_collections.cisco.nd.plugins.module_utils.models.manage_switches.config_models import (
    SwitchConfigModel,
)
from ansible_collections.cisco.nd.plugins.module_utils.models.manage_switches.switch_data_models import (
    SwitchDataModel,
)


class SwitchesValidate(BaseModel):
    """Matches playbook config entries against live ND inventory.

    After construction, inspect:
      - ``response``: True if all config entries matched, False otherwise.
      - ``missing_ips``: seed_ips from config with no matching ND switch.
      - ``role_mismatches``: {seed_ip: {"expected_role": ..., "response_role": ...}}
    """

    config_data: list[SwitchConfigModel] | None = None
    nd_data: list[SwitchDataModel] | None = None
    ignore_fields: dict[str, int] = Field(default_factory=lambda: {"seed_ip": 0, "role": 0})
    response: bool | None = None
    missing_ips: list[str] = Field(default_factory=list)
    role_mismatches: dict[str, dict[str, Any]] = Field(default_factory=dict)

    @field_validator("config_data", mode="before")
    @classmethod
    def parse_config_data(cls, value: Any) -> list[SwitchConfigModel] | None:
        """Coerce raw dicts into SwitchConfigModel instances."""
        if value is None:
            return None
        if isinstance(value, dict):
            return [SwitchConfigModel.model_validate(value)]
        if isinstance(value, list):
            try:
                return [
                    SwitchConfigModel.model_validate(item) if isinstance(item, dict) else item
                    for item in value
                ]
            except (ValidationError, ValueError) as e:
                raise ValueError(f"Invalid format in Config Data: {e}")
        raise ValueError("Config Data must be a dict, list of dicts, or None.")

    @field_validator("nd_data", mode="before")
    @classmethod
    def parse_nd_data(cls, value: Any) -> list[SwitchDataModel] | None:
        """Coerce raw ND API switch dicts into SwitchDataModel instances."""
        if value is None:
            return None
        if isinstance(value, list):
            try:
                return [
                    SwitchDataModel.from_response(item) if isinstance(item, dict) else item
                    for item in value
                ]
            except (ValidationError, ValueError) as e:
                raise ValueError(f"Invalid format in ND Response: {e}")
        raise ValueError("ND Response must be a list of dictionaries.")

    @field_validator("ignore_fields", mode="before")
    @classmethod
    def default_ignore_fields(cls, value: dict[str, int] | None) -> dict[str, int]:
        """Ensure both keys are always present, defaulting to 0 (strict match)."""
        if value is None:
            return {"seed_ip": 0, "role": 0}
        return {"seed_ip": value.get("seed_ip", 0), "role": value.get("role", 0)}

    @model_validator(mode="after")
    def validate_lists_equality(self) -> SwitchesValidate:
        """Match every config entry against the live ND switch inventory."""
        config_data = self.config_data
        nd_data_list = self.nd_data
        ignore_seed_ip = bool(self.ignore_fields["seed_ip"])
        ignore_role = bool(self.ignore_fields["role"])

        # Both empty → nothing to validate, treat as success.
        # Exactly one empty → mismatch, treat as failure.
        if not config_data and not nd_data_list:
            self.response = True
            return self
        if not config_data or not nd_data_list:
            self.response = False
            return self

        matched_indices: set[int] = set()

        for config_item in config_data:
            seed_ip = config_item.seed_ip
            role_expected = config_item.role  # SwitchRole enum or None
            found_match = False
            role_mismatch_for_this_entry: dict[str, Any] | None = None

            for i, nd_item in enumerate(nd_data_list):
                if i in matched_indices:
                    continue

                ip_address = nd_item.fabric_management_ip
                switch_role = nd_item.switch_role  # SwitchRole enum or None

                seed_ip_match = ignore_seed_ip or (
                    seed_ip is not None
                    and ip_address is not None
                    and ip_address == seed_ip
                )
                role_match = ignore_role or (
                    role_expected is not None
                    and switch_role is not None
                    and switch_role == role_expected
                )

                if seed_ip_match and role_match:
                    matched_indices.add(i)
                    found_match = True
                    break

                # Full match failed. If seed_ip matched but role didn't,
                # remember it as a candidate role mismatch. We only commit
                # this diagnostic if no better match turns up later.
                if (
                    not ignore_role
                    and seed_ip_match
                    and not ignore_seed_ip
                    and role_expected is not None
                    and switch_role is not None
                    and switch_role != role_expected
                ):
                    role_mismatch_for_this_entry = {
                        "nd_index": i,
                        "ip_address": ip_address,
                        "expected_role": role_expected.value,
                        "response_role": switch_role.value,
                    }

            if not found_match:
                if role_mismatch_for_this_entry is not None:
                    # Seed IP was found but role disagreed — record and
                    # consume that nd_item so it's not re-matched.
                    rm = role_mismatch_for_this_entry
                    matched_indices.add(rm["nd_index"])
                    self.role_mismatches[seed_ip or rm["ip_address"]] = {
                        "expected_role": rm["expected_role"],
                        "response_role": rm["response_role"],
                    }
                elif seed_ip is not None:
                    self.missing_ips.append(seed_ip)

        self.response = not self.missing_ips and not self.role_mismatches
        return self

[allenrobel]

Below is a rewritten ActionModule using modern type annotations, that should go into a separate file that contains only code that requires the Ansible dependencies.

# -*- coding: utf-8 -*-

# Copyright: (c) 2026, Cisco Systems
# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)

"""ND Switches Validation Action Plugin (thin Ansible adapter)."""

from __future__ import annotations

import json
from typing import Any

from ansible.plugins.action import ActionBase
from ansible.utils.display import Display

from ansible_collections.cisco.nd.plugins.module_utils.common.pydantic_compat import (
    HAS_PYDANTIC,
)

try:
    from ansible_collections.cisco.nd.plugins.module_utils.validators.switches_validator import (
        SwitchesValidate,
    )

    HAS_VALIDATOR = True
except ImportError:
    HAS_VALIDATOR = False

display = Display()


class ActionModule(ActionBase):
    """Ansible action plugin for validating ND switch inventory data.

    Task arguments:
        nd_data   (dict): The registered result of a cisco.nd.nd_rest GET call.
        test_data (list|dict): Expected switch entries, each with ``seed_ip``
                               and optionally ``role``.
        changed   (bool, optional): If provided and False, the task fails
                                    immediately (used to assert an upstream
                                    operation produced a change).
        mode      (str, optional): ``"both"`` (default), ``"ip"``, or ``"role"``.
    """

    def run(
        self,
        tmp: Any = None,
        task_vars: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        results = super().run(tmp, task_vars)
        results["failed"] = False

        if not HAS_PYDANTIC or not HAS_VALIDATOR:
            results["failed"] = True
            results["msg"] = (
                "pydantic and the ND collection validators are required "
                "for nd_switches_validate"
            )
            return results

        args = self._task.args
        nd_data = args["nd_data"]
        test_data = args["test_data"]

        # Fail fast if the caller signals that no change occurred when one was expected.
        if "changed" in args and not args["changed"]:
            results["failed"] = True
            results["msg"] = 'Changed is "false"'
            return results

        # Fail fast if the upstream nd_rest task itself failed.
        if nd_data.get("failed"):
            results["failed"] = True
            results["msg"] = nd_data.get("msg", "ND module returned a failure")
            return results

        switches = nd_data.get("current", {}).get("switches", [])

        if isinstance(test_data, dict):
            test_data = [test_data]

        if not switches and not test_data:
            results["msg"] = "Validation Successful!"
            return results

        if not switches:
            results["failed"] = True
            results["msg"] = "No switches found in ND response"
            return results

        # Resolve matching mode.
        ignore_fields: dict[str, int] = {"seed_ip": 0, "role": 0}
        mode = args.get("mode", "both").lower()
        if mode == "ip":
            ignore_fields["role"] = 1
        elif mode == "role":
            ignore_fields["seed_ip"] = 1

        validation = SwitchesValidate(
            config_data=test_data,
            nd_data=switches,
            ignore_fields=ignore_fields,
        )

        if validation.response:
            results["msg"] = "Validation Successful!"
            return results

        # Surface diagnostics via Ansible's Display so they show up with -v.
        display.display("Invalid Data:")
        if validation.missing_ips:
            display.display(f"  Missing IPs: {validation.missing_ips}")
        if validation.role_mismatches:
            display.display(
                f"  Role mismatches: {json.dumps(validation.role_mismatches, indent=2)}"
            )

        results["failed"] = True
        results["msg"] = "Validation Failed! Please check output above."
        results["missing_ips"] = validation.missing_ips
        results["role_mismatches"] = validation.role_mismatches
        return results