From cb798efdb7da4c8d34e794891291a9d173a7ed2a Mon Sep 17 00:00:00 2001 From: Andrei Huides Date: Fri, 21 Aug 2026 11:40:57 +0300 Subject: [PATCH 1/2] ci(lh-114517): add public release smoke harness Add a credential-free harness that resolves matching PyPI and Ansible Galaxy versions, installs the exact public artifacts, and verifies the CLI schema and discovered Ansible plugins in an isolated controller. Cover version resolution and report generation with unit tests. --- .github/workflows/ci.yml | 2 + cisco_sccfm_scripts/verify_public_release.py | 310 +++++++++++++++++++ tests/test_verify_public_release.py | 89 ++++++ 3 files changed, 401 insertions(+) create mode 100644 cisco_sccfm_scripts/verify_public_release.py create mode 100644 tests/test_verify_public_release.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7d9136e7..51fc8646 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,6 +101,7 @@ jobs: cisco_sccfm_scripts/verify_ansible_collection.py \ cisco_sccfm_scripts/verify_clean_controller.py \ cisco_sccfm_scripts/verify_pypi_release.py \ + cisco_sccfm_scripts/verify_public_release.py \ cisco_sccfm_scripts/verify_python_artifacts.py - name: Test @@ -378,6 +379,7 @@ jobs: cisco_sccfm_scripts/verify_ansible_collection.py \ cisco_sccfm_scripts/verify_clean_controller.py \ cisco_sccfm_scripts/verify_pypi_release.py \ + cisco_sccfm_scripts/verify_public_release.py \ cisco_sccfm_scripts/verify_python_artifacts.py poetry run pytest --color=yes poetry run check-doc-links diff --git a/cisco_sccfm_scripts/verify_public_release.py b/cisco_sccfm_scripts/verify_public_release.py new file mode 100644 index 00000000..fed3ae7d --- /dev/null +++ b/cisco_sccfm_scripts/verify_public_release.py @@ -0,0 +1,310 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Smoke-test matching SCCFM artifacts installed from public registries.""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +import tempfile +from collections.abc import Sequence +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import Request, urlopen + +from cisco_sccfm_scripts.verify_clean_controller import ( + _Controller, + _create_controller, + _discovered_plugins, + _offline_checks, + _run, +) + +_ANSIBLE_CORE = "ansible-core>=2.20,<2.22" +_PYPI_LATEST_URL = "https://pypi.org/pypi/cisco-sccfm-devkit/json" +_PYPI_VERSION_URL = "https://pypi.org/pypi/cisco-sccfm-devkit/{version}/json" +_GALAXY_LATEST_URL = ( + "https://galaxy.ansible.com/api/v3/plugin/ansible/content/published/" + "collections/index/cisco/sccfm/" +) +_GALAXY_VERSION_URL = f"{_GALAXY_LATEST_URL}versions/{{version}}/" +_MAX_RESPONSE_BYTES = 1024 * 1024 +_REQUEST_TIMEOUT_SECONDS = 30.0 +_VERSION_PATTERN = re.compile(r"^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$") + + +class PublicReleaseVerificationError(RuntimeError): + """Raised when public artifacts do not form a working matching release.""" + + +@dataclass(frozen=True) +class PublicReleaseSummary: + """Successful public-release smoke-test result.""" + + version: str + module_count: int + inventory_count: int + lookup_count: int + offline_probe: str + + +def _read_json(response: Any, expected_url: str) -> object: + """Read one bounded JSON response from an official registry endpoint.""" + if response.geturl() != expected_url: + raise PublicReleaseVerificationError("registry redirected to an unexpected endpoint") + try: + raw: bytes = response.read(_MAX_RESPONSE_BYTES + 1) + except OSError as exc: + raise PublicReleaseVerificationError("could not read registry response") from exc + if len(raw) > _MAX_RESPONSE_BYTES: + raise PublicReleaseVerificationError("registry response exceeds the size limit") + try: + return json.loads(raw) + except (UnicodeDecodeError, ValueError) as exc: + raise PublicReleaseVerificationError("registry returned invalid JSON") from exc + + +def _fetch_json(url: str, timeout: float = _REQUEST_TIMEOUT_SECONDS) -> object: + """Fetch JSON from one fixed PyPI or Galaxy endpoint.""" + request = Request( + url, + headers={ + "Accept": "application/json", + "User-Agent": "cisco-sccfm-devkit-public-release-smoke", + }, + ) + try: + with urlopen(request, timeout=timeout) as response: + return _read_json(response, url) + except (HTTPError, URLError, TimeoutError, OSError) as exc: + raise PublicReleaseVerificationError("could not query public registry") from exc + + +def _nested_string(payload: object, *path: str) -> str | None: + """Read a nested string from an untrusted JSON object.""" + value = payload + for key in path: + if not isinstance(value, dict): + return None + value = value.get(key) + return value if isinstance(value, str) else None + + +def resolve_public_version(requested: str = "") -> str: + """Resolve one stable version that is present in both public registries.""" + requested = requested.strip() + if requested and _VERSION_PATTERN.fullmatch(requested) is None: + raise PublicReleaseVerificationError("version must be a stable X.Y.Z value") + + if requested: + encoded = quote(requested, safe="") + pypi = _fetch_json(_PYPI_VERSION_URL.format(version=encoded)) + galaxy = _fetch_json(_GALAXY_VERSION_URL.format(version=encoded)) + pypi_version = _nested_string(pypi, "info", "version") + galaxy_version = _nested_string(galaxy, "version") + else: + pypi = _fetch_json(_PYPI_LATEST_URL) + galaxy = _fetch_json(_GALAXY_LATEST_URL) + pypi_version = _nested_string(pypi, "info", "version") + galaxy_version = _nested_string(galaxy, "highest_version", "version") + + if pypi_version is None or galaxy_version is None: + raise PublicReleaseVerificationError("could not resolve versions from public registries") + if pypi_version != galaxy_version: + raise PublicReleaseVerificationError( + f"public registry versions differ: PyPI={pypi_version}, Galaxy={galaxy_version}" + ) + if requested and pypi_version != requested: + raise PublicReleaseVerificationError( + f"registries returned {pypi_version} instead of requested {requested}" + ) + if _VERSION_PATTERN.fullmatch(pypi_version) is None: + raise PublicReleaseVerificationError("public registries returned a non-stable version") + return pypi_version + + +def _install_public_artifacts(controller: _Controller, version: str) -> None: + """Install an exact matching release from PyPI and Ansible Galaxy.""" + python = controller.binaries / "python" + _run( + controller, + [ + python, + "-I", + "-m", + "pip", + "install", + "--no-cache-dir", + "--index-url", + "https://pypi.org/simple", + _ANSIBLE_CORE, + f"cisco-sccfm-devkit=={version}", + ], + ) + _run(controller, [python, "-I", "-m", "pip", "check"]) + _run( + controller, + [ + controller.binaries / "ansible-galaxy", + "collection", + "install", + f"cisco.sccfm:=={version}", + "--server", + "https://galaxy.ansible.com", + "--collections-path", + controller.collections, + ], + ) + + +def _verify_cli(controller: _Controller, version: str) -> None: + """Verify imports, entry points, help, and the discovered CLI schema.""" + python = controller.binaries / "python" + check = """\ +import importlib, importlib.metadata, importlib.util, sys +assert importlib.metadata.version("cisco-sccfm-devkit") == sys.argv[1] +for name in ("cisco_sccfm_cli", "cisco_sccfm_core", "scc_firewall_manager_sdk"): + importlib.import_module(name) +assert importlib.util.find_spec("cisco_sccfm_scripts") is None +scripts = { + entry.name + for entry in importlib.metadata.distribution("cisco-sccfm-devkit").entry_points + if entry.group == "console_scripts" +} +assert {"sccfm-cli", "sccfm-cli-interactive"}.issubset(scripts) +""" + _run(controller, [python, "-I", "-c", check, version]) + _run(controller, [controller.binaries / "sccfm-cli", "--help"]) + _run(controller, [controller.binaries / "sccfm-cli-interactive", "--help"]) + schema_raw = _run( + controller, + [controller.binaries / "sccfm-cli", "schema", "export", "--format", "json"], + ).stdout + try: + schema: object = json.loads(schema_raw) + except json.JSONDecodeError as exc: + raise PublicReleaseVerificationError("CLI schema is not valid JSON") from exc + commands = schema.get("commands") if isinstance(schema, dict) else None + if ( + not isinstance(schema, dict) + or schema.get("version") != version + or not isinstance(commands, list) + or not commands + ): + raise PublicReleaseVerificationError("CLI schema does not describe the installed release") + + +def _verify_collection_version(controller: _Controller, version: str) -> None: + """Verify that Galaxy installed the exact requested collection version.""" + manifest = controller.collections / "ansible_collections/cisco/sccfm/MANIFEST.json" + try: + payload: object = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise PublicReleaseVerificationError("installed collection manifest is invalid") from exc + installed = _nested_string(payload, "collection_info", "version") + if installed != version: + raise PublicReleaseVerificationError( + f"installed collection version {installed} does not match {version}" + ) + + +def _discover_plugins( + controller: _Controller, +) -> tuple[dict[str, str], dict[str, str], dict[str, str]]: + """Discover every supported cisco.sccfm plugin surface through ansible-doc.""" + ansible_doc = controller.binaries / "ansible-doc" + discovered = [] + for plugin_type in ("module", "inventory", "lookup"): + raw = _run( + controller, + [ansible_doc, "-j", "-l", "-t", plugin_type, "cisco.sccfm"], + ).stdout + discovered.append(_discovered_plugins(raw, plugin_type)) + return discovered[0], discovered[1], discovered[2] + + +def _documented_probe(controller: _Controller, modules: dict[str, str]) -> str: + """Select a documented readonly list module with no required business arguments.""" + ansible_doc = controller.binaries / "ansible-doc" + for name, description in modules.items(): + if not description.casefold().startswith("list "): + continue + raw = _run(controller, [ansible_doc, "-j", name]).stdout + try: + payload: object = json.loads(raw) + except json.JSONDecodeError: + continue + module = payload.get(name) if isinstance(payload, dict) else None + doc = module.get("doc") if isinstance(module, dict) else None + options = doc.get("options") if isinstance(doc, dict) else None + if isinstance(options, dict) and not any( + isinstance(option, dict) and option.get("required") is True + for option in options.values() + ): + return name + raise PublicReleaseVerificationError("no argument-free readonly list module was discovered") + + +def verify_public_release(requested: str = "") -> PublicReleaseSummary: + """Install and smoke-test one matching release from the public registries.""" + version = resolve_public_version(requested) + with tempfile.TemporaryDirectory(prefix="sccfm-public-release-") as temporary: + controller = _create_controller(Path(temporary)) + _install_public_artifacts(controller, version) + _verify_cli(controller, version) + _verify_collection_version(controller, version) + modules, inventory, lookups = _discover_plugins(controller) + probe = _documented_probe(controller, modules) + _offline_checks(controller, probe) + return PublicReleaseSummary( + version=version, + module_count=len(modules), + inventory_count=len(inventory), + lookup_count=len(lookups), + offline_probe=probe, + ) + + +def _parser() -> argparse.ArgumentParser: + """Build the public-release smoke-test parser.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--version", + default="", + help="Exact stable X.Y.Z release; omit to use the latest matching public version.", + ) + parser.add_argument("--report", type=Path, help="Optional JSON summary output path.") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the public-release smoke test.""" + args = _parser().parse_args(argv) + try: + summary = verify_public_release(args.version) + if args.report is not None: + args.report.parent.mkdir(parents=True, exist_ok=True) + args.report.write_text( + json.dumps(asdict(summary), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + except (OSError, RuntimeError, ValueError) as exc: + print(f"Public release verification failed: {exc}", file=sys.stderr) + return 1 + print( + f"Public release verified: version={summary.version} modules={summary.module_count} " + f"inventory={summary.inventory_count} lookups={summary.lookup_count} " + f"probe={summary.offline_probe}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_verify_public_release.py b/tests/test_verify_public_release.py new file mode 100644 index 00000000..82f55603 --- /dev/null +++ b/tests/test_verify_public_release.py @@ -0,0 +1,89 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the public-registry release smoke harness.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +import cisco_sccfm_scripts.verify_public_release as verifier + +_VERSION = "1.2.3" + + +def test_resolve_latest_requires_matching_registry_versions( + monkeypatch: pytest.MonkeyPatch, +) -> None: + responses: dict[str, object] = { + verifier._PYPI_LATEST_URL: {"info": {"version": _VERSION}}, + verifier._GALAXY_LATEST_URL: {"highest_version": {"version": _VERSION}}, + } + monkeypatch.setattr(verifier, "_fetch_json", responses.__getitem__) + + assert verifier.resolve_public_version() == _VERSION + + +def test_resolve_exact_version_uses_versioned_registry_endpoints( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + def fetch(url: str) -> object: + calls.append(url) + if "pypi.org" in url: + return {"info": {"version": _VERSION}} + return {"version": _VERSION} + + monkeypatch.setattr(verifier, "_fetch_json", fetch) + + assert verifier.resolve_public_version(_VERSION) == _VERSION + assert calls == [ + verifier._PYPI_VERSION_URL.format(version=_VERSION), + verifier._GALAXY_VERSION_URL.format(version=_VERSION), + ] + + +def test_resolve_rejects_invalid_version_without_network( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def unexpected_fetch(url: str) -> object: + raise AssertionError(f"unexpected request: {url}") + + monkeypatch.setattr(verifier, "_fetch_json", unexpected_fetch) + + with pytest.raises(verifier.PublicReleaseVerificationError, match="stable X.Y.Z"): + verifier.resolve_public_version("v1.2.3") + + +def test_resolve_rejects_registry_mismatch(monkeypatch: pytest.MonkeyPatch) -> None: + responses: dict[str, object] = { + verifier._PYPI_LATEST_URL: {"info": {"version": _VERSION}}, + verifier._GALAXY_LATEST_URL: {"highest_version": {"version": "1.2.2"}}, + } + monkeypatch.setattr(verifier, "_fetch_json", responses.__getitem__) + + with pytest.raises(verifier.PublicReleaseVerificationError, match="versions differ"): + verifier.resolve_public_version() + + +def test_main_writes_machine_readable_summary( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + report = tmp_path / "reports" / "summary.json" + summary = verifier.PublicReleaseSummary(_VERSION, 49, 1, 1, "cisco.sccfm.list_devices") + monkeypatch.setattr(verifier, "verify_public_release", lambda requested: summary) + + assert verifier.main(["--version", _VERSION, "--report", str(report)]) == 0 + assert json.loads(report.read_text(encoding="utf-8")) == { + "inventory_count": 1, + "lookup_count": 1, + "module_count": 49, + "offline_probe": "cisco.sccfm.list_devices", + "version": _VERSION, + } From 5573c8883f96897c531d76881dd72d5ec37b5bcc Mon Sep 17 00:00:00 2001 From: Andrei Huides Date: Fri, 21 Aug 2026 13:50:25 +0300 Subject: [PATCH 2/2] test(lh-114517): expand public release runtime probes Validate the CLI schema safety contract and execute the discovered Ansible inventory and lookup plugins against an isolated controller. Require each runtime path to stop at missing-profile validation and include the selected plugins in the smoke report. --- cisco_sccfm_scripts/verify_public_release.py | 151 +++++++++++++++++-- tests/test_verify_public_release.py | 149 +++++++++++++++++- 2 files changed, 287 insertions(+), 13 deletions(-) diff --git a/cisco_sccfm_scripts/verify_public_release.py b/cisco_sccfm_scripts/verify_public_release.py index fed3ae7d..eebf391d 100644 --- a/cisco_sccfm_scripts/verify_public_release.py +++ b/cisco_sccfm_scripts/verify_public_release.py @@ -20,6 +20,7 @@ from urllib.request import Request, urlopen from cisco_sccfm_scripts.verify_clean_controller import ( + _PROFILE_ERROR, _Controller, _create_controller, _discovered_plugins, @@ -53,6 +54,8 @@ class PublicReleaseSummary: inventory_count: int lookup_count: int offline_probe: str + inventory_probe: str + lookup_probe: str def _read_json(response: Any, expected_url: str) -> object: @@ -191,14 +194,35 @@ def _verify_cli(controller: _Controller, version: str) -> None: schema: object = json.loads(schema_raw) except json.JSONDecodeError as exc: raise PublicReleaseVerificationError("CLI schema is not valid JSON") from exc + _validate_cli_schema(schema, version) + + +def _validate_cli_schema(schema: object, version: str) -> None: + """Require the public schema contract used by safe automation consumers.""" commands = schema.get("commands") if isinstance(schema, dict) else None - if ( - not isinstance(schema, dict) - or schema.get("version") != version - or not isinstance(commands, list) - or not commands - ): + if not isinstance(schema, dict) or schema.get("version") != version: raise PublicReleaseVerificationError("CLI schema does not describe the installed release") + if not isinstance(commands, list) or not commands: + raise PublicReleaseVerificationError("CLI schema did not expose any commands") + for command in commands: + if not isinstance(command, dict): + raise PublicReleaseVerificationError("CLI schema contains an invalid command") + path = command.get("path") + auth = command.get("auth") + if ( + not isinstance(path, list) + or not path + or any(not isinstance(part, str) or not part for part in path) + or not isinstance(command.get("command"), str) + or not command["command"] + or not isinstance(command.get("readonly"), bool) + or not isinstance(auth, dict) + or not isinstance(auth.get("requires_profile"), bool) + or not isinstance(auth.get("requires_api_token"), bool) + or not isinstance(command.get("options"), list) + or not isinstance(command.get("constraints"), list) + ): + raise PublicReleaseVerificationError("CLI schema command metadata is incomplete") def _verify_collection_version(controller: _Controller, version: str) -> None: @@ -230,17 +254,35 @@ def _discover_plugins( return discovered[0], discovered[1], discovered[2] +def _plugin_documentation( + controller: _Controller, + plugin_type: str, + name: str, +) -> dict[str, object]: + """Load and validate one discovered plugin's ansible-doc payload.""" + raw = _run( + controller, + [controller.binaries / "ansible-doc", "-j", "-t", plugin_type, name], + ).stdout + try: + payload: object = json.loads(raw) + except json.JSONDecodeError as exc: + raise PublicReleaseVerificationError(f"{plugin_type} documentation is invalid") from exc + plugin = payload.get(name) if isinstance(payload, dict) else None + doc = plugin.get("doc") if isinstance(plugin, dict) else None + if not isinstance(doc, dict): + raise PublicReleaseVerificationError(f"{plugin_type} documentation is missing") + return doc + + def _documented_probe(controller: _Controller, modules: dict[str, str]) -> str: """Select a documented readonly list module with no required business arguments.""" ansible_doc = controller.binaries / "ansible-doc" for name, description in modules.items(): if not description.casefold().startswith("list "): continue - raw = _run(controller, [ansible_doc, "-j", name]).stdout - try: - payload: object = json.loads(raw) - except json.JSONDecodeError: - continue + raw = _run(controller, [ansible_doc, "-j", "-t", "module", name]).stdout + payload: object = json.loads(raw) module = payload.get(name) if isinstance(payload, dict) else None doc = module.get("doc") if isinstance(module, dict) else None options = doc.get("options") if isinstance(doc, dict) else None @@ -252,6 +294,86 @@ def _documented_probe(controller: _Controller, modules: dict[str, str]) -> str: raise PublicReleaseVerificationError("no argument-free readonly list module was discovered") +def _expect_missing_profile( + controller: _Controller, + command: list[str | Path], + surface: str, + *, + allow_zero: bool = False, +) -> None: + """Require an offline runtime path to stop at profile validation.""" + result = _run(controller, command, check=False) + rendered = f"{result.stdout}\n{result.stderr}" + if (result.returncode == 0 and not allow_zero) or _PROFILE_ERROR not in rendered: + raise PublicReleaseVerificationError(f"{surface} did not reach missing-profile validation") + + +def _inventory_runtime_probe(controller: _Controller, inventory: dict[str, str]) -> str: + """Load a documented inventory plugin without contacting SCCFM.""" + for name in inventory: + doc = _plugin_documentation(controller, "inventory", name) + options = doc.get("options") + plugin_option = options.get("plugin") if isinstance(options, dict) else None + choices = plugin_option.get("choices") if isinstance(plugin_option, dict) else None + if ( + isinstance(plugin_option, dict) + and plugin_option.get("required") is True + and isinstance(choices, list) + and name in choices + ): + config = controller.work / "inventory.sccfm.yml" + config.write_text(f"plugin: {json.dumps(name)}\n", encoding="utf-8") + _expect_missing_profile( + controller, + [controller.binaries / "ansible-inventory", "-i", config, "--graph"], + "inventory plugin", + allow_zero=True, + ) + return name + raise PublicReleaseVerificationError("no safe inventory runtime probe was discovered") + + +def _lookup_runtime_probe(controller: _Controller, lookups: dict[str, str]) -> str: + """Load a documented lookup plugin using an explicitly non-secret field.""" + for name in lookups: + doc = _plugin_documentation(controller, "lookup", name) + options = doc.get("options") + if not isinstance(options, dict): + continue + safe_option = None + for option_name, option in options.items(): + choices = option.get("choices") if isinstance(option, dict) else None + if ( + isinstance(option_name, str) + and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", option_name) + and isinstance(choices, list) + and "region" in choices + ): + safe_option = option_name + break + terms = options.get("_terms") + if safe_option is None or not isinstance(terms, dict) or terms.get("required") is not True: + continue + expression = f"{{{{ lookup('{name}', 'default', {safe_option}='region') }}}}" + playbook = controller.work / "lookup-probe.yml" + playbook.write_text( + "---\n" + "- hosts: localhost\n" + " gather_facts: false\n" + " tasks:\n" + " - ansible.builtin.debug:\n" + f" msg: {json.dumps(expression)}\n", + encoding="utf-8", + ) + _expect_missing_profile( + controller, + [controller.binaries / "ansible-playbook", playbook], + "lookup plugin", + ) + return name + raise PublicReleaseVerificationError("no safe lookup runtime probe was discovered") + + def verify_public_release(requested: str = "") -> PublicReleaseSummary: """Install and smoke-test one matching release from the public registries.""" version = resolve_public_version(requested) @@ -263,12 +385,16 @@ def verify_public_release(requested: str = "") -> PublicReleaseSummary: modules, inventory, lookups = _discover_plugins(controller) probe = _documented_probe(controller, modules) _offline_checks(controller, probe) + inventory_probe = _inventory_runtime_probe(controller, inventory) + lookup_probe = _lookup_runtime_probe(controller, lookups) return PublicReleaseSummary( version=version, module_count=len(modules), inventory_count=len(inventory), lookup_count=len(lookups), offline_probe=probe, + inventory_probe=inventory_probe, + lookup_probe=lookup_probe, ) @@ -301,7 +427,8 @@ def main(argv: Sequence[str] | None = None) -> int: print( f"Public release verified: version={summary.version} modules={summary.module_count} " f"inventory={summary.inventory_count} lookups={summary.lookup_count} " - f"probe={summary.offline_probe}" + f"module_probe={summary.offline_probe} inventory_probe={summary.inventory_probe} " + f"lookup_probe={summary.lookup_probe}" ) return 0 diff --git a/tests/test_verify_public_release.py b/tests/test_verify_public_release.py index 82f55603..d1edb75d 100644 --- a/tests/test_verify_public_release.py +++ b/tests/test_verify_public_release.py @@ -7,6 +7,7 @@ from __future__ import annotations import json +import subprocess from pathlib import Path import pytest @@ -16,6 +17,16 @@ _VERSION = "1.2.3" +def _controller(tmp_path: Path) -> verifier._Controller: + work = tmp_path / "work" + work.mkdir() + binaries = tmp_path / "bin" + binaries.mkdir() + collections = tmp_path / "collections" + collections.mkdir() + return verifier._Controller(work, collections, binaries, {}) + + def test_resolve_latest_requires_matching_registry_versions( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -71,18 +82,154 @@ def test_resolve_rejects_registry_mismatch(monkeypatch: pytest.MonkeyPatch) -> N verifier.resolve_public_version() +def test_cli_schema_requires_safety_and_auth_metadata() -> None: + schema = { + "version": _VERSION, + "commands": [ + { + "path": ["status"], + "command": "sccfm-cli status", + "readonly": True, + "auth": {"requires_profile": True, "requires_api_token": True}, + "options": [], + "constraints": [], + } + ], + } + + verifier._validate_cli_schema(schema, _VERSION) + del schema["commands"][0]["readonly"] + with pytest.raises(verifier.PublicReleaseVerificationError, match="metadata is incomplete"): + verifier._validate_cli_schema(schema, _VERSION) + + +def test_inventory_probe_loads_documented_plugin( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = _controller(tmp_path) + plugin = "cisco.sccfm.dynamic" + monkeypatch.setattr( + verifier, + "_plugin_documentation", + lambda selected, plugin_type, name: { + "options": {"plugin": {"required": True, "choices": [plugin]}} + }, + ) + commands: list[list[str | Path]] = [] + monkeypatch.setattr( + verifier, + "_expect_missing_profile", + lambda selected, command, surface, allow_zero: commands.append(command), + ) + + assert verifier._inventory_runtime_probe(controller, {plugin: "Load inventory"}) == plugin + assert controller.work.joinpath("inventory.sccfm.yml").read_text() == ( + 'plugin: "cisco.sccfm.dynamic"\n' + ) + assert commands == [ + [ + controller.binaries / "ansible-inventory", + "-i", + controller.work / "inventory.sccfm.yml", + "--graph", + ] + ] + + +def test_lookup_probe_requests_only_documented_region_field( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = _controller(tmp_path) + plugin = "cisco.sccfm.profile" + monkeypatch.setattr( + verifier, + "_plugin_documentation", + lambda selected, plugin_type, name: { + "options": { + "_terms": {"required": True}, + "field": {"choices": ["region", "api_token"]}, + } + }, + ) + commands: list[list[str | Path]] = [] + monkeypatch.setattr( + verifier, + "_expect_missing_profile", + lambda selected, command, surface: commands.append(command), + ) + + assert verifier._lookup_runtime_probe(controller, {plugin: "Read profile"}) == plugin + playbook = controller.work.joinpath("lookup-probe.yml").read_text(encoding="utf-8") + assert "field='region'" in playbook + assert "api_token" not in playbook + assert commands == [ + [controller.binaries / "ansible-playbook", controller.work / "lookup-probe.yml"] + ] + + +def test_missing_profile_probe_rejects_other_failures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = _controller(tmp_path) + monkeypatch.setattr( + verifier, + "_run", + lambda selected, command, check: subprocess.CompletedProcess(command, 2, "", "other"), + ) + + with pytest.raises(verifier.PublicReleaseVerificationError, match="lookup plugin"): + verifier._expect_missing_profile(controller, ["ansible-playbook"], "lookup plugin") + + +def test_missing_profile_probe_accepts_inventory_warning_with_zero_exit( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + controller = _controller(tmp_path) + monkeypatch.setattr( + verifier, + "_run", + lambda selected, command, check: subprocess.CompletedProcess( + command, + 0, + "", + "SCCFM profile 'default' not found", + ), + ) + + verifier._expect_missing_profile( + controller, + ["ansible-inventory"], + "inventory plugin", + allow_zero=True, + ) + + def test_main_writes_machine_readable_summary( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: report = tmp_path / "reports" / "summary.json" - summary = verifier.PublicReleaseSummary(_VERSION, 49, 1, 1, "cisco.sccfm.list_devices") + summary = verifier.PublicReleaseSummary( + _VERSION, + 49, + 1, + 1, + "cisco.sccfm.list_devices", + "cisco.sccfm.sccfm", + "cisco.sccfm.profile", + ) monkeypatch.setattr(verifier, "verify_public_release", lambda requested: summary) assert verifier.main(["--version", _VERSION, "--report", str(report)]) == 0 assert json.loads(report.read_text(encoding="utf-8")) == { "inventory_count": 1, + "inventory_probe": "cisco.sccfm.sccfm", "lookup_count": 1, + "lookup_probe": "cisco.sccfm.profile", "module_count": 49, "offline_probe": "cisco.sccfm.list_devices", "version": _VERSION,