From a6b1bac40ff0539c190edec9d708aca4e4d9a1fc Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Sat, 18 Jul 2026 14:13:32 +0200 Subject: [PATCH 1/4] Classify SSL errors before connection errors in netbox sync check In Sync._check_netbox_connectivity the SSLError handler was unreachable: requests.exceptions.SSLError is a subclass of ConnectionError, so SSL failures in the reachability stage were reported as "Error: Connection refused". Catch SSLError first so certificate problems are reported as "Error: SSL error", matching the classification of the authentication stage. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/commands/netbox.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/osism/commands/netbox.py b/osism/commands/netbox.py index 42fa1f4e4..b8bc20c78 100644 --- a/osism/commands/netbox.py +++ b/osism/commands/netbox.py @@ -235,12 +235,12 @@ def _check_netbox_connectivity(self, nb, url, token, ignore_ssl_errors, timeout= base_url = nb.base_url # Make simple GET request without auth to test reachability requests.get(base_url, timeout=timeout, verify=not ignore_ssl_errors) + except requests.exceptions.SSLError: + return "Error: SSL error" except requests.exceptions.Timeout: return "Error: Timeout" except requests.exceptions.ConnectionError: return "Error: Connection refused" - except requests.exceptions.SSLError: - return "Error: SSL error" except Exception as e: error_msg = str(e) short_msg = error_msg[:50] if len(error_msg) > 50 else error_msg From ff6d444d8ed5027880ce933dab148b9ff12d5af9 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Thu, 23 Jul 2026 11:24:21 +0200 Subject: [PATCH 2/4] Classify SSL errors before connection errors in netbox instance check The string-based error classifiers in Sync._check_netbox_instance and in the authentication stage of Sync._check_netbox_connectivity tested for "connection" before "ssl"/"certificate". An SSL failure raised by requests through nb.status() stringifies as "HTTPSConnectionPool(host='...', port=443): Max retries exceeded ... (Caused by SSLError(...))", which contains "connection", so certificate problems were reported as "Error: Connection refused". Check for SSL markers first so they are reported as "Error: SSL error", matching the classification of the reachability stage. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/commands/netbox.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/osism/commands/netbox.py b/osism/commands/netbox.py index b8bc20c78..b878c8ceb 100644 --- a/osism/commands/netbox.py +++ b/osism/commands/netbox.py @@ -198,10 +198,10 @@ def _check_netbox_instance(self, nb, timeout=20): or "unauthorized" in error_msg.lower() ): return "Error: Auth failed" - elif "connection" in error_msg.lower() or "refused" in error_msg.lower(): - return "Error: Connection refused" elif "ssl" in error_msg.lower() or "certificate" in error_msg.lower(): return "Error: SSL error" + elif "connection" in error_msg.lower() or "refused" in error_msg.lower(): + return "Error: Connection refused" else: # Truncate long error messages short_msg = error_msg[:50] if len(error_msg) > 50 else error_msg @@ -268,10 +268,10 @@ def _check_netbox_connectivity(self, nb, url, token, ignore_ssl_errors, timeout= or "unauthorized" in error_msg.lower() ): return "Error: Auth failed" - elif "connection" in error_msg.lower() or "refused" in error_msg.lower(): - return "Error: Connection refused" elif "ssl" in error_msg.lower() or "certificate" in error_msg.lower(): return "Error: SSL error" + elif "connection" in error_msg.lower() or "refused" in error_msg.lower(): + return "Error: Connection refused" else: # Truncate long error messages short_msg = error_msg[:50] if len(error_msg) > 50 else error_msg From d27216a3a0dd3bafca4b4c92623942c26b2db290 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Thu, 23 Jul 2026 11:24:32 +0200 Subject: [PATCH 3/4] Return failure when SONiC config reload fails Reload.take_action returned 0 when "sudo config reload -y" failed on the device: it skipped the config save, logged a success summary, and exited cleanly, so callers saw a success exit code although the operation the command is named for had failed. A load failure in the same flow already exits 1. Return 1 on reload failure instead, keeping the warning that the config save is skipped. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- osism/commands/sonic.py | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/osism/commands/sonic.py b/osism/commands/sonic.py index 551e8fff4..9902b790a 100644 --- a/osism/commands/sonic.py +++ b/osism/commands/sonic.py @@ -652,26 +652,20 @@ def take_action(self, parsed_args): return 1 # Reload configuration - reload_successful = self._reload_configuration(ssh) - - # Save configuration only if reload was successful - if reload_successful: - if not self._save_configuration(ssh): - return 1 - else: + if not self._reload_configuration(ssh): logger.warning("Skipping config save due to reload failure") + return 1 + + # Save configuration + if not self._save_configuration(ssh): + return 1 # Cleanup self._cleanup_temp_file(ssh, switch_config_file) logger.info("SONiC configuration reload completed successfully") logger.info(f"- Config context saved locally to: {config_context_file}") - if reload_successful: - logger.info("- Configuration loaded, reloaded, and saved on switch") - else: - logger.info( - "- Configuration loaded on switch (save skipped due to reload failure)" - ) + logger.info("- Configuration loaded, reloaded, and saved on switch") logger.info(f"- Backup created on switch: {backup_filename}") return 0 From 20ee355dfc8dffe6f43b5ff58ad845a94c3b98da Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Sat, 18 Jul 2026 14:13:38 +0200 Subject: [PATCH 4/4] Add unit tests for netbox and sonic command modules Cover the network-facing CLI command modules with unit tests: - tests/unit/commands/test_netbox.py: extend the existing exit-code tests with the Sync error-classification helpers (_check_netbox_instance, _check_netbox_connectivity), table building (_build_netbox_table), the --list/--check/--no-wait/--filter paths of Sync.take_action, argument assembly in Ironic and Manage, Versions, the nbcli bootstrap and argument quoting in Console, and the device lookup fallback chain and report formatting in Dump. - tests/unit/commands/test_sonic.py: SonicCommandBase helpers (device lookup, config context filtering, SSH connection details, backup filename probing, the exec-command helpers) and the take_action control flow of Load, Backup, Ztp, Reload, Reboot, Reset, Show, Console, and Dump. - tests/unit/commands/test_sonic_validate.py: source collection and config acquisition of Validate (_collect_sources, _configs_from_export_dir, _config_from_netbox, _config_from_generate), its exit-code contract and report formatting, and the provision-state derivation and sorting of List. Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- tests/unit/commands/test_netbox.py | 810 +++++++++++++++++++- tests/unit/commands/test_sonic.py | 823 +++++++++++++++++++++ tests/unit/commands/test_sonic_validate.py | 436 +++++++++++ 3 files changed, 2054 insertions(+), 15 deletions(-) create mode 100644 tests/unit/commands/test_sonic.py create mode 100644 tests/unit/commands/test_sonic_validate.py diff --git a/tests/unit/commands/test_netbox.py b/tests/unit/commands/test_netbox.py index 1a1d748c7..45ab21011 100644 --- a/tests/unit/commands/test_netbox.py +++ b/tests/unit/commands/test_netbox.py @@ -2,16 +2,62 @@ """Tests for the ``osism netbox`` commands. -These focus on the exit-code contract: a command must return a non-zero exit -status when it gives up waiting for a task (a timeout is an operational -failure), rather than falling through to an implicit success. +Covers the exit-code contract (a command must return non-zero when it gives +up waiting for a task or cannot reach NetBox), the pure decision logic in +``Sync`` (error classification, table building), the argument assembly in +``Ironic``/``Manage``, and the device lookup and report formatting in +``Dump``. """ -from unittest.mock import MagicMock, patch +import subprocess +from types import SimpleNamespace +from unittest.mock import MagicMock, call, mock_open, patch + +import pytest +import requests from osism.commands import netbox +class _FakeSession: + """Minimal stand-in for a requests session with a timeout attribute.""" + + def __init__(self, timeout=5): + self.timeout = timeout + + +class _BareNb: + """NetBox API stand-in without an http_session attribute.""" + + def __init__(self, exc=None): + self._exc = exc + + def status(self): + if self._exc: + raise self._exc + return {"netbox-version": "4.0"} + + +class _FakeDevice: + """NetBox device record stand-in; absent attributes stay absent.""" + + def __init__(self, name, **attrs): + self.name = name + for key, value in attrs.items(): + setattr(self, key, value) + + +def _make_sync(): + return netbox.Sync(MagicMock(), MagicMock()) + + +def _make_dump(): + return netbox.Dump(MagicMock(), MagicMock()) + + +# --- Ironic.take_action --- + + def test_ironic_returns_nonzero_on_task_timeout(): cmd = netbox.Ironic(MagicMock(), MagicMock()) parsed_args = cmd.get_parser("test").parse_args([]) @@ -27,6 +73,272 @@ def test_ironic_returns_nonzero_on_task_timeout(): assert result == 1 +def test_ironic_forwards_arguments_and_returns_fetch_result(): + cmd = netbox.Ironic(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args( + [ + "node1", + "--adopt", + "--skip-kernel-param", + "a", + "--extra-kernel-param", + "b=1", + "--task-timeout", + "60", + ] + ) + + task = MagicMock() + task.id = "task-id" + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_ironic.delay", return_value=task + ) as mock_delay, patch( + "osism.commands.netbox.utils.fetch_task_output", return_value=0 + ) as mock_fetch: + result = cmd.take_action(parsed_args) + + assert result == 0 + mock_delay.assert_called_once_with( + node_name="node1", + adopt=True, + force=False, + dry_run=False, + skip_kernel_params=["a"], + extra_kernel_params=["b=1"], + ) + mock_fetch.assert_called_once_with("task-id", timeout=60) + + +def test_ironic_no_wait_returns_none_without_fetch(): + cmd = netbox.Ironic(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["--no-wait"]) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_ironic.delay", return_value=MagicMock() + ), patch("osism.commands.netbox.utils.fetch_task_output") as mock_fetch: + result = cmd.take_action(parsed_args) + + assert result is None + mock_fetch.assert_not_called() + + +# --- Sync._check_netbox_instance --- + + +def test_check_netbox_instance_not_configured(): + assert _make_sync()._check_netbox_instance(None) == "Error: Not configured" + + +def test_check_netbox_instance_success_sets_and_restores_timeout(): + cmd = _make_sync() + nb = MagicMock() + nb.http_session = _FakeSession(timeout=5) + seen = {} + nb.status.side_effect = lambda: seen.setdefault("timeout", nb.http_session.timeout) + + assert cmd._check_netbox_instance(nb, timeout=42) == "Success" + assert seen["timeout"] == 42 + assert nb.http_session.timeout == 5 + + +@pytest.mark.parametrize( + "message,expected", + [ + ("connection timed out", "Error: Timeout"), + ("401 Client Error", "Error: Auth failed"), + ("Unauthorized", "Error: Auth failed"), + ("connection refused", "Error: Connection refused"), + ("certificate verify failed", "Error: SSL error"), + ("SSL handshake failed", "Error: SSL error"), + ( + "HTTPSConnectionPool(host='nb', port=443): " + "Max retries exceeded (Caused by SSLError(...))", + "Error: SSL error", + ), + ], +) +def test_check_netbox_instance_error_classification(message, expected): + cmd = _make_sync() + nb = MagicMock() + nb.http_session = _FakeSession() + nb.status.side_effect = Exception(message) + + assert cmd._check_netbox_instance(nb) == expected + + +def test_check_netbox_instance_truncates_long_error_message(): + cmd = _make_sync() + nb = MagicMock() + nb.http_session = _FakeSession() + nb.status.side_effect = Exception("x" * 60) + + assert cmd._check_netbox_instance(nb) == f"Error: {'x' * 50}" + + +def test_check_netbox_instance_without_http_session_success(): + assert _make_sync()._check_netbox_instance(_BareNb()) == "Success" + + +def test_check_netbox_instance_without_http_session_classifies_error(): + result = _make_sync()._check_netbox_instance(_BareNb(Exception("boom"))) + assert result == "Error: boom" + + +# --- Sync._check_netbox_connectivity --- + + +def test_check_netbox_connectivity_not_configured(): + cmd = _make_sync() + with patch("osism.commands.netbox.requests.get") as mock_get: + result = cmd._check_netbox_connectivity(None, "url", "token", False) + + assert result == "Error: Not configured" + mock_get.assert_not_called() + + +@pytest.mark.parametrize( + "exc,expected", + [ + (requests.exceptions.Timeout(), "Error: Timeout"), + (requests.exceptions.ConnectionError(), "Error: Connection refused"), + (requests.exceptions.SSLError(), "Error: SSL error"), + ], + ids=["timeout", "connection", "ssl"], +) +def test_check_netbox_connectivity_stage1_errors(exc, expected): + cmd = _make_sync() + nb = MagicMock() + nb.base_url = "https://netbox.example/api" + + with patch("osism.commands.netbox.requests.get", side_effect=exc): + result = cmd._check_netbox_connectivity(nb, "url", "token", False) + + assert result == expected + nb.status.assert_not_called() + + +def test_check_netbox_connectivity_stage1_generic_error_truncated(): + cmd = _make_sync() + nb = MagicMock() + nb.base_url = "https://netbox.example/api" + + with patch("osism.commands.netbox.requests.get", side_effect=Exception("y" * 70)): + result = cmd._check_netbox_connectivity(nb, "url", "token", False) + + assert result == f"Error: {'y' * 50}" + nb.status.assert_not_called() + + +@pytest.mark.parametrize("ignore_ssl_errors", [True, False]) +def test_check_netbox_connectivity_success_passes_verify(ignore_ssl_errors): + cmd = _make_sync() + nb = MagicMock() + nb.base_url = "https://netbox.example/api" + nb.http_session = _FakeSession() + + with patch("osism.commands.netbox.requests.get") as mock_get: + result = cmd._check_netbox_connectivity( + nb, "url", "token", ignore_ssl_errors, timeout=7 + ) + + assert result == "Success" + mock_get.assert_called_once_with( + "https://netbox.example/api", timeout=7, verify=not ignore_ssl_errors + ) + + +@pytest.mark.parametrize( + "message,expected", + [ + ("Request timed out", "Error: Timeout"), + ("401 Unauthorized", "Error: Auth failed"), + ], +) +def test_check_netbox_connectivity_stage2_classification(message, expected): + cmd = _make_sync() + nb = MagicMock() + nb.base_url = "https://netbox.example/api" + nb.http_session = _FakeSession() + nb.status.side_effect = Exception(message) + + with patch("osism.commands.netbox.requests.get"): + result = cmd._check_netbox_connectivity(nb, "url", "token", False) + + assert result == expected + + +# --- Sync._build_netbox_table --- + + +def test_build_netbox_table_empty(monkeypatch): + cmd = _make_sync() + monkeypatch.setattr(netbox.settings, "NETBOX_URL", None) + + with patch.dict("osism.utils.__dict__", {"secondary_nb_list": []}): + table, headers = cmd._build_netbox_table() + + assert table == [] + assert headers == ["Name", "URL", "Site"] + + +def test_build_netbox_table_primary_row_first(monkeypatch): + cmd = _make_sync() + monkeypatch.setattr(netbox.settings, "NETBOX_URL", "https://primary.example") + secondary = SimpleNamespace( + base_url="https://sec.example", netbox_name="sec1", netbox_site="site1" + ) + + with patch.dict("osism.utils.__dict__", {"secondary_nb_list": [secondary]}): + table, headers = cmd._build_netbox_table() + + assert headers == ["Name", "URL", "Site"] + assert table[0] == ["primary", "https://primary.example", "N/A"] + assert table[1] == ["sec1", "https://sec.example", "site1"] + + +def test_build_netbox_table_secondary_without_name_and_site(monkeypatch): + cmd = _make_sync() + monkeypatch.setattr(netbox.settings, "NETBOX_URL", None) + + class _UrlOnly: + base_url = "https://sec.example" + + with patch.dict("osism.utils.__dict__", {"secondary_nb_list": [_UrlOnly()]}): + table, _ = cmd._build_netbox_table() + + assert table == [["N/A", "https://sec.example", "N/A"]] + + +def test_build_netbox_table_check_connectivity(monkeypatch): + cmd = _make_sync() + monkeypatch.setattr(netbox.settings, "NETBOX_URL", "https://primary.example") + monkeypatch.setattr(netbox.settings, "NETBOX_TOKEN", "token123") + monkeypatch.setattr(netbox.settings, "IGNORE_SSL_ERRORS", True) + cmd._check_netbox_connectivity = MagicMock(return_value="Success") + cmd._check_netbox_instance = MagicMock(return_value="Error: Timeout") + + fake_nb = MagicMock() + secondary = SimpleNamespace( + base_url="https://sec.example", netbox_name="sec1", netbox_site="site1" + ) + + with patch.dict( + "osism.utils.__dict__", {"nb": fake_nb, "secondary_nb_list": [secondary]} + ): + table, headers = cmd._build_netbox_table(check_connectivity=True, timeout=9) + + assert headers == ["Name", "URL", "Site", "Status"] + cmd._check_netbox_connectivity.assert_called_once_with( + fake_nb, "https://primary.example", "token123", True, 9 + ) + cmd._check_netbox_instance.assert_called_once_with(secondary, 9) + assert table[0][-1] == "Success" + assert table[1][-1] == "Error: Timeout" + + +# --- Sync.take_action --- + + def test_sync_returns_nonzero_on_task_timeout(): cmd = netbox.Sync(MagicMock(), MagicMock()) parsed_args = cmd.get_parser("test").parse_args([]) @@ -42,8 +354,299 @@ def test_sync_returns_nonzero_on_task_timeout(): assert result == 1 +def test_sync_list_prints_table_without_scheduling(capsys): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["--list"]) + cmd._build_netbox_table = MagicMock( + return_value=([["primary", "https://x", "N/A"]], ["Name", "URL", "Site"]) + ) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_netbox.delay" + ) as mock_delay: + result = cmd.take_action(parsed_args) + + assert result is None + mock_delay.assert_not_called() + cmd._build_netbox_table.assert_called_once_with(check_connectivity=False) + out = capsys.readouterr().out + assert "primary" in out + assert "https://x" in out + + +def test_sync_list_empty_warns_and_prints_nothing(capsys, loguru_logs): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["--list"]) + cmd._build_netbox_table = MagicMock(return_value=([], ["Name", "URL", "Site"])) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"): + result = cmd.take_action(parsed_args) + + assert result is None + assert capsys.readouterr().out == "" + assert any( + "No NetBox instances configured" in record["message"] for record in loguru_logs + ) + + +def test_sync_check_uses_connectivity_table(capsys): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["--check"]) + cmd._build_netbox_table = MagicMock( + return_value=( + [["primary", "https://x", "N/A", "Success"]], + ["Name", "URL", "Site", "Status"], + ) + ) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"): + result = cmd.take_action(parsed_args) + + assert result is None + cmd._build_netbox_table.assert_called_once_with(check_connectivity=True, timeout=20) + assert "Success" in capsys.readouterr().out + + +def test_sync_no_wait_schedules_without_fetch(): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["--no-wait"]) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_netbox.delay", return_value=MagicMock() + ) as mock_delay, patch( + "osism.commands.netbox.utils.fetch_task_output" + ) as mock_fetch: + result = cmd.take_action(parsed_args) + + assert result is None + mock_delay.assert_called_once_with(node_name=None, netbox_filter=None) + mock_fetch.assert_not_called() + + +def test_sync_forwards_filter(): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["--filter", "foo", "--no-wait"]) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_netbox.delay", return_value=MagicMock() + ) as mock_delay: + cmd.take_action(parsed_args) + + mock_delay.assert_called_once_with(node_name=None, netbox_filter="foo") + + +def test_sync_wait_returns_fetch_result(): + cmd = _make_sync() + parsed_args = cmd.get_parser("test").parse_args(["node1", "--task-timeout", "60"]) + + task = MagicMock() + task.id = "task-id" + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.conductor.sync_netbox.delay", return_value=task + ) as mock_delay, patch( + "osism.commands.netbox.utils.fetch_task_output", return_value=0 + ) as mock_fetch: + result = cmd.take_action(parsed_args) + + assert result == 0 + mock_delay.assert_called_once_with(node_name="node1", netbox_filter=None) + mock_fetch.assert_called_once_with("task-id", timeout=60) + + +# --- Manage.take_action --- + + +def _run_manage(argv): + cmd = netbox.Manage(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(argv) + + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.netbox.manage" + ) as mock_manage, patch( + "osism.tasks.handle_task", return_value=0 + ) as mock_handle_task: + result = cmd.take_action(parsed_args) + + return result, mock_manage, mock_handle_task + + +def test_manage_default_arguments(): + result, mock_manage, mock_handle_task = _run_manage([]) + + mock_manage.si.assert_called_once_with( + "run", "--wait", "--no-skipdtl", "--no-skipmtl", "--no-skipres" + ) + task = mock_manage.si.return_value.apply_async.return_value + mock_handle_task.assert_called_once_with(task, True, format="script", timeout=3600) + assert result == 0 + + +def test_manage_no_netbox_wait(): + _, mock_manage, _ = _run_manage(["--no-netbox-wait"]) + + mock_manage.si.assert_called_once_with( + "run", "--no-wait", "--no-skipdtl", "--no-skipmtl", "--no-skipres" + ) + + +def test_manage_parallel_and_limit(): + _, mock_manage, _ = _run_manage(["--parallel", "4", "--limit", "foo"]) + + mock_manage.si.assert_called_once_with( + "run", + "--wait", + "--parallel", + "4", + "--limit", + "foo", + "--no-skipdtl", + "--no-skipmtl", + "--no-skipres", + ) + + +def test_manage_skip_flags(): + _, mock_manage, _ = _run_manage(["--skipdtl", "--skipmtl", "--skipres"]) + + mock_manage.si.assert_called_once_with( + "run", "--wait", "--skipdtl", "--skipmtl", "--skipres" + ) + + +def test_manage_no_wait_passed_to_handle_task(): + _, mock_manage, mock_handle_task = _run_manage(["--no-wait"]) + + task = mock_manage.si.return_value.apply_async.return_value + mock_handle_task.assert_called_once_with(task, False, format="script", timeout=3600) + + +# --- Versions.take_action --- + + +def test_versions_prints_task_result(capsys): + cmd = netbox.Versions(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args([]) + + task = MagicMock() + task.get.return_value = "netbox 4.1" + with patch("osism.commands.netbox.utils.check_task_lock_and_exit"), patch( + "osism.tasks.netbox.ping" + ) as mock_ping: + mock_ping.delay.return_value = task + result = cmd.take_action(parsed_args) + + assert result is None + mock_ping.delay.assert_called_once_with() + task.wait.assert_called_once_with(timeout=None, interval=0.5) + assert "netbox 4.1" in capsys.readouterr().out + + +# --- Console.take_action --- + + +def test_console_returns_nonzero_when_netbox_not_configured(): + cmd = netbox.Console(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["info"]) + + with patch("osism.commands.netbox.os.path.exists", return_value=False), patch( + "osism.commands.netbox.os.mkdir" + ), patch("osism.commands.netbox.os.environ.get", return_value=None), patch( + "builtins.open", side_effect=FileNotFoundError + ): + result = cmd.take_action(parsed_args) + + assert result == 1 + + +def test_console_existing_config_runs_nbcli_directly(): + cmd = netbox.Console(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["info", "device"]) + + with patch("osism.commands.netbox.os.path.exists", return_value=True), patch( + "osism.commands.netbox.os.path.expanduser", return_value="/fakehome" + ), patch("osism.commands.netbox.os.remove") as mock_remove, patch( + "osism.commands.netbox.subprocess.call" + ) as mock_call: + result = cmd.take_action(parsed_args) + + assert result is None + mock_remove.assert_not_called() + mock_call.assert_called_once_with("/usr/local/bin/nbcli info device", shell=True) + + +def test_console_initializes_config_when_missing(): + cmd = netbox.Console(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["shell"]) + + m_open = mock_open(read_data="secrettoken\n") + with patch("osism.commands.netbox.os.path.exists", return_value=False), patch( + "osism.commands.netbox.os.path.expanduser", return_value="/fakehome" + ), patch("osism.commands.netbox.os.mkdir") as mock_mkdir, patch( + "osism.commands.netbox.os.environ.get", return_value="https://netbox.example" + ), patch( + "builtins.open", m_open + ), patch( + "osism.commands.netbox.os.remove" + ) as mock_remove, patch( + "osism.commands.netbox.subprocess.call" + ) as mock_call, patch( + "osism.commands.netbox.yaml.dump" + ) as mock_yaml_dump: + result = cmd.take_action(parsed_args) + + assert result is None + mock_mkdir.assert_called_once_with("/fakehome/.nbcli") + assert m_open.call_args_list[0] == call("/run/secrets/NETBOX_TOKEN", "r") + assert mock_call.call_args_list[0] == call( + ["/usr/local/bin/nbcli", "init"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + mock_remove.assert_called_once_with("/fakehome/.nbcli/user_config.yml") + assert m_open.call_args_list[1] == call("/fakehome/.nbcli/user_config.yml", "w") + config = mock_yaml_dump.call_args[0][0] + assert config["pynetbox"] == { + "url": "https://netbox.example", + "token": "secrettoken", + } + assert config["requests"] == {"verify": False} + assert config["nbcli"] == {"filter_limit": 50} + assert mock_call.call_args_list[-1] == call( + "/usr/local/bin/nbcli shell ", shell=True + ) + + +def test_console_quotes_arguments_with_spaces(): + cmd = netbox.Console(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["filter", "name foo", "bar"]) + + with patch("osism.commands.netbox.os.path.exists", return_value=True), patch( + "osism.commands.netbox.os.path.expanduser", return_value="/fakehome" + ), patch("osism.commands.netbox.subprocess.call") as mock_call: + cmd.take_action(parsed_args) + + mock_call.assert_called_once_with( + "/usr/local/bin/nbcli filter 'name foo' bar", shell=True + ) + + +# --- Dump.take_action --- + + +def _filter_dispatch(mapping): + """Route ``devices.filter(...)`` calls by the filter keyword used.""" + + def _filter(**kwargs): + for key, value in mapping.items(): + if key in kwargs: + return value + return [] + + return _filter + + def test_dump_returns_nonzero_when_netbox_not_configured(): - cmd = netbox.Dump(MagicMock(), MagicMock()) + cmd = _make_dump() parsed_args = cmd.get_parser("test").parse_args(["somehost"]) with patch.dict("osism.utils.__dict__", {"nb": None}): @@ -53,7 +656,7 @@ def test_dump_returns_nonzero_when_netbox_not_configured(): def test_dump_returns_nonzero_when_device_not_found(): - cmd = netbox.Dump(MagicMock(), MagicMock()) + cmd = _make_dump() parsed_args = cmd.get_parser("test").parse_args(["somehost"]) fake_nb = MagicMock() @@ -65,15 +668,192 @@ def test_dump_returns_nonzero_when_device_not_found(): assert result == 1 -def test_console_returns_nonzero_when_netbox_not_configured(): - cmd = netbox.Console(MagicMock(), MagicMock()) - parsed_args = cmd.get_parser("test").parse_args(["info"]) +def test_dump_selects_exact_name_match(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) - with patch("osism.commands.netbox.os.path.exists", return_value=False), patch( - "osism.commands.netbox.os.mkdir" - ), patch("osism.commands.netbox.os.environ.get", return_value=None), patch( - "builtins.open", side_effect=FileNotFoundError - ): + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [ + _FakeDevice("sw1-old"), + _FakeDevice("sw1"), + ] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): result = cmd.take_action(parsed_args) - assert result == 1 + assert result is None + fake_nb.dcim.devices.filter.assert_called_once_with(name="sw1") + assert "sw1" in capsys.readouterr().out + + +def test_dump_falls_back_through_custom_field_lookups(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["ext-host"]) + + # The alternative_name stage returns a device whose custom field does not + # match exactly, so the lookup must continue to inventory_hostname. + wrong = _FakeDevice("other", custom_fields={"alternative_name": "different"}) + right = _FakeDevice("sw2", custom_fields={"inventory_hostname": "ext-host"}) + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.side_effect = _filter_dispatch( + { + "name": [], + "cf_alternative_name": [wrong], + "cf_inventory_hostname": [right], + } + ) + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + result = cmd.take_action(parsed_args) + + assert result is None + assert fake_nb.dcim.devices.filter.call_count == 3 + out = capsys.readouterr().out + assert "sw2" in out + assert "Inventory Hostname" in out + assert "ext-host" in out + + +def test_dump_prints_device_details(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + device = _FakeDevice( + "sw1", + device_type="Accton AS7326", + role="leaf", + site="site1", + status="Active", + oob_ip=SimpleNamespace(address="192.0.2.1/24"), + primary_ip4=SimpleNamespace(address="192.0.2.2/24"), + primary_ip6=SimpleNamespace(address="2001:db8::1/64"), + ) + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [device] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + for value in [ + "Accton AS7326", + "leaf", + "site1", + "Active", + "192.0.2.1/24", + "192.0.2.2/24", + "2001:db8::1/64", + ]: + assert value in out + + +def test_dump_prints_na_for_missing_details(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [_FakeDevice("sw1")] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + # Device Type, Device Role, Site, Status, OOB IP, Primary IPv4, Primary IPv6 + assert out.count("N/A") == 7 + + +def test_dump_formats_yaml_custom_fields(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + device = _FakeDevice( + "sw1", + custom_fields={ + "sonic_parameters": "hwsku: Accton-AS7326-56X", + "netplan_parameters": {"dummy0": {"addresses": ["192.0.2.1/32"]}}, + "frr_parameters": "a: [1, 2", + }, + ) + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [device] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + result = cmd.take_action(parsed_args) + + assert result is None + out = capsys.readouterr().out + # String value parsed and re-dumped as YAML + assert "hwsku: Accton-AS7326-56X" in out + # Dict value dumped directly + assert "dummy0:" in out + assert "- 192.0.2.1/32" in out + # Invalid YAML falls back to the raw string representation + assert "a: [1, 2" in out + + +def test_dump_appends_hostname_custom_fields(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + device = _FakeDevice( + "sw1", + custom_fields={ + "alternative_name": "alt1", + "inventory_hostname": "inv1", + "external_hostname": "ext1", + }, + ) + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [device] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert "Alternative Name" in out + assert "alt1" in out + assert "Inventory Hostname" in out + assert "inv1" in out + assert "External Hostname" in out + assert "ext1" in out + + +def test_dump_field_filter_matches_case_insensitive(capsys): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1", "IP"]) + + device = _FakeDevice( + "sw1", + site="site1", + oob_ip=SimpleNamespace(address="192.0.2.1/24"), + ) + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [device] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + result = cmd.take_action(parsed_args) + + assert result is None + out = capsys.readouterr().out + assert "Out-of-band IP" in out + assert "Primary IPv4" in out + assert "Primary IPv6" in out + assert "Site" not in out + assert "Device Type" not in out + + +def test_dump_field_filter_without_match_warns(capsys, loguru_logs): + cmd = _make_dump() + parsed_args = cmd.get_parser("test").parse_args(["sw1", "zzz"]) + + fake_nb = MagicMock() + fake_nb.dcim.devices.filter.return_value = [_FakeDevice("sw1")] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + result = cmd.take_action(parsed_args) + + assert result is None + assert capsys.readouterr().out == "" + assert any( + "No fields matching 'zzz' found" in record["message"] for record in loguru_logs + ) diff --git a/tests/unit/commands/test_sonic.py b/tests/unit/commands/test_sonic.py new file mode 100644 index 000000000..f166313cf --- /dev/null +++ b/tests/unit/commands/test_sonic.py @@ -0,0 +1,823 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the SONiC command base helpers and command orchestration. + +The SSH-key handling of ``_create_ssh_connection`` and the +``--refresh-host-key`` wiring are covered in ``test_sonic_ssh.py``; this +module covers the remaining ``SonicCommandBase`` helpers and the +``take_action`` control flow of the SSH-driven commands. +""" + +import json +from types import SimpleNamespace +from unittest.mock import MagicMock, call, mock_open, patch + +import paramiko +import pytest + +from osism.commands import sonic + + +class _ConcreteSonicCommand(sonic.SonicCommandBase): + """Concrete subclass so we can instantiate the abstract base in tests.""" + + def take_action(self, parsed_args): # pragma: no cover - not exercised + return 0 + + +def _make_base(): + return _ConcreteSonicCommand(MagicMock(), MagicMock()) + + +def _make_exec_result(exit_status=0, stdout=b"", stderr=b""): + out, err = MagicMock(), MagicMock() + out.channel.recv_exit_status.return_value = exit_status + out.read.return_value = stdout + err.read.return_value = stderr + return (MagicMock(), out, err) + + +def make_ssh(exit_status=0, stdout=b"", stderr=b""): + ssh = MagicMock() + ssh.exec_command.return_value = _make_exec_result(exit_status, stdout, stderr) + return ssh + + +# --- SonicCommandBase._get_device_from_netbox --- + + +def test_get_device_from_netbox_by_name(): + device = MagicMock() + fake_nb = MagicMock() + fake_nb.dcim.devices.get.return_value = device + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + assert _make_base()._get_device_from_netbox("sw1") is device + + fake_nb.dcim.devices.get.assert_called_once_with(name="sw1") + fake_nb.dcim.devices.filter.assert_not_called() + + +def test_get_device_from_netbox_by_inventory_hostname(loguru_logs): + first, second = MagicMock(), MagicMock() + fake_nb = MagicMock() + fake_nb.dcim.devices.get.return_value = None + fake_nb.dcim.devices.filter.return_value = [first, second] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + assert _make_base()._get_device_from_netbox("sw1") is first + + fake_nb.dcim.devices.filter.assert_called_once_with(cf_inventory_hostname="sw1") + assert any( + "found by inventory_hostname" in record["message"] for record in loguru_logs + ) + + +def test_get_device_from_netbox_not_found(loguru_logs): + fake_nb = MagicMock() + fake_nb.dcim.devices.get.return_value = None + fake_nb.dcim.devices.filter.return_value = [] + + with patch.dict("osism.utils.__dict__", {"nb": fake_nb}): + assert _make_base()._get_device_from_netbox("sw1") is None + + assert any( + "not found in NetBox" in record["message"] and record["level"] == "ERROR" + for record in loguru_logs + ) + + +# --- SonicCommandBase._get_config_context --- + + +def test_get_config_context_without_attribute(): + device = SimpleNamespace() + assert _make_base()._get_config_context(device, "sw1") is None + + +@pytest.mark.parametrize("context", [None, {}], ids=["none", "empty"]) +def test_get_config_context_empty(context): + device = SimpleNamespace(local_context_data=context) + assert _make_base()._get_config_context(device, "sw1") is None + + +def test_get_config_context_filters_underscore_keys(): + device = SimpleNamespace( + local_context_data={"_meta": 1, "management": {"ip": "10.0.0.1"}} + ) + result = _make_base()._get_config_context(device, "sw1") + assert result == {"management": {"ip": "10.0.0.1"}} + + +# --- SonicCommandBase._save_config_context --- + + +def test_save_config_context_writes_json(): + base = _make_base() + m_open = mock_open() + + with patch("builtins.open", m_open), patch( + "osism.commands.sonic.json.dump" + ) as mock_dump: + result = base._save_config_context({"a": 1}, "sw1", "20260718") + + assert result == "/tmp/config_db_sw1_20260718.json" + m_open.assert_called_once_with("/tmp/config_db_sw1_20260718.json", "w") + assert mock_dump.call_args[0][0] == {"a": 1} + assert mock_dump.call_args[1] == {"indent": 2} + + +def test_save_config_context_returns_none_on_error(loguru_logs): + base = _make_base() + + with patch("builtins.open", side_effect=OSError("denied")): + result = base._save_config_context({"a": 1}, "sw1", "20260718") + + assert result is None + assert any( + "Failed to save config context" in record["message"] for record in loguru_logs + ) + + +# --- SonicCommandBase._get_ssh_connection_details --- + + +def test_ssh_details_from_config_context(): + config_context = {"management": {"ip": "10.0.0.5", "username": "ops"}} + + with patch("osism.tasks.conductor.netbox.get_device_oob_ip") as mock_oob: + host, username = _make_base()._get_ssh_connection_details( + config_context, MagicMock(), "sw1" + ) + + assert (host, username) == ("10.0.0.5", "ops") + mock_oob.assert_not_called() + + +def test_ssh_details_fall_back_to_oob_ip(): + config_context = {"management": {"username": "ops"}} + device = MagicMock() + + with patch( + "osism.tasks.conductor.netbox.get_device_oob_ip", + return_value=("10.0.0.9", "eth0"), + ) as mock_oob: + host, username = _make_base()._get_ssh_connection_details( + config_context, device, "sw1" + ) + + assert (host, username) == ("10.0.0.9", "ops") + mock_oob.assert_called_once_with(device) + + +def test_ssh_details_none_when_no_host(loguru_logs): + with patch("osism.tasks.conductor.netbox.get_device_oob_ip", return_value=None): + result = _make_base()._get_ssh_connection_details({}, MagicMock(), "sw1") + + assert result == (None, None) + assert any("No SSH host found" in record["message"] for record in loguru_logs) + + +def test_ssh_details_default_username(): + config_context = {"management": {"ip": "10.0.0.5"}} + + with patch("osism.tasks.conductor.netbox.get_device_oob_ip") as mock_oob: + host, username = _make_base()._get_ssh_connection_details( + config_context, MagicMock(), "sw1" + ) + + assert (host, username) == ("10.0.0.5", "admin") + mock_oob.assert_not_called() + + +# --- SonicCommandBase._generate_backup_filename --- + + +def test_generate_backup_filename_first_free_slot(): + ssh = MagicMock() + ssh.exec_command.return_value = _make_exec_result(stdout=b"") + + result = _make_base()._generate_backup_filename(ssh, "sw1", "20260718") + + assert result == "/home/admin/config_db_sw1_20260718_1.json" + ssh.exec_command.assert_called_once_with( + "ls /home/admin/config_db_sw1_20260718_1.json 2>/dev/null" + ) + + +def test_generate_backup_filename_skips_existing_files(): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(stdout=b"/home/admin/config_db_sw1_20260718_1.json\n"), + _make_exec_result(stdout=b"/home/admin/config_db_sw1_20260718_2.json\n"), + _make_exec_result(stdout=b""), + ] + + result = _make_base()._generate_backup_filename(ssh, "sw1", "20260718") + + assert result == "/home/admin/config_db_sw1_20260718_3.json" + assert ssh.exec_command.call_count == 3 + assert ssh.exec_command.call_args == call( + "ls /home/admin/config_db_sw1_20260718_3.json 2>/dev/null" + ) + + +# --- exec-command helpers --- + +_EXEC_HELPER_CASES = [ + ( + "_backup_current_config", + ("/home/admin/backup.json",), + "sudo cp /etc/sonic/config_db.json /home/admin/backup.json", + ), + ("_load_configuration", ("/tmp/cfg.json",), "sudo config load -y /tmp/cfg.json"), + ("_reload_configuration", (), "sudo config reload -y"), + ("_save_configuration", (), "sudo config save -y"), + ("_enable_ztp", (), "sudo config ztp enable"), + ("_disable_ztp", (), "sudo config ztp disable"), +] + + +@pytest.mark.parametrize( + "method,args,expected_cmd", + _EXEC_HELPER_CASES, + ids=[c[0] for c in _EXEC_HELPER_CASES], +) +def test_exec_helper_success(method, args, expected_cmd): + ssh = make_ssh(exit_status=0) + + assert getattr(_make_base(), method)(ssh, *args) is True + ssh.exec_command.assert_called_once_with(expected_cmd) + + +@pytest.mark.parametrize( + "method,args,expected_cmd", + _EXEC_HELPER_CASES, + ids=[c[0] for c in _EXEC_HELPER_CASES], +) +def test_exec_helper_failure_logs_stderr(method, args, expected_cmd, loguru_logs): + ssh = make_ssh(exit_status=1, stderr=b"boom happened") + + assert getattr(_make_base(), method)(ssh, *args) is False + assert any( + "boom happened" in record["message"] and record["level"] == "ERROR" + for record in loguru_logs + ) + + +# --- SonicCommandBase._cleanup_temp_file / _get_ztp_status --- + + +def test_cleanup_temp_file_success(loguru_logs): + ssh = make_ssh(exit_status=0) + + assert _make_base()._cleanup_temp_file(ssh, "/tmp/x.json") is None + ssh.exec_command.assert_called_once_with("rm /tmp/x.json") + assert not any(record["level"] == "WARNING" for record in loguru_logs) + + +def test_cleanup_temp_file_warns_on_failure(loguru_logs): + ssh = make_ssh(exit_status=1, stderr=b"nope") + + assert _make_base()._cleanup_temp_file(ssh, "/tmp/x.json") is None + assert any( + "nope" in record["message"] and record["level"] == "WARNING" + for record in loguru_logs + ) + + +def test_get_ztp_status_returns_stripped_output(): + ssh = make_ssh(exit_status=0, stdout=b" ZTP Admin Mode : True\n") + + assert _make_base()._get_ztp_status(ssh) == "ZTP Admin Mode : True" + ssh.exec_command.assert_called_once_with("show ztp status") + + +def test_get_ztp_status_none_on_failure(): + ssh = make_ssh(exit_status=1, stderr=b"err") + + assert _make_base()._get_ztp_status(ssh) is None + + +# --- Load.take_action --- + +_LOAD_SSH_STEPS = [ + "_generate_backup_filename", + "_backup_current_config", + "_upload_config_context", + "_load_configuration", + "_save_configuration", + "_cleanup_temp_file", +] + + +def _stub_pre_ssh_helpers(cmd, ssh): + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value={"management": {"ip": "10.0.0.1"}}) + cmd._save_config_context = MagicMock(return_value="/tmp/cfg.json") + cmd._get_ssh_connection_details = MagicMock(return_value=("10.0.0.1", "admin")) + cmd._create_ssh_connection = MagicMock(return_value=ssh) + + +def _make_load(ssh): + cmd = sonic.Load(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + cmd._generate_backup_filename = MagicMock(return_value="/home/admin/backup_1.json") + cmd._backup_current_config = MagicMock(return_value=True) + cmd._upload_config_context = MagicMock( + return_value="/tmp/config_db_sw1_current.json" + ) + cmd._load_configuration = MagicMock(return_value=True) + cmd._save_configuration = MagicMock(return_value=True) + cmd._cleanup_temp_file = MagicMock() + return cmd + + +def test_load_happy_path_runs_steps_in_order(): + ssh = MagicMock() + cmd = _make_load(ssh) + manager = MagicMock() + for name in _LOAD_SSH_STEPS: + manager.attach_mock(getattr(cmd, name), name) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + + assert [name for name, _, _ in manager.mock_calls] == _LOAD_SSH_STEPS + ssh.close.assert_called_once_with() + + +@pytest.mark.parametrize( + "failing", + [ + "_get_device_from_netbox", + "_get_config_context", + "_save_config_context", + "_create_ssh_connection", + "_backup_current_config", + "_upload_config_context", + "_load_configuration", + "_save_configuration", + ], +) +def test_load_returns_one_when_step_fails(failing): + ssh = MagicMock() + cmd = _make_load(ssh) + getattr(cmd, failing).return_value = None + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + if failing in { + "_backup_current_config", + "_upload_config_context", + "_load_configuration", + "_save_configuration", + }: + ssh.close.assert_called_once_with() + + +def test_load_returns_one_when_no_ssh_host(): + cmd = _make_load(MagicMock()) + cmd._get_ssh_connection_details.return_value = (None, None) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + cmd._create_ssh_connection.assert_not_called() + + +@pytest.mark.parametrize( + "exc", + [ + paramiko.AuthenticationException(), + paramiko.SSHException("broken"), + Exception("boom"), + ], + ids=["auth", "ssh", "generic"], +) +def test_load_returns_one_on_ssh_exception(exc): + ssh = MagicMock() + cmd = _make_load(ssh) + cmd._backup_current_config.side_effect = exc + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + ssh.close.assert_called_once_with() + + +# --- Backup.take_action --- + + +def _make_backup(ssh): + cmd = sonic.Backup(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + cmd._generate_backup_filename = MagicMock(return_value="/home/admin/backup_1.json") + cmd._backup_current_config = MagicMock(return_value=True) + return cmd + + +def test_backup_happy_path(): + ssh = MagicMock() + cmd = _make_backup(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + + cmd._backup_current_config.assert_called_once_with(ssh, "/home/admin/backup_1.json") + ssh.close.assert_called_once_with() + + +def test_backup_returns_one_when_backup_fails(): + ssh = MagicMock() + cmd = _make_backup(ssh) + cmd._backup_current_config.return_value = False + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + ssh.close.assert_called_once_with() + + +# --- Ztp.take_action --- + + +def _make_ztp(ssh): + cmd = sonic.Ztp(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + cmd._enable_ztp = MagicMock(return_value=True) + cmd._disable_ztp = MagicMock(return_value=True) + cmd._get_ztp_status = MagicMock(return_value="ZTP Admin Mode : True") + return cmd + + +@pytest.mark.parametrize( + "action,helper,other_helpers", + [ + ("enable", "_enable_ztp", ["_disable_ztp", "_get_ztp_status"]), + ("disable", "_disable_ztp", ["_enable_ztp", "_get_ztp_status"]), + ], +) +def test_ztp_enable_disable(action, helper, other_helpers): + ssh = MagicMock() + cmd = _make_ztp(ssh) + parsed_args = cmd.get_parser("test").parse_args([action, "sw1"]) + + assert cmd.take_action(parsed_args) == 0 + + getattr(cmd, helper).assert_called_once_with(ssh) + for name in other_helpers: + getattr(cmd, name).assert_not_called() + ssh.close.assert_called_once_with() + + +@pytest.mark.parametrize( + "action,helper", [("enable", "_enable_ztp"), ("disable", "_disable_ztp")] +) +def test_ztp_enable_disable_failure(action, helper): + cmd = _make_ztp(MagicMock()) + getattr(cmd, helper).return_value = False + parsed_args = cmd.get_parser("test").parse_args([action, "sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + +def test_ztp_status_success(): + cmd = _make_ztp(MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["status", "sw1"]) + + assert cmd.take_action(parsed_args) == 0 + cmd._get_ztp_status.assert_called_once() + + +def test_ztp_status_failure(): + cmd = _make_ztp(MagicMock()) + cmd._get_ztp_status.return_value = None + parsed_args = cmd.get_parser("test").parse_args(["status", "sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + +# --- Reload.take_action --- + + +def _make_reload(ssh): + cmd = sonic.Reload(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + cmd._generate_backup_filename = MagicMock(return_value="/home/admin/backup_1.json") + cmd._backup_current_config = MagicMock(return_value=True) + cmd._upload_config_context = MagicMock( + return_value="/tmp/config_db_sw1_current.json" + ) + cmd._load_configuration = MagicMock(return_value=True) + cmd._reload_configuration = MagicMock(return_value=True) + cmd._save_configuration = MagicMock(return_value=True) + cmd._cleanup_temp_file = MagicMock() + return cmd + + +def test_reload_happy_path(): + ssh = MagicMock() + cmd = _make_reload(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + cmd._save_configuration.assert_called_once_with(ssh) + ssh.close.assert_called_once_with() + + +def test_reload_skips_save_when_reload_fails(loguru_logs): + cmd = _make_reload(MagicMock()) + cmd._reload_configuration.return_value = False + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + cmd._save_configuration.assert_not_called() + assert any( + "Skipping config save due to reload failure" in record["message"] + for record in loguru_logs + ) + + +def test_reload_returns_one_when_save_fails(): + cmd = _make_reload(MagicMock()) + cmd._save_configuration.return_value = False + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + +# --- Reboot.take_action --- + + +def test_reboot_happy_path(): + exec_result = _make_exec_result() + ssh = MagicMock() + ssh.exec_command.return_value = exec_result + cmd = sonic.Reboot(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + + ssh.exec_command.assert_called_once_with("sudo reboot") + # The exit status is deliberately not checked: the reboot kills the + # connection before a status could be received. + exec_result[1].channel.recv_exit_status.assert_not_called() + ssh.close.assert_called_once_with() + + +# --- Reset.take_action --- + + +def _make_reset(ssh): + cmd = sonic.Reset(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + return cmd + + +def test_reset_cancelled_by_prompt(): + cmd = _make_reset(MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.commands.sonic.prompt", return_value="no" + ): + assert cmd.take_action(parsed_args) == 0 + + cmd._get_device_from_netbox.assert_not_called() + + +@pytest.mark.parametrize("answer", ["yes", "y", "YES", "Y"]) +def test_reset_proceeds_on_confirmation(answer): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(), + _make_exec_result(), + _make_exec_result(), + ] + cmd = _make_reset(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.commands.sonic.prompt", return_value=answer + ), patch( + "osism.commands.sonic.cleanup_ssh_known_hosts_for_node", return_value=True + ), patch( + "osism.tasks.netbox.set_provision_state" + ): + assert cmd.take_action(parsed_args) == 0 + + cmd._get_device_from_netbox.assert_called_once_with("sw1") + + +def test_reset_force_skips_prompt(): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(), + _make_exec_result(), + _make_exec_result(), + ] + cmd = _make_reset(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "--force"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.commands.sonic.prompt" + ) as mock_prompt, patch( + "osism.commands.sonic.cleanup_ssh_known_hosts_for_node", return_value=True + ), patch( + "osism.tasks.netbox.set_provision_state" + ): + assert cmd.take_action(parsed_args) == 0 + + mock_prompt.assert_not_called() + + +def test_reset_returns_one_when_first_grub_command_fails(): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(exit_status=1, stderr=b"grub err") + ] + cmd = _make_reset(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "--force"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.tasks.netbox.set_provision_state" + ) as mock_state: + assert cmd.take_action(parsed_args) == 1 + + mock_state.delay.assert_not_called() + ssh.close.assert_called_once_with() + + +def test_reset_returns_one_when_second_grub_command_fails(): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(), + _make_exec_result(exit_status=1, stderr=b"grub err"), + ] + cmd = _make_reset(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "--force"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.tasks.netbox.set_provision_state" + ) as mock_state: + assert cmd.take_action(parsed_args) == 1 + + mock_state.delay.assert_not_called() + + +@pytest.mark.parametrize("cleanup_result", [True, False]) +def test_reset_happy_path(cleanup_result): + ssh = MagicMock() + ssh.exec_command.side_effect = [ + _make_exec_result(), + _make_exec_result(), + _make_exec_result(), + ] + cmd = _make_reset(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "--force"]) + + with patch("osism.commands.sonic.utils.check_task_lock_and_exit"), patch( + "osism.commands.sonic.cleanup_ssh_known_hosts_for_node", + return_value=cleanup_result, + ) as mock_cleanup, patch("osism.tasks.netbox.set_provision_state") as mock_state: + assert cmd.take_action(parsed_args) == 0 + + assert ssh.exec_command.call_args_list == [ + call("sudo grub-editenv /host/grub/grubenv set onie_mode=uninstall"), + call("sudo grub-editenv /host/grub/grubenv set next_entry=ONIE"), + call("sudo reboot"), + ] + mock_cleanup.assert_called_once_with("sw1") + mock_state.delay.assert_called_once_with("sw1", "ztp") + ssh.close.assert_called_once_with() + + +# --- Show.take_action --- + + +def _make_show(ssh): + cmd = sonic.Show(MagicMock(), MagicMock()) + _stub_pre_ssh_helpers(cmd, ssh) + return cmd + + +def test_show_builds_command_from_parts(capsys): + ssh = make_ssh(exit_status=0, stdout=b"route table\n") + cmd = _make_show(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "ip", "route"]) + + assert cmd.take_action(parsed_args) == 0 + + ssh.exec_command.assert_called_once_with("show ip route") + assert "route table" in capsys.readouterr().out + ssh.close.assert_called_once_with() + + +def test_show_bare_command_without_parts(): + ssh = make_ssh(exit_status=0, stdout=b"help\n") + cmd = _make_show(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + ssh.exec_command.assert_called_once_with("show") + + +def test_show_empty_output_logs_info(capsys, loguru_logs): + ssh = make_ssh(exit_status=0, stdout=b"") + cmd = _make_show(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "version"]) + + assert cmd.take_action(parsed_args) == 0 + + assert capsys.readouterr().out == "" + assert any("no output" in record["message"] for record in loguru_logs) + + +def test_show_returns_one_on_failure(loguru_logs): + ssh = make_ssh(exit_status=2, stderr=b"bad command") + cmd = _make_show(ssh) + parsed_args = cmd.get_parser("test").parse_args(["sw1", "bogus"]) + + assert cmd.take_action(parsed_args) == 1 + assert any( + "bad command" in record["message"] and record["level"] == "ERROR" + for record in loguru_logs + ) + + +# --- Console.take_action --- + + +def _make_console(): + cmd = sonic.Console(MagicMock(), MagicMock()) + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value={"management": {"ip": "10.0.0.1"}}) + cmd._get_ssh_connection_details = MagicMock(return_value=("10.0.0.1", "admin")) + return cmd + + +def test_console_returns_one_when_key_missing(): + cmd = _make_console() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + with patch("osism.commands.sonic.os.path.exists", return_value=False): + assert cmd.take_action(parsed_args) == 1 + + +def test_console_builds_ssh_command(): + cmd = _make_console() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + with patch("osism.commands.sonic.os.path.exists", return_value=True), patch( + "osism.commands.sonic.ensure_known_hosts_file", return_value=True + ), patch("osism.commands.sonic.os.system", return_value=0) as mock_system: + assert cmd.take_action(parsed_args) == 0 + + ssh_command = mock_system.call_args[0][0] + assert "-i /ansible/secrets/id_rsa.operator" in ssh_command + assert "UserKnownHostsFile=" in ssh_command + assert "admin@10.0.0.1" in ssh_command + + +def test_console_returns_one_on_nonzero_exit(): + cmd = _make_console() + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + with patch("osism.commands.sonic.os.path.exists", return_value=True), patch( + "osism.commands.sonic.ensure_known_hosts_file", return_value=True + ), patch("osism.commands.sonic.os.system", return_value=256): + assert cmd.take_action(parsed_args) == 1 + + +# --- Dump.take_action --- + + +def test_dump_prints_config_context(capsys): + cmd = sonic.Dump(MagicMock(), MagicMock()) + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value={"management": {"ip": "10.0.0.1"}}) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 0 + + out = capsys.readouterr().out + assert json.loads(out) == {"management": {"ip": "10.0.0.1"}} + assert out.startswith("{\n ") + + +def test_dump_returns_one_without_device(): + cmd = sonic.Dump(MagicMock(), MagicMock()) + cmd._get_device_from_netbox = MagicMock(return_value=None) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 + + +def test_dump_returns_one_without_config_context(): + cmd = sonic.Dump(MagicMock(), MagicMock()) + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value=None) + parsed_args = cmd.get_parser("test").parse_args(["sw1"]) + + assert cmd.take_action(parsed_args) == 1 diff --git a/tests/unit/commands/test_sonic_validate.py b/tests/unit/commands/test_sonic_validate.py new file mode 100644 index 000000000..9033dacfa --- /dev/null +++ b/tests/unit/commands/test_sonic_validate.py @@ -0,0 +1,436 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism sonic`` Validate and List commands.""" + +import json +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from osism.commands import sonic + +SUPPORTED_HWSKU = "Accton-AS7326-56X" + + +# --- List.take_action --- + + +def _run_list(devices=None, argv=None, wait_exc=None): + cmd = sonic.List(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(argv or []) + + task = MagicMock() + if wait_exc is not None: + task.wait.side_effect = wait_exc + else: + task.wait.return_value = devices or [] + + with patch("osism.tasks.conductor.get_sonic_devices") as mock_get: + mock_get.delay.return_value = task + result = cmd.take_action(parsed_args) + + return result, mock_get + + +def test_list_provision_state_derivation_and_sorting(capsys, loguru_logs): + devices = [ + { + "name": "sw-d", + "hwsku": SUPPORTED_HWSKU, + "provision_state": "active", + "role_name": "leaf", + "oob_ip": "10.0.0.4", + "primary_ip": "192.0.2.4", + "version": "4.1.1", + }, + {"name": "sw-b", "hwsku": None}, + {"name": "sw-c", "hwsku": "Bogus-HWSKU"}, + {"name": "sw-a", "hwsku": SUPPORTED_HWSKU, "provision_state": None}, + ] + + result, _ = _run_list(devices) + + assert result == 0 + out = capsys.readouterr().out + # Rows are sorted by device name. + assert out.index("sw-a") < out.index("sw-b") < out.index("sw-c") < out.index("sw-d") + + lines = { + name: line + for line in out.splitlines() + for name in ("sw-a", "sw-b", "sw-c", "sw-d") + if name in line + } + assert "N/A" in lines["sw-a"] # supported HWSKU without provision state + assert "No HWSKU" in lines["sw-b"] + assert "Unsupported HWSKU" in lines["sw-c"] + assert "active" in lines["sw-d"] + # Missing role/oob_ip/primary_ip/version columns fall back to N/A. + assert lines["sw-b"].count("N/A") >= 4 + assert "Total: 4 devices" in out + assert any( + "unsupported HWSKU: Bogus-HWSKU" in record["message"] + and record["level"] == "WARNING" + for record in loguru_logs + ) + + +def test_list_forwards_device_name(): + _, mock_get = _run_list([], argv=["spine1"]) + mock_get.delay.assert_called_once_with(device_name="spine1") + + +def test_list_without_devices(capsys): + result, mock_get = _run_list([]) + + assert result == 0 + mock_get.delay.assert_called_once_with(device_name=None) + assert "No SONiC devices found matching the criteria" in capsys.readouterr().out + + +def test_list_returns_one_when_task_fails(loguru_logs): + result, _ = _run_list(wait_exc=Exception("broker down")) + + assert result == 1 + assert any( + "broker down" in record["message"] and record["level"] == "ERROR" + for record in loguru_logs + ) + + +# --- Validate helpers --- + + +def _make_validate(): + return sonic.Validate(MagicMock(), MagicMock()) + + +def _parse_validate(argv): + cmd = _make_validate() + return cmd, cmd.get_parser("test").parse_args(argv) + + +def _result(valid=True, errors=None, warnings=None, to_dict=None): + result = MagicMock() + result.valid = valid + result.errors = errors or [] + result.warnings = warnings or [] + if to_dict is not None: + result.to_dict.return_value = to_dict + return result + + +@pytest.fixture +def export_settings(monkeypatch): + monkeypatch.setattr("osism.settings.SONIC_EXPORT_PREFIX", "osism_") + monkeypatch.setattr("osism.settings.SONIC_EXPORT_SUFFIX", "_config_db.json") + + +# --- Validate._collect_sources --- + + +def test_collect_sources_file(tmp_path): + config = {"PORT": {}} + path = tmp_path / "config_db.json" + path.write_text(json.dumps(config)) + cmd, parsed_args = _parse_validate(["--file", str(path)]) + + assert cmd._collect_sources(parsed_args) == [(str(path), config)] + + +def test_collect_sources_from_netbox_requires_hostname(): + cmd, parsed_args = _parse_validate(["--from-netbox"]) + + with pytest.raises(ValueError): + cmd._collect_sources(parsed_args) + + +def test_collect_sources_from_netbox_one_tuple_per_hostname(): + cmd, parsed_args = _parse_validate(["h1", "h2", "--from-netbox"]) + cmd._config_from_netbox = MagicMock(side_effect=[{"A": {}}, None]) + + sources = cmd._collect_sources(parsed_args) + + assert sources == [("h1", {"A": {}}), ("h2", None)] + assert [c.args[0] for c in cmd._config_from_netbox.call_args_list] == ["h1", "h2"] + + +def test_collect_sources_generate_requires_hostname(): + cmd, parsed_args = _parse_validate(["--generate"]) + + with pytest.raises(ValueError): + cmd._collect_sources(parsed_args) + + +def test_collect_sources_generate_labels(): + cmd, parsed_args = _parse_validate(["h1", "--generate"]) + cmd._config_from_generate = MagicMock(return_value={"B": {}}) + + assert cmd._collect_sources(parsed_args) == [("h1 (generated)", {"B": {}})] + cmd._config_from_generate.assert_called_once_with("h1") + + +def test_collect_sources_export_dir_explicit(tmp_path): + cmd, parsed_args = _parse_validate(["--from-export-dir", str(tmp_path)]) + cmd._configs_from_export_dir = MagicMock(return_value=[]) + + cmd._collect_sources(parsed_args) + + cmd._configs_from_export_dir.assert_called_once_with(str(tmp_path), []) + + +def test_collect_sources_export_dir_falls_back_to_setting(monkeypatch, tmp_path): + monkeypatch.setattr("osism.settings.SONIC_EXPORT_DIR", str(tmp_path)) + cmd, parsed_args = _parse_validate(["--from-export-dir"]) + cmd._configs_from_export_dir = MagicMock(return_value=[]) + + cmd._collect_sources(parsed_args) + + cmd._configs_from_export_dir.assert_called_once_with(str(tmp_path), []) + + +# --- Validate._configs_from_export_dir --- + + +def test_configs_from_export_dir_missing_dir(export_settings, tmp_path): + cmd = _make_validate() + + with pytest.raises(ValueError): + cmd._configs_from_export_dir(str(tmp_path / "missing"), []) + + +def test_configs_from_export_dir_collects_matching_files( + export_settings, tmp_path, loguru_logs +): + (tmp_path / "osism_sw2_config_db.json").write_text('{"B": {}}') + (tmp_path / "osism_sw1_config_db.json").write_text('{"A": {}}') + (tmp_path / "README.txt").write_text("skip me") + (tmp_path / "osism_bad_config_db.json").write_text("{not json") + + sources = _make_validate()._configs_from_export_dir(str(tmp_path), []) + + assert sources == [ + (str(tmp_path / "osism_bad_config_db.json"), None), + (str(tmp_path / "osism_sw1_config_db.json"), {"A": {}}), + (str(tmp_path / "osism_sw2_config_db.json"), {"B": {}}), + ] + assert any("Could not read" in record["message"] for record in loguru_logs) + + +def test_configs_from_export_dir_filters_by_hostname(export_settings, tmp_path): + (tmp_path / "osism_sw1_config_db.json").write_text('{"A": {}}') + (tmp_path / "osism_sw2_config_db.json").write_text('{"B": {}}') + + sources = _make_validate()._configs_from_export_dir(str(tmp_path), ["sw2"]) + + assert sources == [(str(tmp_path / "osism_sw2_config_db.json"), {"B": {}})] + + +# --- Validate._config_from_netbox --- + + +def test_config_from_netbox_without_device(): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock(return_value=None) + + assert cmd._config_from_netbox("h1") is None + + +def test_config_from_netbox_without_context(): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value=None) + + assert cmd._config_from_netbox("h1") is None + + +def test_config_from_netbox_without_sonic_config(loguru_logs): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value={"management": {}}) + + assert cmd._config_from_netbox("h1") is None + assert any("sonic_config" in record["message"] for record in loguru_logs) + + +def test_config_from_netbox_returns_config(): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock(return_value=MagicMock()) + cmd._get_config_context = MagicMock(return_value={"sonic_config": {"PORT": {}}}) + + assert cmd._config_from_netbox("h1") == {"PORT": {}} + + +# --- Validate._config_from_generate --- + + +def _device_with(custom_fields): + return SimpleNamespace(custom_fields=custom_fields) + + +def test_config_from_generate_without_hwsku(loguru_logs): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock(return_value=_device_with({})) + + with patch( + "osism.tasks.conductor.sonic.config_generator.generate_sonic_config" + ) as mock_generate: + assert cmd._config_from_generate("h1") is None + + mock_generate.assert_not_called() + assert any("no HWSKU configured" in record["message"] for record in loguru_logs) + + +def test_config_from_generate_unsupported_hwsku(loguru_logs): + cmd = _make_validate() + cmd._get_device_from_netbox = MagicMock( + return_value=_device_with({"sonic_parameters": {"hwsku": "Bogus"}}) + ) + + with patch( + "osism.tasks.conductor.sonic.config_generator.generate_sonic_config" + ) as mock_generate: + assert cmd._config_from_generate("h1") is None + + mock_generate.assert_not_called() + assert any("is not supported" in record["message"] for record in loguru_logs) + + +@pytest.mark.parametrize("config_version", [None, "4.1.1"]) +def test_config_from_generate_calls_generator(config_version): + cmd = _make_validate() + sonic_parameters = {"hwsku": SUPPORTED_HWSKU} + if config_version: + sonic_parameters["config_version"] = config_version + device = _device_with({"sonic_parameters": sonic_parameters}) + cmd._get_device_from_netbox = MagicMock(return_value=device) + + with patch( + "osism.tasks.conductor.sonic.config_generator.generate_sonic_config", + return_value={"PORT": {}}, + ) as mock_generate: + assert cmd._config_from_generate("h1") == {"PORT": {}} + + mock_generate.assert_called_once_with(device, SUPPORTED_HWSKU, None, config_version) + + +# --- Validate.take_action --- + + +def test_take_action_returns_two_on_value_error(): + cmd, parsed_args = _parse_validate(["--from-netbox"]) + + assert cmd.take_action(parsed_args) == 2 + + +def test_take_action_returns_two_without_sources(loguru_logs): + cmd, parsed_args = _parse_validate(["h1", "--from-netbox"]) + cmd._collect_sources = MagicMock(return_value=[]) + + assert cmd.take_action(parsed_args) == 2 + assert any("No configurations found" in record["message"] for record in loguru_logs) + + +def test_take_action_all_valid_returns_zero(capsys): + cmd, parsed_args = _parse_validate(["h1", "--from-netbox"]) + cmd._collect_sources = MagicMock(return_value=[("h1", {"PORT": {}})]) + + with patch( + "osism.tasks.conductor.sonic.validator.validate_config", + return_value=_result(valid=True), + ) as mock_validate: + assert cmd.take_action(parsed_args) == 0 + + mock_validate.assert_called_once_with({"PORT": {}}) + assert "[OK]" in capsys.readouterr().out + + +def test_take_action_invalid_returns_one(capsys): + cmd, parsed_args = _parse_validate(["h1", "h2", "--from-netbox"]) + cmd._collect_sources = MagicMock( + return_value=[("h1", {"PORT": {}}), ("h2", {"VLAN": {}})] + ) + invalid = _result( + valid=False, + errors=[SimpleNamespace(message="bad", table="PORT", path=None)], + ) + + with patch( + "osism.tasks.conductor.sonic.validator.validate_config", + side_effect=[_result(valid=True), invalid], + ): + assert cmd.take_action(parsed_args) == 1 + + +def test_take_action_none_config_returns_two_despite_valid_results(capsys): + cmd, parsed_args = _parse_validate(["h1", "h2", "--from-netbox"]) + cmd._collect_sources = MagicMock(return_value=[("h1", {"PORT": {}}), ("h2", None)]) + + with patch( + "osism.tasks.conductor.sonic.validator.validate_config", + return_value=_result(valid=True), + ) as mock_validate: + assert cmd.take_action(parsed_args) == 2 + + # The unavailable configuration is never passed to the validator. + mock_validate.assert_called_once_with({"PORT": {}}) + + +def test_take_action_json_format(capsys): + cmd, parsed_args = _parse_validate( + ["h1", "h2", "--from-netbox", "--format", "json"] + ) + cmd._collect_sources = MagicMock(return_value=[("h1", {"PORT": {}}), ("h2", None)]) + to_dict = {"valid": True, "errors": [], "warnings": []} + + with patch( + "osism.tasks.conductor.sonic.validator.validate_config", + return_value=_result(valid=True, to_dict=to_dict), + ): + assert cmd.take_action(parsed_args) == 2 + + payload = json.loads(capsys.readouterr().out) + assert payload["h1"] == to_dict + assert payload["h2"] == { + "valid": False, + "errors": [{"message": "config not available"}], + } + + +# --- Validate._print_text_report --- + + +def _error(message, table=None, path=None): + return SimpleNamespace(message=message, table=table, path=path) + + +def test_print_text_report_formats_results(capsys): + results = [ + ("good", SimpleNamespace(valid=True, errors=[], warnings=["minor issue"])), + ( + "bad", + SimpleNamespace( + valid=False, + errors=[ + _error("wrong speed", table="PORT", path="Ethernet0.speed"), + _error("missing key", table="VLAN"), + _error("just broken"), + ], + warnings=[], + ), + ), + ("missing", None), + ] + + _make_validate()._print_text_report(results) + + out = capsys.readouterr().out + assert "[OK] good" in out + assert "[WARN] good: minor issue" in out + assert "[FAIL] bad: 3 error(s)" in out + assert "- wrong speed (PORT.Ethernet0.speed)" in out + assert "- missing key (VLAN)" in out + assert "- just broken" in out + assert "[ERROR] missing: configuration not available" in out + assert "Summary: 1 valid, 2 failed, 3 total" in out