diff --git a/cisco_sccfm_cli/commands/inventory/devices/asa/cli/execute/command.py b/cisco_sccfm_cli/commands/inventory/devices/asa/cli/execute/command.py index 1257301d..bb019f5b 100644 --- a/cisco_sccfm_cli/commands/inventory/devices/asa/cli/execute/command.py +++ b/cisco_sccfm_cli/commands/inventory/devices/asa/cli/execute/command.py @@ -20,7 +20,7 @@ ) from cisco_sccfm_cli.commands.inventory.options import config_path_option, format_option from cisco_sccfm_cli.utils import with_spinner -from cisco_sccfm_core import AsaCommandLineService +from cisco_sccfm_core import AsaCommandLineService, is_failed_cli_result class AsaExecuteCliCommand(AsaDeviceTargetCommand): @@ -30,7 +30,10 @@ def name(self) -> str: @property def help_text(self) -> str: - return "Execute CLI commands on ASA devices." + return ( + "Execute CLI commands on ASA devices. Exits with status 1 if SCC Firewall Manager " + "reports a command failure on any device; every device's result is still printed." + ) def build_params(self) -> Sequence[click.Parameter]: return [ @@ -115,6 +118,12 @@ def _render_results( script=script, output_format=format, ) + failed = [result for result in results if is_failed_cli_result(result)] + if failed: + names = ", ".join(_device_label(result.device_uid, uid_to_device) for result in failed) + raise click.ClickException( + f"CLI commands failed on {len(failed)} of {len(results)} device(s): {names}" + ) def _validate_script_filters( self, @@ -127,3 +136,8 @@ def _validate_script_filters( has_script_file = bool(script_file) if has_script == has_script_file: ctx.fail("Provide exactly one of --script or --script-file.") + + +def _device_label(device_uid: str, uid_to_device: dict[str, Device]) -> str: + device = uid_to_device.get(device_uid) + return device.name if device else device_uid diff --git a/cisco_sccfm_cli/commands/inventory/devices/asa/command.py b/cisco_sccfm_cli/commands/inventory/devices/asa/command.py index 5d17bfd0..8a5f4cb5 100644 --- a/cisco_sccfm_cli/commands/inventory/devices/asa/command.py +++ b/cisco_sccfm_cli/commands/inventory/devices/asa/command.py @@ -26,6 +26,7 @@ AsaListNotOnVersionCommand, ) from cisco_sccfm_cli.commands.inventory.devices.asa.onboard import AsaOnboardCommand +from cisco_sccfm_cli.commands.inventory.devices.asa.read_config import AsaReadConfigCommand from cisco_sccfm_cli.commands.inventory.devices.asa.shun import AsaShunCommand from cisco_sccfm_cli.commands.inventory.devices.asa.smartlicense.command import SmartlicenseCommand from cisco_sccfm_cli.commands.inventory.devices.asa.upgrade import AsaUpgradeCommand @@ -53,6 +54,7 @@ def __init__(self, console: Console) -> None: SmartlicenseCommand(console), AsaOnboardCommand(console), AsaListLocalUsersCommand(console), + AsaReadConfigCommand(console), AsaShunCommand(console), AsaUpgradeCommand(console), AsaUserCommand(console), diff --git a/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/__init__.py b/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/__init__.py new file mode 100644 index 00000000..fe92a215 --- /dev/null +++ b/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/__init__.py @@ -0,0 +1,11 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +__all__ = ["AsaReadConfigCommand"] + +from cisco_sccfm_cli.commands.inventory.devices.asa.read_config.command import ( + AsaReadConfigCommand, +) diff --git a/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/command.py b/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/command.py new file mode 100644 index 00000000..8049fd2f --- /dev/null +++ b/cisco_sccfm_cli/commands/inventory/devices/asa/read_config/command.py @@ -0,0 +1,98 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import Any, Sequence, cast + +import click +from scc_firewall_manager_sdk import CdoTransaction + +from cisco_sccfm_cli.commands.inventory.devices.asa.shared import ( + AsaDeviceTargetCommand, + asa_check_option, + asa_device_filter_params, +) +from cisco_sccfm_cli.commands.inventory.options import ( + config_path_option, + format_option, + timeout_option, + wait_option, +) +from cisco_sccfm_cli.utils import print_json +from cisco_sccfm_core.services.inventory import AsaConfigReadService + + +class AsaReadConfigCommand(AsaDeviceTargetCommand): + @property + def name(self) -> str: + return "read-config" + + @property + def help_text(self) -> str: + return ( + "Read the running configuration of ASA devices into SCC Firewall Manager. " + "Use after changing a device outside SCC Firewall Manager (for example with " + "'cli execute'). The read replaces SCC Firewall Manager's copy of the " + "configuration, so changes staged there but not yet deployed can be lost." + ) + + def build_params(self) -> Sequence[click.Parameter]: + return [ + *asa_device_filter_params( + include_device_name=True, + query_help_text="Filter devices to read by a Lucene query.", + device_uids_help_text="List of device UIDs to read.", + ), + asa_check_option(), + wait_option(), + timeout_option(), + format_option(), + config_path_option(), + ] + + def handle(self, ctx: click.Context, **kwargs: Any) -> None: + output_format = cast(str, kwargs.get("format")) + config = self.get_profile(ctx=ctx, **kwargs) + targets = self.resolve_asa_targets_from_kwargs( + ctx=ctx, + kwargs=kwargs, + config=config, + include_device_name=True, + ) + + if cast(bool, kwargs.get("check", False)): + self.report_check_targets( + targets, output_format=output_format, operation="configuration read" + ) + return + if not targets.device_uids: + raise click.ClickException("No ASA devices matched the filter.") + + transaction = AsaConfigReadService(config=config).read_configuration( + device_uids=targets.device_uids + ) + transaction = self.wait_for_transaction( + config=config, + transaction=transaction, + spinner_text="Reading ASA configuration..." if output_format != "json" else None, + **kwargs, + ) + self._render( + transaction, device_count=len(targets.device_uids), output_format=output_format + ) + + def _render( + self, transaction: CdoTransaction, *, device_count: int, output_format: str + ) -> None: + if self.is_failed_transaction(transaction): + self.print_failed_transaction_details(cdo_transaction=transaction, format=output_format) + if output_format == "json": + print_json(transaction.to_dict()) + return + self.console.print( + f"[green]✓[/green] Configuration read for {device_count} device(s): " + f"{transaction.cdo_transaction_status}" + ) + self.console.print(f"[bold]Transaction UID:[/bold] {transaction.transaction_uid}") diff --git a/cisco_sccfm_cli/commands/tests/inventory/devices/asa/cli/execute/test_execute_cli.py b/cisco_sccfm_cli/commands/tests/inventory/devices/asa/cli/execute/test_execute_cli.py index 1be1705d..2c5ac36b 100644 --- a/cisco_sccfm_cli/commands/tests/inventory/devices/asa/cli/execute/test_execute_cli.py +++ b/cisco_sccfm_cli/commands/tests/inventory/devices/asa/cli/execute/test_execute_cli.py @@ -319,3 +319,97 @@ def stub_asa_cli_init(self: AsaCommandLineService, config: Any) -> None: assert payload["matched_devices"] == 1 assert payload["devices"][0]["uid"] == "uid-asa-1" assert payload["devices"][0]["device_type"] == "ASA" + + +def _patch_execute_with_results( + monkeypatch: MonkeyPatch, devices: list[Device], results: list[CdoCliResult] +) -> None: + def fake_get_devices( + self: InventoryService, *, limit: int, offset: int, query: str | None = None + ) -> DevicePage: + return DevicePage(count=len(devices), items=devices) + + def fake_execute_cli( + self: AsaCommandLineService, *, device_uids: list[str], asa_commands: list[str] + ) -> list[CdoCliResult]: + return results + + def stub_asa_cli_init(self: AsaCommandLineService, config: Any) -> None: + return None + + monkeypatch.setattr(InventoryService, "get_devices", fake_get_devices) + monkeypatch.setattr(AsaCommandLineService, "execute_cli", fake_execute_cli) + monkeypatch.setattr(AsaCommandLineService, "__init__", stub_asa_cli_init) + + +def _results_with_one_failure() -> list[CdoCliResult]: + return [ + CdoCliResult(uid="r-1", device_uid="uid-1", result="", error_msg=None), + CdoCliResult(uid="r-2", device_uid="uid-2", result="Command failed", error_msg=None), + ] + + +def test_should_exit_non_zero_and_keep_json_results_when_a_device_fails( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, + sample_devices: list[Device], +) -> None: + """A per-device 'Command failed' must not be reported as success.""" + results = _results_with_one_failure() + _patch_execute_with_results(monkeypatch, sample_devices, results) + + result = cli_runner.invoke( + cli, + [ + "inventory", + "devices", + "asa", + "cli", + "execute", + "-u", + "uid-1", + "-u", + "uid-2", + "--script", + "access-list acl permit ip any any", + "--format", + "json", + ], + ) + + assert result.exit_code == 1 + assert json.loads(result.stdout) == [item.model_dump(mode="json") for item in results] + assert "CLI commands failed on 1 of 2 device(s): edge-nva" in result.stderr + + +def test_should_exit_non_zero_after_printing_table_when_a_device_fails( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, + sample_devices: list[Device], +) -> None: + _patch_execute_with_results(monkeypatch, sample_devices, _results_with_one_failure()) + + result = cli_runner.invoke( + cli, + [ + "inventory", + "devices", + "asa", + "cli", + "execute", + "-u", + "uid-1", + "-u", + "uid-2", + "--script", + "access-list acl permit ip any any", + ], + ) + + assert result.exit_code == 1 + assert "Command failed" in result.stdout + assert "CLI commands failed on 1 of 2 device(s): edge-nva" in result.stderr diff --git a/cisco_sccfm_cli/commands/tests/inventory/devices/asa/read_config/test_read_config.py b/cisco_sccfm_cli/commands/tests/inventory/devices/asa/read_config/test_read_config.py new file mode 100644 index 00000000..4badd2ae --- /dev/null +++ b/cisco_sccfm_cli/commands/tests/inventory/devices/asa/read_config/test_read_config.py @@ -0,0 +1,154 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``cisco_sccfm_cli inventory devices asa read-config`` command.""" + +from __future__ import annotations + +import json +from typing import Any + +from _pytest.monkeypatch import MonkeyPatch +from click.testing import CliRunner +from scc_firewall_manager_sdk import CdoTransaction, Device, DevicePage, EntityType + +from cisco_sccfm_cli.cli import cli +from cisco_sccfm_cli.models import Config +from cisco_sccfm_core.services import InventoryService +from cisco_sccfm_core.services.inventory import AsaConfigReadService +from cisco_sccfm_core.services.transaction_service import TransactionService + +_READ_CONFIG = ["inventory", "devices", "asa", "read-config"] + + +def _transaction(status: str = "PENDING") -> CdoTransaction: + return CdoTransaction(transactionUid="txn-1", cdoTransactionStatus=status) + + +def _devices(count: int) -> list[Device]: + return [ + Device(uid=f"uid-{i}", name=f"branch-asa-0{i}", deviceType=EntityType.ASA) + for i in range(1, count + 1) + ] + + +def _patch_inventory(monkeypatch: MonkeyPatch, devices: list[Device]) -> None: + def fake_get_devices( + self: InventoryService, *, limit: int, offset: int, query: str | None = None + ) -> DevicePage: + return DevicePage(count=len(devices), items=devices) + + monkeypatch.setattr(InventoryService, "get_devices", fake_get_devices) + + +def _patch_read(monkeypatch: MonkeyPatch, captured: dict[str, Any]) -> None: + def stub_init(self: AsaConfigReadService, config: Any) -> None: + return None + + def fake_read(self: AsaConfigReadService, *, device_uids: list[str]) -> CdoTransaction: + captured["device_uids"] = device_uids + return _transaction() + + monkeypatch.setattr(AsaConfigReadService, "__init__", stub_init) + monkeypatch.setattr(AsaConfigReadService, "read_configuration", fake_read) + + +def _patch_wait(monkeypatch: MonkeyPatch, final_status: str) -> None: + def stub_init(self: TransactionService, config: Any) -> None: + return None + + def fake_wait(self: TransactionService, **kwargs: Any) -> CdoTransaction: + return _transaction(final_status) + + monkeypatch.setattr(TransactionService, "__init__", stub_init) + monkeypatch.setattr(TransactionService, "wait_for_transaction_to_finish", fake_wait) + + +def test_should_read_all_matched_devices_and_return_transaction_json( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, +) -> None: + _patch_inventory(monkeypatch, _devices(2)) + captured: dict[str, Any] = {} + _patch_read(monkeypatch, captured) + + result = cli_runner.invoke(cli, [*_READ_CONFIG, "--query", "name:branch-*", "--format", "json"]) + + assert result.exit_code == 0, f"Command failed: {result.output}" + assert json.loads(result.output)["transactionUid"] == "txn-1" + assert captured["device_uids"] == ["uid-1", "uid-2"] + + +def test_should_wait_for_transaction_when_requested( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, +) -> None: + _patch_inventory(monkeypatch, _devices(1)) + _patch_read(monkeypatch, {}) + _patch_wait(monkeypatch, "DONE") + + result = cli_runner.invoke( + cli, ["--silent", *_READ_CONFIG, "-u", "uid-1", "--wait", "--format", "json"] + ) + + assert result.exit_code == 0, f"Command failed: {result.output}" + assert json.loads(result.output)["cdoTransactionStatus"] == "DONE" + + +def test_should_exit_non_zero_when_read_transaction_fails( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, +) -> None: + _patch_inventory(monkeypatch, _devices(1)) + _patch_read(monkeypatch, {}) + _patch_wait(monkeypatch, "ERROR") + + result = cli_runner.invoke( + cli, ["--silent", *_READ_CONFIG, "-u", "uid-1", "--wait", "--format", "json"] + ) + + assert result.exit_code != 0 + assert json.loads(result.output)["cdoTransactionStatus"] == "ERROR" + + +def test_should_fail_when_no_devices_match( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, +) -> None: + _patch_inventory(monkeypatch, []) + captured: dict[str, Any] = {} + _patch_read(monkeypatch, captured) + + result = cli_runner.invoke(cli, [*_READ_CONFIG, "--query", "name:missing-*"]) + + assert result.exit_code != 0 + assert "No ASA devices matched" in result.output + assert captured == {} + + +def test_check_mode_should_report_targets_without_reading( + cli_runner: CliRunner, + default_config: Config, + mock_inventory_service: None, + monkeypatch: MonkeyPatch, +) -> None: + _patch_inventory(monkeypatch, _devices(1)) + captured: dict[str, Any] = {} + _patch_read(monkeypatch, captured) + + result = cli_runner.invoke(cli, [*_READ_CONFIG, "-u", "uid-1", "--check", "--format", "json"]) + + assert result.exit_code == 0, f"Command failed: {result.output}" + payload = json.loads(result.output) + assert payload["operation"] == "configuration read" + assert payload["matched_devices"] == 1 + assert captured == {} diff --git a/cisco_sccfm_cli/e2e/access_rules/phases/cleanup.py b/cisco_sccfm_cli/e2e/access_rules/phases/cleanup.py index 24bd9a31..84ce6e49 100644 --- a/cisco_sccfm_cli/e2e/access_rules/phases/cleanup.py +++ b/cisco_sccfm_cli/e2e/access_rules/phases/cleanup.py @@ -47,10 +47,8 @@ def run(ctx: ProfileContext) -> None: # will succeed once the device has synced. teardown_script = "\n".join( [ - "configure terminal", f"no access-group {TEST_ACL_NAME} global", f"clear configure access-list {TEST_ACL_NAME}", - "end", ] ) run_cli( diff --git a/cisco_sccfm_cli/e2e/access_rules/phases/list_access_groups.py b/cisco_sccfm_cli/e2e/access_rules/phases/list_access_groups.py index 2abf2253..47841206 100644 --- a/cisco_sccfm_cli/e2e/access_rules/phases/list_access_groups.py +++ b/cisco_sccfm_cli/e2e/access_rules/phases/list_access_groups.py @@ -2,7 +2,12 @@ # # SPDX-License-Identifier: Apache-2.0 -"""List access groups for the test ACL, retrying until CDO indexes it.""" +"""List access groups for the test ACL, retrying until CDO indexes it. + +If it never appears, the failure includes read-only diagnostics: the +provisioning response, the configuration read, device sync state, the ACL +config on the ASA itself, and every access group SCCFM knows about. +""" from __future__ import annotations @@ -12,8 +17,10 @@ from cisco_sccfm_cli.e2e._runner import get_json, run_cli from cisco_sccfm_cli.e2e.access_rules.phases.test_data import ( ACCESS_GROUP_QUERY, + ASA_TEST_QUERY, LIST_ACCESS_GROUP_DELAY_SEC, LIST_ACCESS_GROUP_RETRIES, + TEST_ACL_NAME, ) @@ -43,5 +50,57 @@ def run(ctx: ProfileContext) -> None: raise AssertionError( f"Expected at least one access group for query {ACCESS_GROUP_QUERY!r} " - f"after {LIST_ACCESS_GROUP_RETRIES} retries. Last payload: {last_payload!r}" + f"after {LIST_ACCESS_GROUP_RETRIES} retries. Last payload: {last_payload!r}\n" + f"{_diagnostics(ctx)}" + ) + + +def _diagnostics(ctx: ProfileContext) -> str: + """Collect read-only state that explains why no access group appeared.""" + acl_show_script = ( + f"show running-config access-group\nshow running-config access-list {TEST_ACL_NAME}" ) + sections = { + "Provisioning response (error_msg is the ASA's reply)": str( + ctx.state.pop("provision_cli_output", "(not recorded: provision phase did not finish)") + ), + "Configuration read transaction (status, timings, details)": str( + ctx.state.pop("read_config_output", "(not recorded: provision did not reach the read)") + ), + "CI ASA devices and their sync state": _cli_output( + ctx, "inventory", "devices", "asa", "list", "--query", ASA_TEST_QUERY, "--limit", "10" + ), + "ACL config on the ASA itself": _cli_output( + ctx, + "inventory", + "devices", + "asa", + "cli", + "execute", + "--query", + ASA_TEST_QUERY, + "--script", + acl_show_script, + ), + "All access groups, unfiltered": _cli_output( + ctx, "policies", "access-group", "list", "--limit", "50" + ), + } + return "\n".join(f"--- {title} ---\n{output}" for title, output in sections.items()) + + +def _cli_output(ctx: ProfileContext, *args: str) -> str: + """Run a diagnostic command as JSON, never raising, so the original failure is kept.""" + try: + result = run_cli( + *args, + "--format", + "json", + profile=ctx.profile, + config_path=ctx.config_path, + tolerate_any_rc=True, + parse_json=False, + ) + except AssertionError as exc: # run_cli reports timeouts as AssertionError + return str(exc) + return (result.stdout + result.stderr).strip() or f"(no output, rc={result.returncode})" diff --git a/cisco_sccfm_cli/e2e/access_rules/phases/provision_access_group.py b/cisco_sccfm_cli/e2e/access_rules/phases/provision_access_group.py index 1beb6de8..8fee41e3 100644 --- a/cisco_sccfm_cli/e2e/access_rules/phases/provision_access_group.py +++ b/cisco_sccfm_cli/e2e/access_rules/phases/provision_access_group.py @@ -6,8 +6,11 @@ from __future__ import annotations +import re + +from cisco_sccfm_cli.e2e._payload import normalize_rows from cisco_sccfm_cli.e2e._profile import ProfileContext -from cisco_sccfm_cli.e2e._runner import run_cli +from cisco_sccfm_cli.e2e._runner import get_json, run_cli from cisco_sccfm_cli.e2e._sync import wait_for_synced from cisco_sccfm_cli.e2e.access_rules.phases.test_data import ( ASA_TEST_QUERY, @@ -18,6 +21,9 @@ TEST_SOURCE_VALUE, ) +# Same check asa_boot_image_service applies to CLI results. +_CLI_ERROR_RE = re.compile(r"(?im)^ERROR:|Command failed") + def run(ctx: ProfileContext) -> None: # Source / destination network objects. Tolerate "already exists" @@ -53,15 +59,16 @@ def run(ctx: ProfileContext) -> None: # mid-sync, and the API rejects script pushes against NOT_SYNCED devices. wait_for_synced(ctx, query=ASA_TEST_QUERY) + # No "configure terminal"/"end": SCCFM applies scripts containing them + # but reports "Command failed". The core services (e.g. + # asa_boot_image_service) also send bare config lines. provisioning_script = "\n".join( [ - "configure terminal", f"access-list {TEST_ACL_NAME} extended permit ip any any", f"access-group {TEST_ACL_NAME} global", - "end", ] ) - run_cli( + result = run_cli( "inventory", "devices", "asa", @@ -76,3 +83,44 @@ def run(ctx: ProfileContext) -> None: profile=ctx.profile, config_path=ctx.config_path, ) + # pytest hides output from passing phases. Keep the raw output so + # list_access_groups can report it. + ctx.state.set("provision_cli_output", result.stdout) + _assert_accepted(get_json(result)) + + # SCCFM does not re-read the device after a CLI push: the ACL is on the + # ASA but no access group appears until the configuration is read. + read_result = run_cli( + "inventory", + "devices", + "asa", + "read-config", + "--query", + ASA_TEST_QUERY, + "--wait", + "--timeout", + "120", + "--format", + "json", + profile=ctx.profile, + config_path=ctx.config_path, + ) + ctx.state.set("read_config_output", read_result.stdout) + + +def _assert_accepted(payload: object) -> None: + """Fail if SCCFM reports the script as rejected. + + ``cli execute`` exits 0 even then: the rejection is in each row's + ``error_msg`` or as "Command failed"/"ERROR:" in its ``result``. + """ + rows = normalize_rows(payload) + rejected = [ + row + for row in rows + if row.get("error_msg") or _CLI_ERROR_RE.search(str(row.get("result") or "")) + ] + if len(rows) != 1 or rejected: + raise AssertionError( + f"Expected exactly one device with no CLI errors, got {len(rows)} result(s): {rows!r}" + ) diff --git a/cisco_sccfm_cli/e2e/run_e2e.sh b/cisco_sccfm_cli/e2e/run_e2e.sh index e16feaea..aab18581 100755 --- a/cisco_sccfm_cli/e2e/run_e2e.sh +++ b/cisco_sccfm_cli/e2e/run_e2e.sh @@ -45,6 +45,14 @@ fi echo "Installing sccfm package (with dev extras)..." poetry install --no-interaction --with dev +# ── Debug hold: see sccfm-ansible/e2e/run_e2e.sh ────────────────── +# The Ansible runner keeps its vASA up for manual testing; skip this +# suite entirely, including onboarding the CLI vASA. +if [[ -f "${REPO_ROOT}/sccfm-ansible/e2e/HOLD_VASA_MINUTES" ]]; then + echo "HOLD: HOLD_VASA_MINUTES is set; skipping the sccfm-cli e2e suite." + exit 0 +fi + # ── Onboard a CLI-dedicated vASA (opt-in via ASA_HOST) ──────────── ONBOARDED_VASA=0 diff --git a/cisco_sccfm_core/__init__.py b/cisco_sccfm_core/__init__.py index b7b39439..8353d1c1 100644 --- a/cisco_sccfm_core/__init__.py +++ b/cisco_sccfm_core/__init__.py @@ -21,6 +21,7 @@ AsaBootImageService, AsaBootRegistryService, AsaCommandLineService, + AsaConfigReadService, AsaDiskFileService, AsaHaCheckReport, AsaHaCheckService, @@ -31,6 +32,7 @@ FtdCommandLineService, InventoryService, ShunEntrySpec, + is_failed_cli_result, ) from cisco_sccfm_core.services.object_management import ( NetworkGroupListResponse, @@ -75,6 +77,7 @@ "AsaBootImageService", "AsaBootRegistryService", "AsaCommandLineService", + "AsaConfigReadService", "AsaDiskFileService", "AsaHaCheckReport", "AsaHaCheckService", @@ -87,6 +90,7 @@ "HealthService", "HealthStatus", "InventoryService", + "is_failed_cli_result", "NetworkGroupListResponse", "NetworkGroupMemberMutationResult", "NetworkGroupResponse", diff --git a/cisco_sccfm_core/services/__init__.py b/cisco_sccfm_core/services/__init__.py index 1312a682..ef73f234 100644 --- a/cisco_sccfm_core/services/__init__.py +++ b/cisco_sccfm_core/services/__init__.py @@ -9,6 +9,7 @@ AsaBootImageService, AsaBootRegistryService, AsaCommandLineService, + AsaConfigReadService, AsaDiskFileService, AsaHaCheckReport, AsaHaCheckService, @@ -42,6 +43,7 @@ "AsaBootImageService", "AsaBootRegistryService", "AsaCommandLineService", + "AsaConfigReadService", "AsaDiskFileService", "AsaHaCheckReport", "AsaHaCheckService", diff --git a/cisco_sccfm_core/services/inventory/__init__.py b/cisco_sccfm_core/services/inventory/__init__.py index 131b9133..81930318 100644 --- a/cisco_sccfm_core/services/inventory/__init__.py +++ b/cisco_sccfm_core/services/inventory/__init__.py @@ -6,7 +6,11 @@ from cisco_sccfm_core.services.inventory.asa_boot_image_service import AsaBootImageService from cisco_sccfm_core.services.inventory.asa_boot_registry_service import AsaBootRegistryService -from cisco_sccfm_core.services.inventory.asa_cli_service import AsaCommandLineService +from cisco_sccfm_core.services.inventory.asa_cli_service import ( + AsaCommandLineService, + is_failed_cli_result, +) +from cisco_sccfm_core.services.inventory.asa_config_read_service import AsaConfigReadService from cisco_sccfm_core.services.inventory.asa_disk_file_service import AsaDiskFileService from cisco_sccfm_core.services.inventory.asa_ha_check_service import ( AsaHaCheckReport, @@ -52,6 +56,7 @@ "AsaBootImageService", "AsaBootRegistryService", "AsaCommandLineService", + "AsaConfigReadService", "AsaDiskFileService", "AsaHaCheckReport", "AsaHaCheckService", @@ -79,6 +84,7 @@ "InventoryService", "ShunEntrySpec", "get_asdm_compatibility_info", + "is_failed_cli_result", "is_version_downgrade", "resolve_upgrade_package_uid", ] diff --git a/cisco_sccfm_core/services/inventory/asa_cli_service.py b/cisco_sccfm_core/services/inventory/asa_cli_service.py index 63b189c4..ab2269c4 100644 --- a/cisco_sccfm_core/services/inventory/asa_cli_service.py +++ b/cisco_sccfm_core/services/inventory/asa_cli_service.py @@ -4,6 +4,8 @@ from __future__ import annotations +import re + from scc_firewall_manager_sdk import ( CdoCliResult, CdoTransaction, @@ -18,6 +20,19 @@ from cisco_sccfm_core.types import ConfigLike from cisco_sccfm_core.utils import validate_uids +# SCCFM reports "Command failed" as the whole result, and the ASA prefixes its +# own rejections with "ERROR:" (e.g. "ERROR: % Invalid input detected"). +_FAILED_RESULT_RE = re.compile(r"(?im)^\s*(?:ERROR:|Command failed)") + + +def is_failed_cli_result(result: CdoCliResult) -> bool: + """Return True if SCCFM reports the script as failed on this device. + + A completed CLI transaction can still hold per-device failures: either an + ``error_msg`` or a failure marker in the ``result`` text. + """ + return bool(result.error_msg) or bool(_FAILED_RESULT_RE.search(result.result or "")) + class AsaCommandLineService: def __init__(self, config: ConfigLike) -> None: diff --git a/cisco_sccfm_core/services/inventory/asa_config_read_service.py b/cisco_sccfm_core/services/inventory/asa_config_read_service.py new file mode 100644 index 00000000..2d12b6fb --- /dev/null +++ b/cisco_sccfm_core/services/inventory/asa_config_read_service.py @@ -0,0 +1,33 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from scc_firewall_manager_sdk import BulkOperationAsaDeviceRequest, CdoTransaction, InventoryApi + +from cisco_sccfm_core.factories import ApiClientFactory +from cisco_sccfm_core.types import ConfigLike +from cisco_sccfm_core.utils import validate_uids + + +class AsaConfigReadService: + """Reads the running configuration of ASA devices into SCC Firewall Manager. + + SCC Firewall Manager does not re-read a device after a change made outside it, + such as a CLI push, so its copy of the configuration (access groups, rules, + objects) stays stale until a read. A read replaces that copy, so changes staged + in SCC Firewall Manager but not yet deployed can be lost. + """ + + def __init__(self, config: ConfigLike) -> None: + self._inventory_api = InventoryApi(ApiClientFactory().build(config=config)) + + def read_configuration(self, *, device_uids: list[str]) -> CdoTransaction: + """Start a configuration read and return the :class:`CdoTransaction` tracking it.""" + validate_uids(device_uids) + if len(device_uids) == 1: + return self._inventory_api.read_asa_device_configuration(device_uid=device_uids[0]) + return self._inventory_api.bulk_read_asa_device_configurations( + bulk_operation_asa_device_request=BulkOperationAsaDeviceRequest(deviceUids=device_uids) + ) diff --git a/cisco_sccfm_core/services/inventory/asa_onboard_service.py b/cisco_sccfm_core/services/inventory/asa_onboard_service.py index 979c8971..a57bfbf9 100644 --- a/cisco_sccfm_core/services/inventory/asa_onboard_service.py +++ b/cisco_sccfm_core/services/inventory/asa_onboard_service.py @@ -4,6 +4,8 @@ from __future__ import annotations +from typing import cast + from scc_firewall_manager_sdk import AsaCreateOrUpdateInput, CdoTransaction, Device, InventoryApi from cisco_sccfm_core import ApiClientFactory @@ -12,15 +14,25 @@ from cisco_sccfm_core.services.transaction_service import TransactionService from cisco_sccfm_core.types import ConfigLike +# The SCCFM UI creates ASAs with this device metadata so they use the New ASA +# Policy View; without it the access-groups API never lists their ACLs. +_NEW_POLICY_VIEW_METADATA = {"isNewPolicyObjectModel": "true"} + class AsaOnboardService: def __init__(self, config: ConfigLike) -> None: self.inventory_api = InventoryApi(ApiClientFactory().build(config=config)) self.transaction_service = TransactionService(config=config) - def onboard_asa(self, asa_create_or_update_input: AsaCreateOrUpdateInput) -> Device: - transaction: CdoTransaction = self.inventory_api.onboard_asa_device( - asa_create_or_update_input=asa_create_or_update_input + def onboard_asa( + self, asa_create_or_update_input: AsaCreateOrUpdateInput, *, new_policy_view: bool = False + ) -> Device: + transaction: CdoTransaction = ( + self._onboard_with_new_policy_view(asa_create_or_update_input) + if new_policy_view + else self.inventory_api.onboard_asa_device( + asa_create_or_update_input=asa_create_or_update_input + ) ) completed_transaction: CdoTransaction = ( self.transaction_service.wait_for_transaction_to_finish( @@ -40,3 +52,28 @@ def onboard_asa(self, asa_create_or_update_input: AsaCreateOrUpdateInput) -> Dev raise Exception(error_msg) return self.inventory_api.get_device(device_uid=completed_transaction.entity_uid) + + def _onboard_with_new_policy_view( + self, asa_create_or_update_input: AsaCreateOrUpdateInput + ) -> CdoTransaction: + """Send the onboarding request with the device metadata the UI sends. + + AsaCreateOrUpdateInput has no metadata field and would drop it, so build + the same request as InventoryApi.onboard_asa_device from a plain body. + """ + api_client = self.inventory_api.api_client + request = api_client.param_serialize( + method="POST", + resource_path="/v1/inventory/devices/asas", + header_params={"Accept": "application/json", "Content-Type": "application/json"}, + body={**asa_create_or_update_input.to_dict(), "metadata": _NEW_POLICY_VIEW_METADATA}, + auth_settings=["bearerAuth"], + ) + response = api_client.call_api(*request) + response.read() + return cast( + CdoTransaction, + api_client.response_deserialize( + response_data=response, response_types_map={"202": "CdoTransaction"} + ).data, + ) diff --git a/cisco_sccfm_core/tests/test_asa_cli_service.py b/cisco_sccfm_core/tests/test_asa_cli_service.py index 81f6e519..4b7b1ccb 100644 --- a/cisco_sccfm_core/tests/test_asa_cli_service.py +++ b/cisco_sccfm_core/tests/test_asa_cli_service.py @@ -6,10 +6,14 @@ from unittest.mock import Mock +import pytest from scc_firewall_manager_sdk import CdoCliResult, CdoTransaction from cisco_sccfm_core.constants import FAST_POLLING_INTERVAL_SEC -from cisco_sccfm_core.services.inventory.asa_cli_service import AsaCommandLineService +from cisco_sccfm_core.services.inventory.asa_cli_service import ( + AsaCommandLineService, + is_failed_cli_result, +) DEVICE_UID = "11111111-1111-4111-8111-111111111111" @@ -69,3 +73,23 @@ def test_get_cli_result_uses_keyword_uid() -> None: service.command_line_interface_api.get_cli_result.assert_called_once_with( cli_result_uid="result-1" ) + + +@pytest.mark.parametrize( + ("result_text", "error_msg", "expected"), + [ + ("Cisco Adaptive Security Appliance Software Version 9.4(1)200", None, False), + ("", None, False), + (None, None, False), + ("Command failed", None, True), + ("access-list ok\nERROR: % Invalid input detected at '^' marker.", None, True), + ("", "Device unreachable", True), + ("hostname ERROR:-not-a-prefix", None, False), + ], +) +def test_is_failed_cli_result_detects_per_device_failures( + result_text: str | None, error_msg: str | None, expected: bool +) -> None: + result = CdoCliResult(uid="r1", deviceUid=DEVICE_UID, result=result_text, errorMsg=error_msg) + + assert is_failed_cli_result(result) is expected diff --git a/cisco_sccfm_core/tests/test_asa_config_read_service.py b/cisco_sccfm_core/tests/test_asa_config_read_service.py new file mode 100644 index 00000000..379178a3 --- /dev/null +++ b/cisco_sccfm_core/tests/test_asa_config_read_service.py @@ -0,0 +1,77 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for cisco_sccfm_core.services.inventory.asa_config_read_service module.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from scc_firewall_manager_sdk import BulkOperationAsaDeviceRequest, CdoTransaction + +from cisco_sccfm_core.services.inventory.asa_config_read_service import AsaConfigReadService + +_UID_1 = "a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d" +_UID_2 = "b2c3d4e5-f6a7-4b8c-9d0e-1f2a3b4c5d6e" + + +@pytest.fixture +def mock_config() -> MagicMock: + config = MagicMock() + config.region = "us" + config.api_token = "test-token" + return config + + +@pytest.fixture +def read_service(mock_config: MagicMock) -> AsaConfigReadService: + with patch("cisco_sccfm_core.services.inventory.asa_config_read_service.ApiClientFactory"): + return AsaConfigReadService(config=mock_config) + + +def _fake_transaction(uid: str) -> CdoTransaction: + return CdoTransaction(transactionUid=uid, cdoTransactionStatus="PENDING") + + +def test_should_read_single_device_with_single_device_endpoint( + read_service: AsaConfigReadService, monkeypatch: pytest.MonkeyPatch +) -> None: + mock_read = MagicMock(return_value=_fake_transaction("txn-single")) + mock_bulk = MagicMock() + monkeypatch.setattr(read_service._inventory_api, "read_asa_device_configuration", mock_read) + monkeypatch.setattr( + read_service._inventory_api, "bulk_read_asa_device_configurations", mock_bulk + ) + + result = read_service.read_configuration(device_uids=[_UID_1]) + + assert result.transaction_uid == "txn-single" + mock_read.assert_called_once_with(device_uid=_UID_1) + mock_bulk.assert_not_called() + + +def test_should_read_multiple_devices_with_bulk_endpoint( + read_service: AsaConfigReadService, monkeypatch: pytest.MonkeyPatch +) -> None: + mock_read = MagicMock() + mock_bulk = MagicMock(return_value=_fake_transaction("txn-bulk")) + monkeypatch.setattr(read_service._inventory_api, "read_asa_device_configuration", mock_read) + monkeypatch.setattr( + read_service._inventory_api, "bulk_read_asa_device_configurations", mock_bulk + ) + + result = read_service.read_configuration(device_uids=[_UID_1, _UID_2]) + + assert result.transaction_uid == "txn-bulk" + mock_read.assert_not_called() + request: BulkOperationAsaDeviceRequest = mock_bulk.call_args.kwargs[ + "bulk_operation_asa_device_request" + ] + assert request.device_uids == [_UID_1, _UID_2] + + +def test_should_reject_invalid_uid(read_service: AsaConfigReadService) -> None: + with pytest.raises(ValueError): + read_service.read_configuration(device_uids=["not-a-uuid"]) diff --git a/cisco_sccfm_core/tests/test_asa_onboard_service.py b/cisco_sccfm_core/tests/test_asa_onboard_service.py index 2f0c73f0..25ba42d4 100644 --- a/cisco_sccfm_core/tests/test_asa_onboard_service.py +++ b/cisco_sccfm_core/tests/test_asa_onboard_service.py @@ -7,7 +7,7 @@ from typing import cast from unittest.mock import Mock -from scc_firewall_manager_sdk import AsaCreateOrUpdateInput, CdoTransaction, Device +from scc_firewall_manager_sdk import AsaCreateOrUpdateInput, CdoTransaction, ConnectorType, Device from cisco_sccfm_core.constants import ONBOARD_POLLING_INTERVAL_SEC from cisco_sccfm_core.services.inventory.asa_onboard_service import AsaOnboardService @@ -35,3 +35,36 @@ def test_onboard_asa_waits_for_transaction_and_fetches_device() -> None: polling_interval_sec=ONBOARD_POLLING_INTERVAL_SEC, ) service.inventory_api.get_device.assert_called_once_with(device_uid="device-1") + + +def test_onboard_asa_with_new_policy_view_sends_ui_device_metadata() -> None: + service = AsaOnboardService.__new__(AsaOnboardService) + service.inventory_api = Mock() + service.transaction_service = Mock() + api_client = service.inventory_api.api_client + api_client.param_serialize.return_value = ("POST", "url", {}, "body", []) + api_client.response_deserialize.return_value = Mock(data=CdoTransaction(transactionUid="txn-1")) + service.transaction_service.wait_for_transaction_to_finish.return_value = CdoTransaction( + transactionUid="txn-1", cdoTransactionStatus="DONE", entityUid="device-1" + ) + asa_input = AsaCreateOrUpdateInput( + name="asa-1", + deviceAddress="198.51.100.1:443", + username="admin", + password="secret", + connectorType=ConnectorType.CDG, + ignoreCertificate=True, + ) + + service.onboard_asa(asa_input, new_policy_view=True) + + service.inventory_api.onboard_asa_device.assert_not_called() + serialize_kwargs = api_client.param_serialize.call_args.kwargs + assert serialize_kwargs["method"] == "POST" + assert serialize_kwargs["resource_path"] == "/v1/inventory/devices/asas" + assert serialize_kwargs["body"] == { + **asa_input.to_dict(), + "metadata": {"isNewPolicyObjectModel": "true"}, + } + api_client.call_api.assert_called_once_with("POST", "url", {}, "body", []) + service.inventory_api.get_device.assert_called_once_with(device_uid="device-1") diff --git a/cisco_sccfm_scripts/consistency_check.py b/cisco_sccfm_scripts/consistency_check.py index 1ad099aa..e3d6f49a 100644 --- a/cisco_sccfm_scripts/consistency_check.py +++ b/cisco_sccfm_scripts/consistency_check.py @@ -1072,6 +1072,7 @@ def _cli_expected_ansible_module(file: Path) -> Path | None: "list_not_on_version": "list_asa_not_on_version", "change_boot_image": "change_asa_boot_image", "onboard": "onboard_asa", + "read_config": "read_asa_config", } mapped_module = special.get(parts[3]) return ANSIBLE_MODULES / f"{mapped_module}.py" if mapped_module else None diff --git a/sccfm-ansible/e2e/HOLD_VASA_MINUTES b/sccfm-ansible/e2e/HOLD_VASA_MINUTES new file mode 100644 index 00000000..60d3b2f4 --- /dev/null +++ b/sccfm-ansible/e2e/HOLD_VASA_MINUTES @@ -0,0 +1 @@ +15 diff --git a/sccfm-ansible/e2e/access_rules/playbooks/cleanup.yml b/sccfm-ansible/e2e/access_rules/playbooks/cleanup.yml index cda11f1d..1b80c0d6 100644 --- a/sccfm-ansible/e2e/access_rules/playbooks/cleanup.yml +++ b/sccfm-ansible/e2e/access_rules/playbooks/cleanup.yml @@ -57,10 +57,8 @@ cisco.sccfm.execute_asa_cli: query: "{{ asa_test_query }}" commands: - - "configure terminal" - "no access-group {{ test_acl_name }} global" - "clear configure access-list {{ test_acl_name }}" - - "end" register: cli_result failed_when: false @@ -69,6 +67,16 @@ path: "/tmp/ci_access_group_uid" state: absent + - name: Remove saved provisioning response + ansible.builtin.file: + path: "/tmp/ci_access_group_provision.json" + state: absent + + - name: Remove saved configuration read transaction + ansible.builtin.file: + path: "/tmp/ci_access_group_read.json" + state: absent + # ── Network object cleanup ──────────────────────────────────── # Runs after ACL teardown so objects are no longer referenced. # When the CLI teardown above was skipped (device NOT_SYNCED), diff --git a/sccfm-ansible/e2e/access_rules/playbooks/list_access_groups.yml b/sccfm-ansible/e2e/access_rules/playbooks/list_access_groups.yml index e3c02ed3..38fe01a0 100644 --- a/sccfm-ansible/e2e/access_rules/playbooks/list_access_groups.yml +++ b/sccfm-ansible/e2e/access_rules/playbooks/list_access_groups.yml @@ -15,32 +15,111 @@ profile: default tasks: - - name: List access groups for CI ASA device (with retry for CDO sync) - cisco.sccfm.list_access_groups: - query: "{{ access_group_query }}" - limit: 10 - register: result - # Stop retrying when results arrive OR when a non-retryable error - # occurs (e.g. missing api_token). Only empty-result responses are - # worth retrying while CDO indexes the new access group. - until: >- - (result.access_groups | default([])) | length > 0 - or (result is failed - and 'not found' not in (result.msg | default('') | lower) - and (result.access_groups | default([])) | length == 0) - retries: 12 - delay: 10 - - - name: Assert at least one access group is returned - ansible.builtin.assert: - that: - - not result.changed - - result.access_groups | length > 0 - - result.count > 0 - fail_msg: >- - Expected at least one access group for query '{{ access_group_query }}'. - Ensure the provision_access_group phase ran successfully and CDO - has synced the device config. + - name: List access groups, printing diagnostics if none appear + block: + - name: List access groups for CI ASA device (with retry for CDO sync) + cisco.sccfm.list_access_groups: + query: "{{ access_group_query }}" + limit: 10 + register: result + # Stop retrying when results arrive OR when a non-retryable error + # occurs (e.g. missing api_token). Only empty-result responses are + # worth retrying while CDO indexes the new access group. + until: >- + (result.access_groups | default([])) | length > 0 + or (result is failed + and 'not found' not in (result.msg | default('') | lower) + and (result.access_groups | default([])) | length == 0) + retries: 12 + delay: 10 + + - name: Assert at least one access group is returned + ansible.builtin.assert: + that: + - not result.changed + - result.access_groups | length > 0 + - result.count > 0 + fail_msg: >- + Expected at least one access group for query '{{ access_group_query }}'. + Ensure the provision_access_group phase ran successfully and CDO + has synced the device config. + + # Each diagnostic tolerates its own failure so the original error is + # the one reported. + rescue: + - name: "Diagnostics: read saved provisioning response" + ansible.builtin.slurp: + src: "/tmp/ci_access_group_provision.json" + register: provision_response + failed_when: false + + - name: "Diagnostics: provisioning response (error_msg is the ASA's reply)" + ansible.builtin.debug: + msg: >- + {{ provision_response.content | b64decode | from_json + if provision_response.content is defined + else 'Not saved: provision_access_group.yml did not reach the save task.' }} + + - name: "Diagnostics: read saved configuration read transaction" + ansible.builtin.slurp: + src: "/tmp/ci_access_group_read.json" + register: read_transaction + failed_when: false + + - name: "Diagnostics: configuration read transaction (status, timings, details)" + ansible.builtin.debug: + msg: >- + {{ read_transaction.content | b64decode | from_json + if read_transaction.content is defined + else 'Not saved: provision_access_group.yml did not reach the read step.' }} + + - name: "Diagnostics: look up CI ASA devices (fake version so all are returned)" + cisco.sccfm.list_asa_not_on_version: + query: "{{ asa_test_query }}" + version: "9.99(0)" + register: devices + failed_when: false + + - name: "Diagnostics: CI ASA devices and their sync state" + ansible.builtin.debug: + msg: "{{ devices.devices | default(devices.msg | default('no output')) }}" + + - name: "Diagnostics: read ACL config from the ASA itself" + cisco.sccfm.execute_asa_cli: + query: "{{ asa_test_query }}" + commands: + - "show running-config access-group" + - "show running-config access-list {{ test_acl_name }}" + register: asa_config + failed_when: false + + - name: "Diagnostics: ACL config on the ASA" + ansible.builtin.debug: + msg: "{{ asa_config.results | default(asa_config.msg | default('no output')) }}" + + - name: "Diagnostics: list all access groups, unfiltered" + cisco.sccfm.list_access_groups: + limit: 50 + register: all_groups + failed_when: false + + - name: "Diagnostics: access groups SCCFM knows about, as [name, entity_uid]" + ansible.builtin.debug: + msg: + count: "{{ all_groups.count | default('n/a') }}" + groups: >- + {{ all_groups.access_groups | default([]) | map(attribute='name') + | zip(all_groups.access_groups | default([]) + | map(attribute='entity_uid')) | list }} + error: "{{ all_groups.msg | default('') }}" + + # The module's own error, if any, is in the "fatal:" line above. + - name: Fail with the original error + ansible.builtin.fail: + msg: >- + No access group matched '{{ access_group_query }}' after + {{ result.attempts | default(1) }} attempt(s); see the + diagnostics above. - name: Cache access group UID for subsequent plays ansible.builtin.copy: diff --git a/sccfm-ansible/e2e/access_rules/playbooks/provision_access_group.yml b/sccfm-ansible/e2e/access_rules/playbooks/provision_access_group.yml index 88b122fe..a9c6b910 100644 --- a/sccfm-ansible/e2e/access_rules/playbooks/provision_access_group.yml +++ b/sccfm-ansible/e2e/access_rules/playbooks/provision_access_group.yml @@ -15,6 +15,16 @@ group/cisco.sccfm.all: profile: default + # EXPERIMENT: API-onboarded ASAs are not on the "New ASA Policy View", so + # the access-groups API never lists their ACLs. The UI sets the flag when it + # creates the device, before the first config read, so switch the new CI + # device with a public PATCH of deviceMetadata.additional and re-read its + # config before pushing any CLI. No module wraps modify_device yet, so call + # the REST API like remove_vasa.yml. + vars: + sccfm_api_base: "https://ci.manage.security.cisco.com/api/rest" + profile_token: "{{ lookup('cisco.sccfm.profile', 'default', field='api_token') }}" + tasks: - name: Create source network object for access rule tests cisco.sccfm.create_network_object: @@ -36,18 +46,160 @@ - dst_obj is failed - "'already exists' not in (dst_obj.msg | default('') | lower)" - - name: Create test ACL and apply as global access-group - cisco.sccfm.execute_asa_cli: + # ── EXPERIMENT: switch the new CI device to the new policy view ───── + - name: "Experiment: find the CI device (fake version so it is returned)" + cisco.sccfm.list_asa_not_on_version: query: "{{ asa_test_query }}" + version: "9.99(0)" + register: ci_devices + + - name: "Experiment: remember the CI device UID" + ansible.builtin.set_fact: + ci_device_uid: "{{ ci_devices.devices[0].uid }}" + failed_when: ci_devices.devices | length != 1 + + # no_log keeps the bearer token out of the output; only selected fields + # of the registered results are printed below. + - name: "Experiment: read the CI device before the switch" + ansible.builtin.uri: + url: "{{ sccfm_api_base }}/v1/inventory/devices/{{ ci_device_uid }}" + headers: + Authorization: "Bearer {{ profile_token }}" + register: ci_device_before + no_log: true + failed_when: false + + - name: "Experiment: set deviceMetadata.additional.isNewPolicyObjectModel on the CI device" + ansible.builtin.uri: + url: "{{ sccfm_api_base }}/v1/inventory/devices/{{ ci_device_uid }}" + method: PATCH + headers: + Authorization: "Bearer {{ profile_token }}" + body_format: json + body: + deviceMetadata: + additional: + isNewPolicyObjectModel: "true" + status_code: [200, 202, 204] + register: policy_model_patch + no_log: true + failed_when: false + + - name: "Experiment: read the CI device after the switch" + ansible.builtin.uri: + url: "{{ sccfm_api_base }}/v1/inventory/devices/{{ ci_device_uid }}" + headers: + Authorization: "Bearer {{ profile_token }}" + register: ci_device_after + no_log: true + failed_when: false + + - name: "Experiment: policy view switch results" + ansible.builtin.set_fact: + policy_model_experiment: + ci_device_uid: "{{ ci_device_uid }}" + ci_device_before: "{{ ci_device_before.json.deviceMetadata | default('absent') }}" + patch: + status: "{{ policy_model_patch.status | default('n/a') }}" + response: "{{ policy_model_patch.json | default(policy_model_patch.msg | default('')) }}" + ci_device_after: "{{ ci_device_after.json.deviceMetadata | default('absent') }}" + + - name: "Experiment: show policy view switch results" + ansible.builtin.debug: + var: policy_model_experiment + + - name: "Experiment: save policy view switch results for list_access_groups diagnostics" + ansible.builtin.copy: + content: "{{ policy_model_experiment | to_nice_json }}" + dest: "/tmp/ci_access_group_policy_model.json" + mode: "0600" + + # Stop here if the PATCH did not stick: everything after it depends on the + # flag, and failing makes the results above visible in the pytest output. + - name: "Experiment: assert the CI device is now on the new policy view" + ansible.builtin.assert: + that: + - >- + (ci_device_after.json.deviceMetadata.additional.isNewPolicyObjectModel + | default('')) == 'true' + fail_msg: >- + The PATCH did not set deviceMetadata.additional.isNewPolicyObjectModel + on {{ ci_device_uid }}: {{ policy_model_experiment }} + + - name: "Experiment: read the configuration under the new policy view" + cisco.sccfm.read_asa_config: + uids: ["{{ ci_device_uid }}"] + wait: true + timeout: 90 + + # The API rejects script pushes against a NOT_SYNCED device, and the + # suites that run before this one leave the vASA re-syncing. Filtering + # on configState makes the module report "No devices found" until the + # device is SYNCED, so retry only on that. This wait plus the two reads + # must stay well under the 300s playbook timeout in conftest.py. + # + # No "configure terminal"/"end": SCCFM applies scripts containing them + # but reports "Command failed". The core services (e.g. + # asa_boot_image_service) also send bare config lines. + - name: Create test ACL and apply as global access-group (waits for SYNCED) + cisco.sccfm.execute_asa_cli: + query: "{{ asa_test_query }} AND configState:SYNCED" commands: - - "configure terminal" - "access-list {{ test_acl_name }} extended permit ip any any" - "access-group {{ test_acl_name }} global" - - "end" register: cli_result + until: >- + cli_result is not failed + or 'no devices found' not in (cli_result.msg | default('') | lower) + retries: 12 + delay: 10 - - name: Assert CLI execution succeeded + - name: Show ASA response to the provisioning commands + ansible.builtin.debug: + var: cli_result.results + + # pytest only prints a playbook's output when its phase fails, so the + # debug output above is lost when this phase passes. Save the response + # for list_access_groups.yml to print if the access group never appears. + - name: Save ASA response for list_access_groups diagnostics + ansible.builtin.copy: + content: "{{ cli_result.results | to_nice_json }}" + dest: "/tmp/ci_access_group_provision.json" + mode: "0600" + + # execute_asa_cli only fails on API/transaction errors. A rejected + # script comes back on a successful result, either as a per-device + # error_msg or as "Command failed"/"ERROR:" in the result text (the same + # check asa_boot_image_service applies). + - name: Assert the ASA accepted the provisioning commands ansible.builtin.assert: that: - - cli_result is not failed - fail_msg: "Failed to provision test access group via CLI" + - cli_result.results | length == 1 + - cli_result.results | selectattr('error_msg') | list | length == 0 + - >- + cli_result.results | map(attribute='result') | map('string') + | select('search', '(?im)^ERROR:|Command failed') | list | length == 0 + fail_msg: >- + Expected exactly one device with no CLI errors, got + {{ cli_result.results | length }} result(s): + {{ cli_result.results | items2dict(key_name='device_uid', value_name='result') }} + (error_msg: {{ cli_result.results | map(attribute='error_msg') | list }}) + + # SCCFM does not re-read the device after a CLI push: the ACL is on the + # ASA but no access group appears until the configuration is read. + - name: Read the ASA configuration into SCCFM + cisco.sccfm.read_asa_config: + uids: "{{ cli_result.results | map(attribute='device_uid') | list }}" + wait: true + timeout: 90 + register: read_result + + - name: Show the configuration read transaction + ansible.builtin.debug: + var: read_result.transaction + + - name: Save the read transaction for list_access_groups diagnostics + ansible.builtin.copy: + content: "{{ read_result.transaction | to_nice_json }}" + dest: "/tmp/ci_access_group_read.json" + mode: "0600" diff --git a/sccfm-ansible/e2e/access_rules/playbooks/vars/test_data.yml b/sccfm-ansible/e2e/access_rules/playbooks/vars/test_data.yml index 6762d6f2..ceca6f52 100644 --- a/sccfm-ansible/e2e/access_rules/playbooks/vars/test_data.yml +++ b/sccfm-ansible/e2e/access_rules/playbooks/vars/test_data.yml @@ -5,7 +5,12 @@ asa_test_query: "name:ci-e2e-asa-* AND connectivityState:ONLINE" # Access group provisioning — a freshly onboarded ASA has no ACLs, so the # suite provisions a global access-group via CLI before running CRUD phases. -test_acl_name: "ci-e2e-test-acl" +# +# The ACL and network object names below must not overlap with the CLI e2e +# suite (cisco_sccfm_cli/e2e/access_rules), which runs against the same +# tenant on its own vASA. Access group names are unique per tenant, and each +# suite's cleanup deletes its network objects by name. +test_acl_name: "ci-e2e-ansible-acl" # Access group query — the access groups API only supports: # entityUid, isShared, name, sharedAccessGroupUid, uid @@ -15,8 +20,8 @@ access_group_query: "name:{{ test_acl_name }}" # Protocol and port fields are omitted: the API requires source_port when # protocol is tcp, but the service-layer port builder does not produce valid # port objects yet. Port coverage should be added once that is fixed. -test_source_network: "ci-e2e-src-host" -test_destination_network: "ci-e2e-dst-host" +test_source_network: "ci-e2e-ansible-src-host" +test_destination_network: "ci-e2e-ansible-dst-host" test_rule: index: 1 diff --git a/sccfm-ansible/e2e/asa/playbooks/onboard_vasa.yml b/sccfm-ansible/e2e/asa/playbooks/onboard_vasa.yml index c76e7448..560daa97 100644 --- a/sccfm-ansible/e2e/asa/playbooks/onboard_vasa.yml +++ b/sccfm-ansible/e2e/asa/playbooks/onboard_vasa.yml @@ -33,6 +33,9 @@ password: "{{ vault_vasa_password }}" connector_type: "{{ default_connector_type }}" ignore_certificate: "{{ default_ignore_certificate }}" + # As the UI does; the access_rules suite needs the access-groups API + # to list this device's ACLs. + new_policy_view: true register: onboard_result - name: Show onboard result diff --git a/sccfm-ansible/e2e/run_e2e.sh b/sccfm-ansible/e2e/run_e2e.sh index 574c11cb..8d2a55d9 100755 --- a/sccfm-ansible/e2e/run_e2e.sh +++ b/sccfm-ansible/e2e/run_e2e.sh @@ -37,6 +37,25 @@ if [[ ! -f "${EXAMPLES_DIR}/group_vars/all/vault.yml" ]]; then exit 1 fi +# ── Debug hold (opt-in via HOLD_VASA_MINUTES) ───────────────────── +# Jenkins onboards the vASA before this script and always deletes it, with +# its CloudFormation stack, after the job. When the marker file exists, +# skip the tests and keep the job, and so the device, alive for that many +# minutes. Abort the job to tear down early. Keep the hold short enough +# for setup + hold + teardown to fit in the job's 1h timeout. +HOLD_FILE="${SCRIPT_DIR}/HOLD_VASA_MINUTES" +if [[ -f "${HOLD_FILE}" ]]; then + hold_minutes="$(tr -dc '0-9' < "${HOLD_FILE}")" + hold_minutes="${hold_minutes:-0}" + host="${ASA_HOST:-unknown}" + until_utc="$(date -u -d "+${hold_minutes} min" +%H:%M 2>/dev/null || echo '?')" + echo "HOLD: skipping tests. vASA ${host} is onboarded as" \ + "ci-e2e-asa-${host//[^a-zA-Z0-9]/-} and stays up for ${hold_minutes} min" \ + "(until ${until_utc} UTC)." + sleep "$(( hold_minutes * 60 ))" + exit 0 +fi + # ── Install collection ──────────────────────────────────────────── echo "Installing cisco.sccfm collection..." poetry run ansible-galaxy collection install "${COLLECTION_DIR}" --force diff --git a/sccfm-ansible/meta/runtime.yml b/sccfm-ansible/meta/runtime.yml index e194c82c..070af620 100644 --- a/sccfm-ansible/meta/runtime.yml +++ b/sccfm-ansible/meta/runtime.yml @@ -41,6 +41,7 @@ action_groups: - onboard_asa - onboard_cdfmc_ftd - onboard_cdfmc_ftd_ztp + - read_asa_config - register_cdfmc_ftd - remove_asa_shun - remove_network_group_members diff --git a/sccfm-ansible/plugins/modules/execute_asa_cli.py b/sccfm-ansible/plugins/modules/execute_asa_cli.py index 10bca204..621ef434 100644 --- a/sccfm-ansible/plugins/modules/execute_asa_cli.py +++ b/sccfm-ansible/plugins/modules/execute_asa_cli.py @@ -15,6 +15,10 @@ - See the SCC Firewall Manager API documentation for U(https://developer.cisco.com/docs/cisco-security-cloud-control-firewall-manager/get-devices/) query details. + - The module fails when SCC Firewall Manager reports a command failure on any + device, either as a per-device C(error_msg) or as C(Command failed) or + C(ERROR:) in the result. C(results) is still returned for every device; use + C(failed_when) to tolerate failures. options: query: description: @@ -170,6 +174,7 @@ AsaCommandLineService, InventoryService, SccApiError, + is_failed_cli_result, ) from cisco_sccfm_core.types import ConfigLike except ImportError as exc: @@ -285,6 +290,21 @@ def run_module() -> None: ) results_data = [result.model_dump(mode="json") for result in results] + failed_uids = [result.device_uid for result in results if is_failed_cli_result(result)] + if failed_uids: + # Commands before the failing line may have been applied, so changed=True. + module.fail_json( + changed=True, + msg=( + f"CLI commands failed on {len(failed_uids)} of {len(results)} device(s): " + f"{', '.join(failed_uids)}" + ), + command=command_script, + commands=commands, + device_count=len(device_uids), + results=results_data, + ) + module.exit_json( changed=True, msg=f"Successfully executed CLI commands on {len(device_uids)} device(s)", diff --git a/sccfm-ansible/plugins/modules/onboard_asa.py b/sccfm-ansible/plugins/modules/onboard_asa.py index c72bb484..85627f31 100644 --- a/sccfm-ansible/plugins/modules/onboard_asa.py +++ b/sccfm-ansible/plugins/modules/onboard_asa.py @@ -42,6 +42,15 @@ required: false type: bool default: false + new_policy_view: + description: + - Onboard the device on the New ASA Policy View, as the SCC Firewall + Manager UI does. + - Without it, M(cisco.sccfm.list_access_groups) does not list the + device's ACLs. + required: false + type: bool + default: false grouped_labels: description: Mapping of grouped labels to assign to the device. required: false @@ -159,6 +168,7 @@ def build_argument_spec() -> dict[str, dict[str, str | bool | list[str]]]: }, "connector_name": {"type": "str", "required": False}, "ignore_certificate": {"type": "bool", "required": False, "default": False}, + "new_policy_view": {"type": "bool", "required": False, "default": False}, "grouped_labels": {"type": "dict", "required": False}, "ungrouped_labels": {"type": "list", "elements": "str", "required": False}, **base_argument_spec(), @@ -215,7 +225,11 @@ def run_module() -> None: device={}, ) - asa_device = _onboard_asa(config, asa_create_or_update_input) + asa_device = _onboard_asa( + config, + asa_create_or_update_input, + new_policy_view=bool(params.get("new_policy_view", False)), + ) module.exit_json(changed=True, msg="Onboarded successfully", device=asa_device.to_dict()) except ApiException as e: error = SccApiError.from_exception(e) @@ -224,11 +238,14 @@ def run_module() -> None: module.fail_json(msg=f"Unexpected error: {str(e)}") -def _onboard_asa(config: ConfigLike, asa_create_or_update_input: AsaCreateOrUpdateInput) -> Device: +def _onboard_asa( + config: ConfigLike, asa_create_or_update_input: AsaCreateOrUpdateInput, *, new_policy_view: bool +) -> Device: try: asa_onboard_service = AsaOnboardService(config=config) return asa_onboard_service.onboard_asa( - asa_create_or_update_input=asa_create_or_update_input + asa_create_or_update_input=asa_create_or_update_input, + new_policy_view=new_policy_view, ) except Exception as e: raise e diff --git a/sccfm-ansible/plugins/modules/read_asa_config.py b/sccfm-ansible/plugins/modules/read_asa_config.py new file mode 100644 index 00000000..fde10f34 --- /dev/null +++ b/sccfm-ansible/plugins/modules/read_asa_config.py @@ -0,0 +1,264 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +DOCUMENTATION = r""" +--- +module: read_asa_config +short_description: Read ASA running configuration into SCC Firewall Manager +description: + - Read the running configuration of one or more ASA devices into SCC Firewall + Manager. + - Use after changing a device outside SCC Firewall Manager, for example with + M(cisco.sccfm.execute_asa_cli). SCC Firewall Manager does not re-read the + device on its own, so access groups, rules, and objects added that way do + not appear until a read. + - The read replaces SCC Firewall Manager's copy of the configuration, so + changes staged there but not yet deployed can be lost. +options: + query: + description: + - Lucene query to select ASA devices. + - Mutually exclusive with C(uids). + - The query is automatically combined with C(deviceType:ASA). + required: false + type: str + uids: + description: + - List of device UIDs to read. + - Mutually exclusive with C(query). + required: false + type: list + elements: str + limit: + description: + - Maximum number of devices to return when using C(query). + - Ignored when using C(uids). + required: false + type: int + default: 50 + offset: + description: + - Pagination offset when using C(query). + - Ignored when using C(uids). + required: false + type: int + default: 0 + wait: + description: + - Wait for the read transaction to reach a terminal status. + - When enabled, the module fails if the transaction ends in C(ERROR) + or C(CANCELLED). + required: false + type: bool + default: false + timeout: + description: + - Maximum number of seconds to wait for the read transaction when + C(wait=true). + required: false + type: int + default: 3600 + profile: + description: Named SCCFM profile configured by C(sccfm-cli configure). + required: false + type: str + default: default + config_path: + description: Optional path to the canonical SCCFM profile configuration file. + required: false + type: path +author: + - Cisco SCCFM Team (@CiscoDevNet) +""" + +EXAMPLES = r""" +# Example 1: Push config with the CLI, then read it back into SCC Firewall Manager +- name: Create an ACL on the branch ASAs + cisco.sccfm.execute_asa_cli: + query: "name:branch-*" + commands: + - "access-list branch-acl extended permit ip any any" + register: cli_result + +- name: Read the new configuration into SCC Firewall Manager + cisco.sccfm.read_asa_config: + uids: "{{ cli_result.results | map(attribute='device_uid') | list }}" + wait: true + timeout: 300 + +# Example 2: Shared profile via module_defaults, without waiting for the read +- name: Read ASA configuration + hosts: localhost + gather_facts: false + module_defaults: + group/cisco.sccfm.all: + profile: default + tasks: + - name: Start a configuration read + cisco.sccfm.read_asa_config: + uids: + - "12345678-1234-1234-1234-123456789abc" +""" + +RETURN = r""" +transaction: + description: The started or completed transaction object for the read. + returned: success (changed=True) + type: dict +device_count: + description: Number of devices included in the read. + returned: always + type: int +""" + + +from typing import Any, cast + +from ansible.module_utils.basic import AnsibleModule + +from ..module_utils.dependencies import record_import_error + +try: + from scc_firewall_manager_sdk import ApiException, CdoTransaction, DevicePage + + from cisco_sccfm_core import ASA_DEVICE_TYPE_FILTER, InventoryService, SccApiError + from cisco_sccfm_core.constants import DEFAULT_TRANSACTION_TIMEOUT_SEC + from cisco_sccfm_core.models.cdo_transaction_status import CdoTransactionStatus + from cisco_sccfm_core.services.inventory import AsaConfigReadService + from cisco_sccfm_core.services.transaction_service import TransactionService + from cisco_sccfm_core.types import ConfigLike +except ImportError as exc: + record_import_error(exc) + ApiException = RuntimeError + DEFAULT_TRANSACTION_TIMEOUT_SEC = 3600 + + +from ..module_utils.config import Config, base_argument_spec, create_config + + +def build_argument_spec() -> dict[str, dict[str, Any]]: + return { + "query": {"type": "str", "required": False}, + "uids": {"type": "list", "elements": "str", "required": False}, + "limit": {"type": "int", "required": False, "default": 50}, + "offset": {"type": "int", "required": False, "default": 0}, + "wait": {"type": "bool", "required": False, "default": False}, + "timeout": { + "type": "int", + "required": False, + "default": DEFAULT_TRANSACTION_TIMEOUT_SEC, + }, + **base_argument_spec(), + } + + +def resolve_device_uids_from_query( + config: ConfigLike, + query: str, + limit: int, + offset: int, +) -> list[str]: + """Resolve device UIDs from a Lucene query.""" + inventory_service = InventoryService(config=config) + page: DevicePage = inventory_service.get_devices( + limit=limit, + offset=offset, + query=f"({query}) AND {ASA_DEVICE_TYPE_FILTER}", + ) + return [device.uid for device in (page.items or [])] + + +def _is_failed_transaction(transaction: CdoTransaction) -> bool: + return transaction.cdo_transaction_status in { + CdoTransactionStatus.ERROR, + CdoTransactionStatus.CANCELLED, + } + + +def _wait_for_read( + module: AnsibleModule, config: ConfigLike, transaction: CdoTransaction, device_count: int +) -> CdoTransaction: + if transaction.transaction_uid is None: + module.fail_json(msg="Transaction UID missing from configuration read response.") + transaction = TransactionService(config=config).wait_for_transaction_to_finish( + transaction_uid=cast(str, transaction.transaction_uid), + timeout_sec=module.params["timeout"], + ) + if _is_failed_transaction(transaction): + module.fail_json( + msg=( + f"Configuration read transaction {transaction.transaction_uid} failed with " + f"status: {transaction.cdo_transaction_status}" + ), + device_count=device_count, + transaction=transaction.to_dict(), + ) + return transaction + + +def run_module() -> None: + module = AnsibleModule( + argument_spec=build_argument_spec(), + supports_check_mode=True, + mutually_exclusive=[["query", "uids"]], + required_one_of=[["query", "uids"]], + ) + + config: Config = create_config(module) + uids: list[str] | None = module.params.get("uids") + query: str | None = module.params.get("query") + wait_for_completion: bool = module.params["wait"] + + try: + device_uids = uids or resolve_device_uids_from_query( + config=config, + query=cast(str, query), + limit=module.params["limit"], + offset=module.params["offset"], + ) + if not device_uids: + module.fail_json(msg="No devices found matching the specified query.") + + if module.check_mode: + module.exit_json( + changed=True, + msg=f"Would read configuration from {len(device_uids)} device(s).", + device_count=len(device_uids), + transaction={}, + ) + + transaction = AsaConfigReadService(config=config).read_configuration( + device_uids=device_uids + ) + if wait_for_completion: + transaction = _wait_for_read(module, config, transaction, len(device_uids)) + + module.exit_json( + changed=True, + msg=( + f"Configuration read completed for {len(device_uids)} device(s)." + if wait_for_completion + else f"Configuration read started for {len(device_uids)} device(s)." + ), + device_count=len(device_uids), + transaction=transaction.to_dict(), + ) + + except ApiException as e: + error = SccApiError.from_exception(e) + module.fail_json(**error.to_dict()) + except TimeoutError as e: + module.fail_json(msg=str(e)) + except Exception as e: + module.fail_json(msg=f"Unexpected error: {str(e)}") + + +def main() -> None: + run_module() + + +if __name__ == "__main__": + main() diff --git a/sccfm-ansible/plugins/modules/tests/test_execute_asa_cli.py b/sccfm-ansible/plugins/modules/tests/test_execute_asa_cli.py index 573d1290..58a20545 100644 --- a/sccfm-ansible/plugins/modules/tests/test_execute_asa_cli.py +++ b/sccfm-ansible/plugins/modules/tests/test_execute_asa_cli.py @@ -361,3 +361,37 @@ def test_should_return_structured_error_on_api_exception( assert call_kwargs["error_code"] == "FORBIDDEN" assert call_kwargs["error_details"] == {"resource": "device"} assert call_kwargs["status_code"] == 403 + + +@pytest.mark.parametrize( + ("result_text", "error_msg"), + [ + ("Command failed", None), + ("access-list ok\nERROR: % Invalid input detected at '^' marker.", None), + ("", "Device unreachable"), + ], +) +@patch("plugins.modules.execute_asa_cli.Config") +@patch("plugins.modules.execute_asa_cli.AsaCommandLineService") +@patch("plugins.modules.execute_asa_cli.AnsibleModule") +def test_should_fail_when_a_device_reports_command_failure( + mock_ansible_module_class: MagicMock, + mock_cli_service_class: MagicMock, + _mock_config_class: MagicMock, + result_text: str, + error_msg: str | None, + mock_module_instance_uids: MagicMock, +) -> None: + """A completed transaction with a per-device failure must fail the task.""" + mock_ansible_module_class.return_value = mock_module_instance_uids + failing = CdoCliResult(uid="r-1", device_uid="dev-1", result=result_text, error_msg=error_msg) + mock_cli_service_class.return_value.execute_cli.return_value = [failing] + + with pytest.raises(SystemExit): + execute_asa_cli.run_module() + + mock_module_instance_uids.exit_json.assert_not_called() + call_kwargs = mock_module_instance_uids.fail_json.call_args[1] + assert call_kwargs["changed"] is True + assert call_kwargs["msg"] == "CLI commands failed on 1 of 1 device(s): dev-1" + assert call_kwargs["results"] == [failing.model_dump(mode="json")] diff --git a/sccfm-ansible/plugins/modules/tests/test_onboard_asa.py b/sccfm-ansible/plugins/modules/tests/test_onboard_asa.py index ec896a98..ba0f188f 100644 --- a/sccfm-ansible/plugins/modules/tests/test_onboard_asa.py +++ b/sccfm-ansible/plugins/modules/tests/test_onboard_asa.py @@ -252,3 +252,31 @@ def test_check_mode_should_report_would_onboard_when_device_does_not_exist( assert call_kwargs["changed"] is True assert "Would onboard" in call_kwargs["msg"] mock_onboard.onboard_asa.assert_not_called() + + +@pytest.mark.parametrize("new_policy_view", [True, False]) +@patch("plugins.modules.onboard_asa.Config") +@patch("plugins.modules.onboard_asa.AsaOnboardService") +@patch("plugins.modules.onboard_asa.InventoryService") +@patch("plugins.modules.onboard_asa.AnsibleModule") +def test_should_pass_new_policy_view_to_onboard_service( + mock_ansible_module_class: MagicMock, + mock_inventory_service_class: MagicMock, + mock_asa_onboard_service_class: MagicMock, + _mock_config_class: MagicMock, + new_policy_view: bool, + mock_module_instance: MagicMock, + sample_device: Device, +) -> None: + mock_module_instance.params["new_policy_view"] = new_policy_view + mock_ansible_module_class.return_value = mock_module_instance + mock_inventory_service_class.return_value.get_devices.return_value = DevicePage( + count=0, limit=1, offset=0, items=[] + ) + mock_onboard = mock_asa_onboard_service_class.return_value + mock_onboard.onboard_asa.return_value = sample_device + + with pytest.raises(SystemExit): + onboard_asa.run_module() + + assert mock_onboard.onboard_asa.call_args.kwargs["new_policy_view"] is new_policy_view diff --git a/sccfm-ansible/plugins/modules/tests/test_read_asa_config.py b/sccfm-ansible/plugins/modules/tests/test_read_asa_config.py new file mode 100644 index 00000000..e7e963ba --- /dev/null +++ b/sccfm-ansible/plugins/modules/tests/test_read_asa_config.py @@ -0,0 +1,214 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from plugins.modules import read_asa_config # noqa: E402 +from scc_firewall_manager_sdk import CdoTransaction, Device, DevicePage, EntityType + + +def _make_transaction(status: str = "PENDING") -> CdoTransaction: + return CdoTransaction(transactionUid="txn-001", cdoTransactionStatus=status) + + +@pytest.fixture +def base_params() -> dict[str, Any]: + return { + "query": None, + "uids": ["uid-1"], + "limit": 50, + "offset": 0, + "wait": False, + "timeout": 3600, + "profile": "default", + } + + +@pytest.fixture +def mock_module(base_params: dict[str, Any]) -> MagicMock: + m = MagicMock() + m.params = base_params.copy() + m.check_mode = False + m.exit_json.side_effect = SystemExit(0) + m.fail_json.side_effect = SystemExit(1) + return m + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_start_read_for_given_uids( + mock_ansible_cls: MagicMock, + mock_read_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_ansible_cls.return_value = mock_module + mock_read_cls.return_value.read_configuration.return_value = _make_transaction() + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + mock_read_cls.return_value.read_configuration.assert_called_once_with(device_uids=["uid-1"]) + kw = mock_module.exit_json.call_args[1] + assert kw["changed"] is True + assert kw["device_count"] == 1 + assert "started" in kw["msg"] + assert kw["transaction"]["transactionUid"] == "txn-001" + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.InventoryService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_resolve_devices_from_query( + mock_ansible_cls: MagicMock, + mock_inv_cls: MagicMock, + mock_read_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_module.params.update({"uids": None, "query": "name:branch-*"}) + mock_ansible_cls.return_value = mock_module + devices = [Device(uid=f"uid-{i}", name=f"asa-{i}", deviceType=EntityType.ASA) for i in (1, 2)] + mock_inv_cls.return_value.get_devices.return_value = DevicePage(count=2, items=devices) + mock_read_cls.return_value.read_configuration.return_value = _make_transaction() + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + query = mock_inv_cls.return_value.get_devices.call_args.kwargs["query"] + assert query.startswith("(name:branch-*) AND ") + mock_read_cls.return_value.read_configuration.assert_called_once_with( + device_uids=["uid-1", "uid-2"] + ) + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.TransactionService") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_wait_for_read_completion( + mock_ansible_cls: MagicMock, + mock_read_cls: MagicMock, + mock_txn_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_module.params.update({"wait": True, "timeout": 120}) + mock_ansible_cls.return_value = mock_module + mock_read_cls.return_value.read_configuration.return_value = _make_transaction() + mock_txn_cls.return_value.wait_for_transaction_to_finish.return_value = _make_transaction( + "DONE" + ) + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + mock_txn_cls.return_value.wait_for_transaction_to_finish.assert_called_once_with( + transaction_uid="txn-001", timeout_sec=120 + ) + kw = mock_module.exit_json.call_args[1] + assert "completed" in kw["msg"] + assert kw["transaction"]["cdoTransactionStatus"] == "DONE" + + +@pytest.mark.parametrize("final_status", ["ERROR", "CANCELLED"]) +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.TransactionService") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_fail_when_waited_read_does_not_finish( + mock_ansible_cls: MagicMock, + mock_read_cls: MagicMock, + mock_txn_cls: MagicMock, + _mock_cfg: MagicMock, + final_status: str, + mock_module: MagicMock, +) -> None: + mock_module.params["wait"] = True + mock_ansible_cls.return_value = mock_module + mock_read_cls.return_value.read_configuration.return_value = _make_transaction() + mock_txn_cls.return_value.wait_for_transaction_to_finish.return_value = _make_transaction( + final_status + ) + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + mock_module.exit_json.assert_not_called() + kw = mock_module.fail_json.call_args[1] + assert final_status in kw["msg"] + assert kw["transaction"]["cdoTransactionStatus"] == final_status + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.TransactionService") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_fail_when_wait_times_out( + mock_ansible_cls: MagicMock, + mock_read_cls: MagicMock, + mock_txn_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_module.params["wait"] = True + mock_ansible_cls.return_value = mock_module + mock_read_cls.return_value.read_configuration.return_value = _make_transaction() + mock_txn_cls.return_value.wait_for_transaction_to_finish.side_effect = TimeoutError( + "Transaction txn-001 did not complete within 3600 seconds" + ) + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + assert "did not complete" in mock_module.fail_json.call_args[1]["msg"] + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_check_mode_returns_would_read_without_reading( + mock_ansible_cls: MagicMock, + mock_read_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_module.check_mode = True + mock_ansible_cls.return_value = mock_module + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + mock_read_cls.return_value.read_configuration.assert_not_called() + kw = mock_module.exit_json.call_args[1] + assert kw["changed"] is True + assert "Would read" in kw["msg"] + + +@patch("plugins.modules.read_asa_config.Config") +@patch("plugins.modules.read_asa_config.AsaConfigReadService") +@patch("plugins.modules.read_asa_config.InventoryService") +@patch("plugins.modules.read_asa_config.AnsibleModule") +def test_should_fail_when_query_returns_no_devices( + mock_ansible_cls: MagicMock, + mock_inv_cls: MagicMock, + mock_read_cls: MagicMock, + _mock_cfg: MagicMock, + mock_module: MagicMock, +) -> None: + mock_module.params.update({"uids": None, "query": "name:missing-*"}) + mock_ansible_cls.return_value = mock_module + mock_inv_cls.return_value.get_devices.return_value = DevicePage(count=0, items=[]) + + with pytest.raises(SystemExit): + read_asa_config.run_module() + + mock_read_cls.return_value.read_configuration.assert_not_called() + assert "No devices found" in mock_module.fail_json.call_args[1]["msg"]