Unit tests for osism/commands/ — network (netbox, sonic)

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 8 (#2199). This issue covers the two network-facing CLI command modules: osism/commands/netbox.py (739 LOC — Ironic, Sync, Manage, Versions, Console, Dump) and osism/commands/sonic.py (1415 LOC — SonicCommandBase plus Load, Backup, Ztp, Reload, Reboot, Reset, Show, Console, Dump, List, Validate). Both modules contain substantial pure decision logic (error classification, argument assembly, config-source collection, report formatting) that is well testable without network access.

Note: This is a large group (~2150 LOC combined). The targets below are prioritized: pure-logic / high-value functions first, thin task-dispatch wrappers last. Like the Tier 3 issues, this issue may be split further during implementation (natural split: netbox.py / sonic.py base+commands / sonic.py Validate+List).

Scope

Already covered — do not duplicate:

• tests/unit/commands/test_netbox.py covers: Ironic.take_action and Sync.take_action returning 1 on TimeoutError from utils.fetch_task_output; Dump.take_action returning 1 when utils.nb is None and when no device is found; netbox.Console.take_action returning 1 when token/URL are missing.
• tests/unit/commands/test_sonic_ssh.py covers: --refresh-host-key parser wiring for all seven SSH commands; SonicCommandBase._create_ssh_connection (refresh true/false/default, missing private key → None, refresh failure → continue); and that every SSH command's take_action forwards parsed_args.refresh_host_key to _create_ssh_connection.

This issue adds the remaining gaps: extend tests/unit/commands/test_netbox.py, and add tests/unit/commands/test_sonic.py (base helpers + command orchestration) and tests/unit/commands/test_sonic_validate.py (Validate, List).

Test targets

osism/commands/netbox.py

Sync._check_netbox_instance() — netbox.py:167

No patching needed beyond a MagicMock for nb (pure error-classification logic):

• nb is None/falsy → "Error: Not configured"
• nb.status() succeeds → "Success"; http_session.timeout set to the given timeout and restored in finally
• nb.status() raises with message containing "timed out" → "Error: Timeout"
• message containing "401" / "Unauthorized" → "Error: Auth failed"
• message containing "connection refused" → "Error: Connection refused"
• message containing "certificate" / "SSL" → "Error: SSL error"
• generic message > 50 chars → truncated to first 50 chars in "Error: ..."
• nb without http_session attribute (use object()-like spec) → no AttributeError, classification still works

Sync._check_netbox_connectivity() — netbox.py:214

Patch osism.commands.netbox.requests.get:

• nb is None → "Error: Not configured" (stage 1 never reached)
• requests.get raises requests.exceptions.Timeout → "Error: Timeout" (no nb.status() call)
• raises requests.exceptions.ConnectionError → "Error: Connection refused"
• raises requests.exceptions.SSLError → "Error: SSL error"
• raises generic Exception with long message → truncated "Error: ..."
• stage 1 OK, nb.status() succeeds → "Success"; requests.get called with verify=not ignore_ssl_errors (test both ignore_ssl_errors=True/False)
• stage 1 OK, nb.status() raises → same classification matrix as _check_netbox_instance (one or two parametrized cases suffice; the matrix is fully tested above)

Sync._build_netbox_table() — netbox.py:118

Patch osism.commands.netbox.settings.NETBOX_URL / NETBOX_TOKEN / IGNORE_SSL_ERRORS (monkeypatch) and osism.commands.netbox.utils.secondary_nb_list / osism.commands.netbox.utils.nb:

• NETBOX_URL unset and secondary_nb_list empty → ([], ["Name", "URL", "Site"])
• NETBOX_URL set → primary row ["primary", url, "N/A"] first
• secondary instances with netbox_name/netbox_site attributes → used; missing attributes → "N/A" (the getattr fallbacks)
• check_connectivity=True → headers gain "Status", _check_netbox_connectivity called for primary and _check_netbox_instance per secondary (stub both methods, assert per-row status appended)

Sync.take_action() — netbox.py:320 (gaps)

Existing test covers the timeout exit code. Add (patch osism.commands.netbox.utils.check_task_lock_and_exit, stub _build_netbox_table):

• --list → table printed via tabulate (capsys), returns None, no task scheduled
• --list with empty table → warning "No NetBox instances configured", nothing printed
• --check → _build_netbox_table(check_connectivity=True, timeout=20) called
• --no-wait → conductor.sync_netbox.delay called, fetch_task_output NOT called, returns None
• --filter foo → delay(node_name=None, netbox_filter="foo") kwargs forwarded
• happy path with wait → returns the value of utils.fetch_task_output

Ironic.take_action() — netbox.py:67 (gaps)

• happy path → conductor.sync_ironic.delay called with node_name, adopt, force, dry_run, skip_kernel_params, extra_kernel_params from parsed args (parse e.g. ["node1", "--adopt", "--skip-kernel-param", "a", "--extra-kernel-param", "b=1"] and assert kwargs); returns fetch_task_output result
• --no-wait → no fetch_task_output call, returns None

Manage.take_action() — netbox.py:439

Patch osism.commands.netbox.utils.check_task_lock_and_exit, osism.tasks.netbox.manage (the task object; assert on .si(*arguments)) and osism.tasks.handle_task. Pure argument-assembly logic:

• defaults → manage.si("run", "--wait", "--no-skipdtl", "--no-skipmtl", "--no-skipres")
• --no-netbox-wait → "--no-wait" instead of "--wait"
• --parallel 4 --limit foo → "--parallel", "4", "--limit", "foo" appended
• --skipdtl --skipmtl --skipres → positive flags instead of --no-skip*
• handle_task(task, wait, format="script", timeout=3600) and its return value passed through; --no-wait → wait=False

Versions.take_action() — netbox.py:491

Patch osism.tasks.netbox.ping and utils.check_task_lock_and_exit: ping.delay() called, task.wait(timeout=None, interval=0.5), task.get() result printed (capsys).

Console.take_action() — netbox.py:518 (gaps)

Existing test covers the unconfigured error. Add (patch os.path.exists, os.mkdir, os.remove, os.environ.get, builtins.open / use tmp_path + monkeypatched os.path.expanduser, subprocess.call, yaml.dump):

• config file already exists → no init/write, goes straight to subprocess.call("/usr/local/bin/nbcli info ...", shell=True)
• config file missing, token file + NETBOX_API present → nbcli init called, os.remove of the generated file, YAML config written with url/token/verify: False/filter_limit: 50
• argument quoting: arguments containing spaces wrapped in single quotes (e.g. ["filter", "name foo"] → ... filter 'name foo'), arguments without spaces left bare

Dump.take_action() — netbox.py:583 (gaps)

Existing tests cover unconfigured/not-found. Add (patch.dict("osism.utils.__dict__", {"nb": fake_nb}) as in the existing tests):

• found by exact name match (filter returns several devices; only exact d.name == host selected)
• name search empty → fallback chain cf_alternative_name → cf_inventory_hostname → cf_external_hostname, each verified against custom_fields for an exact match (test at least one fallback hit and one where filter returns a device whose custom field does NOT match exactly → continues to next stage)
• table content: device_type/role/site/status/oob_ip/primary_ip4/primary_ip6 present → stringified; absent (None) → "N/A" (capsys on the tabulate output)
• YAML custom fields (sonic_parameters etc.): string value → parsed via yaml.safe_load and re-dumped; dict value → dumped directly; invalid YAML raising in yaml.dump path → falls back to str(field_value)
• alternative_name / inventory_hostname / external_hostname custom fields appended when set
• field filter: case-insensitive substring match on field names; no match → warning logged, returns None, nothing printed; match → only matching rows printed

osism/commands/sonic.py

SonicCommandBase._get_device_from_netbox() — sonic.py:30

Patch osism.commands.sonic.utils (or patch.dict("osism.utils.__dict__", {"nb": fake_nb})):

• nb.dcim.devices.get(name=...) returns device → returned directly
• get returns None, filter(cf_inventory_hostname=...) returns list → first element returned, info logged
• both empty → None, error logged

SonicCommandBase._get_config_context() — sonic.py:45

• device without local_context_data attribute → None
• local_context_data empty dict / None → None
• keys starting with _ filtered out, others kept ({"_meta": 1, "management": {...}} → {"management": {...}})

SonicCommandBase._save_config_context() — sonic.py:60

• happy path: writes JSON to /tmp/config_db_<hostname>_<today>.json, returns path (patch builtins.open with mock_open or write to tmp_path via monkeypatched path; asserting json.dump content)
• open raises → None, error logged

SonicCommandBase._get_ssh_connection_details() — sonic.py:72

Patch osism.tasks.conductor.netbox.get_device_oob_ip (the import happens inside the function body, so patch at the source module):

• config_context["management"] with ip and username → both used, no OOB lookup
• management.ip missing, get_device_oob_ip returns ("10.0.0.1", ...) → OOB IP used
• no host anywhere → (None, None), error logged
• host found but no username → default "admin"

SonicCommandBase._generate_backup_filename() — sonic.py:162

Fake ssh.exec_command side effects:

• first ls probe returns empty stdout → /home/admin/config_db_<host>_<today>_1.json
• first two probes return non-empty (file exists), third empty → suffix _3

exec-command helpers — _backup_current_config() sonic.py:174, _load_configuration() sonic.py:207, _reload_configuration() sonic.py:222, _save_configuration() sonic.py:237, _enable_ztp() sonic.py:280, _disable_ztp() sonic.py:295

Parametrize over the six helpers (identical pattern): build a fake ssh where exec_command returns (stdin, stdout, stderr) and stdout.channel.recv_exit_status() is 0 or 1:

• exit status 0 → True, expected command string passed to exec_command (sudo cp ..., sudo config load -y ..., sudo config reload -y, sudo config save -y, sudo config ztp enable/disable)
• exit status != 0 → False, stderr decoded and logged

SonicCommandBase._cleanup_temp_file() — sonic.py:252 and _get_ztp_status() — sonic.py:265

• _cleanup_temp_file: non-zero exit → warning only, no exception, no return value
• _get_ztp_status: exit 0 → stripped stdout returned; non-zero → None

Load.take_action() — sonic.py:324

Stub the base-class helpers on the instance (pattern already used in test_sonic_ssh.py):

• happy path: all helpers succeed → returns 0; helpers called in order (_generate_backup_filename → _backup_current_config → _upload_config_context → _load_configuration → _save_configuration → _cleanup_temp_file); ssh.close() called (finally)
• each failing step (_get_device_from_netbox/_get_config_context/_save_config_context/_get_ssh_connection_details/_create_ssh_connection/_backup_current_config/_upload_config_context/_load_configuration/_save_configuration returning falsy) → returns 1 (parametrize)
• _backup_current_config raising paramiko.AuthenticationException / paramiko.SSHException / generic Exception → returns 1, ssh.close() still called

(Backup.take_action — sonic.py:425 — is a strict subset; one happy-path + one failure test is enough.)

Reload.take_action() — sonic.py:595

Key behavioral difference from Load:

• _reload_configuration returns False → _save_configuration NOT called, warning "Skipping config save due to reload failure", still returns 0
• _reload_configuration True but _save_configuration False → returns 1
• full happy path → returns 0

Ztp.take_action() — sonic.py:504

• action="enable" → _enable_ztp called, others not; failure → 1
• action="disable" → _disable_ztp; failure → 1
• action="status" → _get_ztp_status; None → 1; string → 0

Reboot.take_action() — sonic.py:707

• happy path: exec_command("sudo reboot") called, returns 0 without checking exit status, ssh.close() called

Reset.take_action() — sonic.py:785

Patch osism.commands.sonic.prompt, osism.commands.sonic.utils.check_task_lock_and_exit, osism.commands.sonic.cleanup_ssh_known_hosts_for_node, and osism.tasks.netbox.set_provision_state (imported inside the function; assert .delay(hostname, "ztp")):

• no --force, prompt answers "no" → returns 0, no NetBox lookup
• prompt answers "yes" / "y" (case-insensitive) → proceeds
• --force → prompt never called
• first grub-editenv command non-zero exit → 1; second non-zero → 1
• happy path: both grub commands succeed, reboot issued, cleanup_ssh_known_hosts_for_node(hostname) called (both True and False results only change logging), set_provision_state.delay(hostname, "ztp") called, returns 0

Show.take_action() — sonic.py:922

• command=["ip", "route"] → executes "show ip route"; empty command → bare "show"
• exit status 0 with output → printed (capsys), returns 0; empty output → info log, returns 0
• non-zero exit → returns 1, stderr logged

Console.take_action() (sonic) — sonic.py:1010

Patch osism.commands.sonic.os.path.exists, osism.commands.sonic.ensure_known_hosts_file, osism.commands.sonic.os.system:

• missing key file → 1
• os.system returns 0 → 0; non-zero → 1; assert the ssh command string contains -i /ansible/secrets/id_rsa.operator, UserKnownHostsFile=, and user@host

Dump.take_action() (sonic) — sonic.py:1079

• happy path → filtered config context printed as json.dumps(..., indent=2) (capsys), returns 0
• no device / no context → 1

List.take_action() — sonic.py:1119

Patch osism.tasks.conductor.get_sonic_devices (imported inside the function; assert .delay(device_name=...) and stub task.wait()):

• provision-state derivation: hwsku=None → "No HWSKU"; hwsku not in SUPPORTED_HWSKUS → "Unsupported HWSKU" + warning; supported hwsku with provision_state=None → "N/A"; supported with state → state shown
• rows sorted by device name; None role/oob_ip/primary_ip/version → "N/A"
• empty device list → "No SONiC devices found matching the criteria" printed, returns 0
• task.wait() raises → returns 1

Validate — sonic.py:1187 (highest-value pure logic in the module)

_collect_sources() — sonic.py:1294:

• --file → single (path, parsed_json) tuple (use tmp_path)
• --from-netbox without hostname → ValueError; with hostnames → one tuple per hostname via _config_from_netbox (stub)
• --generate without hostname → ValueError; with hostname → label "<hostname> (generated)"
• --from-export-dir with explicit dir vs. empty (falls back to osism.settings.SONIC_EXPORT_DIR — monkeypatch)

_configs_from_export_dir() — sonic.py:1339 (use tmp_path, monkeypatch osism.settings.SONIC_EXPORT_PREFIX/SONIC_EXPORT_SUFFIX):

• non-existent dir → ValueError
• files not matching prefix/suffix skipped; identifier extracted by stripping prefix/suffix; results sorted by filename
• hostnames given → only matching identifiers included
• unreadable / invalid-JSON file → (path, None) entry plus error log

_config_from_netbox() — sonic.py:1324: no device → None; no context → None; context without sonic_config → None + error; with sonic_config → returned.

_config_from_generate() — sonic.py:1367 (patch osism.tasks.conductor.sonic.config_generator.generate_sonic_config and provide a fake device with custom_fields):

• no sonic_parameters.hwsku → None + error
• hwsku not in SUPPORTED_HWSKUS → None + error
• valid hwsku, with and without config_version → generate_sonic_config(device, hwsku, None, config_version) called and result returned

take_action() — sonic.py:1247 (stub _collect_sources and patch osism.tasks.conductor.sonic.validator.validate_config; results can be MagicMock(valid=..., errors=[...], to_dict=...)):

• _collect_sources raises ValueError → rc 2
• empty sources → rc 2, "No configurations found"
• all results valid → rc 0
• one invalid → rc 1
• one (label, None) source → rc 2 even if others are valid (worst_rc)
• --format json → json.dumps payload printed including the {"valid": False, ...} placeholder for None configs (capsys)

_print_text_report() — sonic.py:1395 (pure formatting, capsys):

• valid → [OK], invalid → [FAIL] with error count; None result → [ERROR] ... configuration not available
• error location formatting: table + path → table.path; only table; neither → bare message
• result.warnings → [WARN] lines; summary line counts valid/failed/total

Mocking hints

• Reuse the established patterns from the existing tests: instantiate commands as cmd = Cls(MagicMock(), MagicMock()), parse real argv via cmd.get_parser("test").parse_args([...]), and patch utils.nb via patch.dict("osism.utils.__dict__", {"nb": fake_nb}).
• For SONiC SSH helpers, one fixture builds the fake ssh:

  def make_ssh(exit_status=0, stdout=b"", stderr=b""):
      ssh = MagicMock()
      out, err = MagicMock(), MagicMock()
      out.channel.recv_exit_status.return_value = exit_status
      out.read.return_value = stdout
      err.read.return_value = stderr
      ssh.exec_command.return_value = (MagicMock(), out, err)
      return ssh

• All Celery tasks are dispatched from the commands via .delay() / .si().apply_async() — patch the task objects (osism.tasks.conductor.sync_netbox.delay, osism.tasks.netbox.manage, osism.tasks.netbox.set_provision_state.delay, osism.tasks.conductor.get_sonic_devices.delay); no broker is needed. The task bodies themselves are out of scope here (covered by the tasks-tier issues).
• Imports inside function bodies must be patched at their source module, not in osism.commands.sonic: osism.tasks.conductor.netbox.get_device_oob_ip, osism.tasks.conductor.sonic.validator.validate_config, osism.tasks.conductor.sonic.config_generator.generate_sonic_config, osism.tasks.handle_task.
• Reset uses prompt from prompt_toolkit — patch osism.commands.sonic.prompt to avoid blocking on stdin.
• Validate --file / --from-export-dir work well with real files in tmp_path — no mocks required.
• datetime.now() only feeds filename strings; assert with a wildcard or freeze via mocker.patch("osism.commands.sonic.datetime").
• For table/JSON output assertions use pytest's capsys rather than mocking print/tabulate.

Definition of Done

• tests/unit/commands/test_netbox.py extended with the listed gap cases
• tests/unit/commands/test_sonic.py created (base helpers + command orchestration)
• tests/unit/commands/test_sonic_validate.py created (Validate, List)
• All listed cases covered, no duplication of existing tests
• pytest --cov=osism.commands.netbox --cov=osism.commands.sonic shows ≥ 80 %
• pipenv run pytest tests/unit/commands/test_netbox.py tests/unit/commands/test_sonic.py tests/unit/commands/test_sonic_validate.py tests/unit/commands/test_sonic_ssh.py passes locally
• flake8, mypy, python-black remain green
• Zuul job python-osism-unit-tests passes

Dependencies

• Part of Meta: Unit test coverage for osism/ #2199 (meta). Blocked by Introduce foundation for unit tests (pytest + Zuul CI integration) #2192 / PR Introduce foundation for unit tests (pytest + Zuul CI) #2193.