From 2ef1b88d3a38fda1962f18bec3dbfd752f7fc791 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Fri, 17 Jul 2026 13:21:54 +0200 Subject: [PATCH] Add unit tests for workflow commands Cover the workflow-oriented CLI command modules under osism/commands/ with unit tests following the established pattern (command instantiated with mocked app/options, args built via get_parser().parse_args()): - apply: _prepare_task environment resolution (ceph-/kolla- prefix stripping, sub-environments, kubernetes routing, the loadbalancer-ng chain-into-group case, kolla_action/kolla_action_ng arguments, the osism-ansible runtime override and the custom fallback), collection expansion into celery groups/chains with dry-run and show-tree modes, GroupResult dispatch in handle_role, task-lock enforcement, the Ansible-facts freshness backoff, // role splitting and retries - check: the pure helpers get_file_info, collect_file_info, parse_stat_output and _compare_file_info; container id and mount source detection from cgroup/mountinfo; Mount.take_action guard rails and rc contract; Inode.take_action explicit and sampled file lists - compose: SSH command construction and the known-hosts fallback warning - sync: Facts/CephKeys/Sonic task dispatch with lock ordering, the release base.yml version lookup, SBOM image derivation, skopeo-based SBOM extraction from an OCI layout and the versions.yml rendering/writing paths including --dry-run - get (extended): Hostvars/Hosts happy paths, VersionsManager container labels, Tasks inspect rendering, Facts cache lookups and truncation, States rendering - log: ara parameter forwarding, docker logs over SSH, /var/log path confinement and shell quoting in File, clush/ssh host routing and rc passthrough, the Opensearch query loop - console: the module-level host resolution helpers (DNS, Netbox fallback, inventory groups, interactive selection) and take_action routing for the container/ansible-console/clush/ssh host syntaxes - validate (extended): validator-to-runtime routing with environment overrides and playbook rewriting, the _handle_task wait/no-wait contract, and the SCS compliance check command construction and cleanup - wait (extended): get_all_task_ids merging, the requeue loop for PENDING/STARTED tasks, unavailable-task handling, --refresh and the script output format All Celery task calls, subprocesses, Docker/Netbox/Redis clients and interactive prompts are mocked; log assertions use a loguru sink since caplog does not capture loguru records. Three misbehaviors are pinned as strict xfail tests that assert the corrected behavior, so the marker has to be removed when the fix lands: compose mangles multi-token subcommands into unrunnable commands ("".join instead of " ".join), a successful collection segment silently skips the remaining // segments because handle_collection returns None instead of an exit code, and a failed loadbalancer child playbook does not propagate its rc (the child rcs live in the task-output stream, not in the Celery result). The remaining pinned follow-up fix candidates: handle_role logs task.task_id before the GroupResult check (GroupResult has no such attribute), --refresh in wait is ignored when task ids are passed on the CLI, log.File error messages use printf-style placeholders that loguru does not interpolate, and handle_collection still dispatches a noop group under --dry-run although log and help text claim otherwise. Closes #2363 Assisted-by: Claude:claude-fable-5 Signed-off-by: Christian Berendt --- tests/unit/commands/test_apply.py | 679 ++++++++++++++++++++++++++ tests/unit/commands/test_check.py | 702 +++++++++++++++++++++++++++ tests/unit/commands/test_compose.py | 67 +++ tests/unit/commands/test_console.py | 346 +++++++++++++ tests/unit/commands/test_get.py | 339 ++++++++++++- tests/unit/commands/test_log.py | 272 +++++++++++ tests/unit/commands/test_sync.py | 476 ++++++++++++++++++ tests/unit/commands/test_validate.py | 251 +++++++++- tests/unit/commands/test_wait.py | 166 +++++++ 9 files changed, 3291 insertions(+), 7 deletions(-) create mode 100644 tests/unit/commands/test_apply.py create mode 100644 tests/unit/commands/test_check.py create mode 100644 tests/unit/commands/test_compose.py create mode 100644 tests/unit/commands/test_console.py create mode 100644 tests/unit/commands/test_log.py create mode 100644 tests/unit/commands/test_sync.py diff --git a/tests/unit/commands/test_apply.py b/tests/unit/commands/test_apply.py new file mode 100644 index 000000000..c2229303b --- /dev/null +++ b/tests/unit/commands/test_apply.py @@ -0,0 +1,679 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism apply`` command. + +The apply command routes a role to one of the celery task queues +(osism-ansible, ceph-ansible, kolla-ansible, kubernetes) based on the +role-to-environment mapping, expands collections into task groups and +chains, and retries failed roles. These tests characterize: + +- ``_prepare_task``: environment resolution (explicit, mapped, ``custom`` + fallback), prefix stripping (``ceph-``/``kolla-``), sub-environments, + the ``kolla_action`` extra argument and the osism-ansible runtime + override; +- ``_handle_collection``/``handle_collection``: recursive group/chain + construction from ``Role`` trees, dry-run (noop tasks), show-tree + (log-only) mode and the scheduling/log contract; +- ``handle_role``/``handle_loadbalancer_task``: exit-code pass-through, + the ``GroupResult`` dispatch and the child-task semantics (a group + failure propagates as an exception; a strict xfail pins that a failed + child must propagate its rc); +- ``take_action``: task-lock enforcement, the role/environment table, + the time-based Ansible-facts freshness check, ``//`` role splitting, + retries and the collection branch (a strict xfail pins that a + successful collection must not skip the following ``//`` segments). +""" + +import time +import types +from unittest.mock import MagicMock, call, patch + +import pytest + +from osism import utils as osism_utils +from osism.commands import apply +from osism.data import enums, playbooks +from osism.data.enums import Role + +from ._helpers import assert_not_called_before_lock_check, make_command + + +@pytest.fixture(autouse=True) +def _reset_lazy_state(): + """Keep lazily initialized module state from leaking between tests. + + ``osism.data.playbooks`` caches the playbook maps injected via + ``_set_playbook_maps`` and ``take_action`` stores the facts-check + backoff timestamp on ``osism.utils``. + """ + osism_utils.__dict__.pop("_last_ansible_facts_check", None) + yield + playbooks._reset_caches() + osism_utils.__dict__.pop("_last_ansible_facts_check", None) + + +@pytest.fixture +def task_mocks(mocker): + """Patch the celery task entry points used by ``_prepare_task``.""" + return types.SimpleNamespace( + ansible_run=mocker.patch("osism.tasks.ansible.run"), + ansible_noop=mocker.patch("osism.tasks.ansible.noop"), + ceph_run=mocker.patch("osism.tasks.ceph.run"), + kolla_run=mocker.patch("osism.tasks.kolla.run"), + kubernetes_run=mocker.patch("osism.tasks.kubernetes.run"), + ) + + +@pytest.fixture +def take_action_mocks(mocker): + """Patch the utils helpers consulted by ``take_action``.""" + return types.SimpleNamespace( + lock=mocker.patch("osism.commands.apply.utils.check_task_lock_and_exit"), + facts=mocker.patch("osism.commands.apply.utils.check_ansible_facts"), + ) + + +def _set_playbook_maps(role2environment=None, role2runtime=None): + """Inject the lazy playbook maps; the autouse fixture resets them.""" + playbooks.MAP_ROLE2ENVIRONMENT = role2environment or {} + playbooks.MAP_ROLE2RUNTIME = role2runtime or {} + + +def _prepare_task(cmd, **overrides): + params = dict( + arguments=[], + environment=None, + overwrite=None, + sub=None, + role="testrole", + action="deploy", + wait=True, + format="log", + timeout=300, + task_timeout=3600, + ) + params.update(overrides) + return cmd._prepare_task(**params) + + +def _public_collection_kwargs(**overrides): + params = dict( + arguments=[], + environment=None, + overwrite=None, + sub=None, + collection="testcollection", + action="deploy", + wait=True, + format="log", + timeout=300, + task_timeout=3600, + retry=0, + dry_run=False, + show_tree=False, + ) + params.update(overrides) + return params + + +def _collection_kwargs(**overrides): + counter = overrides.pop("counter", 0) + params = _public_collection_kwargs(**overrides) + params["counter"] = counter + return params + + +def _handle_role(cmd, **overrides): + params = dict( + arguments=[], + environment=None, + overwrite=None, + sub=None, + role="testrole", + action="deploy", + wait=True, + format="log", + timeout=300, + task_timeout=3600, + ) + params.update(overrides) + return cmd.handle_role(**params) + + +# _prepare_task + + +def test_prepare_task_ceph_role_forces_ceph_environment(task_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + + t = _prepare_task(cmd, role="ceph", environment=None, arguments=["-v"]) + + task_mocks.ceph_run.si.assert_called_once_with( + "ceph", "ceph", ["-v"], auto_release_time=3600 + ) + assert t is task_mocks.ceph_run.si.return_value + + +def test_prepare_task_strips_ceph_prefix(task_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + + t = _prepare_task(cmd, role="ceph-osds", environment="ceph") + + task_mocks.ceph_run.si.assert_called_once_with( + "ceph", "osds", [], auto_release_time=3600 + ) + assert t is task_mocks.ceph_run.si.return_value + + +def test_prepare_task_sub_environment_suffix(task_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + + _prepare_task(cmd, role="ceph-osds", environment="ceph", sub="zone-a") + _prepare_task(cmd, role="kubeadm", environment="kubernetes", sub="zone-a") + _prepare_task(cmd, role="keystone", environment="kolla", sub="zone-a") + + task_mocks.ceph_run.si.assert_called_once_with( + "ceph.zone-a", "osds", [], auto_release_time=3600 + ) + task_mocks.kubernetes_run.si.assert_called_once_with( + "kubernetes.zone-a", "kubeadm", [], auto_release_time=3600 + ) + task_mocks.kolla_run.si.assert_called_once_with( + "kolla.zone-a", "keystone", ["-e kolla_action=deploy"], auto_release_time=3600 + ) + + +def test_prepare_task_kubernetes_environment_from_mapping(task_mocks): + _set_playbook_maps(role2environment={"kubeadm": "kubernetes"}) + cmd = make_command(apply.Run) + + t = _prepare_task(cmd, role="kubeadm", arguments=["-l node1"]) + + task_mocks.kubernetes_run.si.assert_called_once_with( + "kubernetes", "kubeadm", ["-l node1"], auto_release_time=3600 + ) + assert t is task_mocks.kubernetes_run.si.return_value + + +def test_prepare_task_kolla_strips_prefix_and_adds_action(task_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + + t = _prepare_task( + cmd, role="kolla-keystone", environment="kolla", arguments=["-l control"] + ) + + task_mocks.kolla_run.si.assert_called_once_with( + "kolla", + "keystone", + ["-e kolla_action=deploy", "-l control"], + auto_release_time=3600, + ) + assert t is task_mocks.kolla_run.si.return_value + + +def test_prepare_task_kolla_role_in_osism_ansible_runtime(task_mocks): + _set_playbook_maps( + role2environment={"keystone": "kolla"}, + role2runtime={"osism-ansible": ["keystone"]}, + ) + cmd = make_command(apply.Run) + + t = _prepare_task(cmd, role="keystone", arguments=["-l control"]) + + task_mocks.ansible_run.si.assert_called_once_with( + "kolla", "keystone", ["-l control"], auto_release_time=3600 + ) + task_mocks.kolla_run.si.assert_not_called() + assert t is task_mocks.ansible_run.si.return_value + + +def test_prepare_task_common_role_stays_in_kolla(task_mocks): + _set_playbook_maps( + role2environment={"common": "kolla"}, + role2runtime={"osism-ansible": ["common"]}, + ) + cmd = make_command(apply.Run) + + _prepare_task(cmd, role="common") + + task_mocks.kolla_run.si.assert_called_once_with( + "kolla", "common", ["-e kolla_action=deploy"], auto_release_time=3600 + ) + task_mocks.ansible_run.si.assert_not_called() + + +def test_prepare_task_unknown_role_falls_back_to_custom(task_mocks, loguru_logs): + _set_playbook_maps() + cmd = make_command(apply.Run) + + t = _prepare_task(cmd, role="myplay", arguments=["-v"]) + + task_mocks.ansible_run.si.assert_called_once_with( + "custom", "myplay", ["-v"], auto_release_time=3600 + ) + messages = [r["message"] for r in loguru_logs] + assert "Trying to run play myplay in environment custom" in messages + assert t is task_mocks.ansible_run.si.return_value + + +def test_prepare_task_overwrite_replaces_default_environment(task_mocks): + _set_playbook_maps(role2environment={"myplay": "generic"}) + cmd = make_command(apply.Run) + + _prepare_task(cmd, role="myplay", overwrite="other") + + task_mocks.ansible_run.si.assert_called_once_with( + "other", "myplay", [], auto_release_time=3600 + ) + + +# _handle_collection + + +def test_handle_collection_rejects_non_role_items(loguru_logs): + cmd = make_command(apply.Run) + + with pytest.raises(TypeError, match="Expected Role object, got str"): + cmd._handle_collection(["not-a-role"], **_collection_kwargs()) + + assert any( + r["level"] == "ERROR" and "Expected Role object" in r["message"] + for r in loguru_logs + ) + + +def test_handle_collection_flat_roles_wrapped_in_group(mocker): + group_mock = mocker.patch("celery.group") + cmd = make_command(apply.Run) + prepared = [MagicMock(name="pt-a"), MagicMock(name="pt-b")] + cmd._prepare_task = MagicMock(side_effect=prepared) + + result = cmd._handle_collection([Role("a"), Role("b")], **_collection_kwargs()) + + assert [c.args[4] for c in cmd._prepare_task.call_args_list] == ["a", "b"] + group_mock.assert_called_once_with(prepared) + assert result is group_mock.return_value + + +def test_handle_collection_nested_dependencies_chained(mocker): + group_mock = mocker.patch("celery.group") + chain_mock = mocker.patch("celery.chain") + cmd = make_command(apply.Run) + parent_pt = MagicMock(name="pt-parent") + child_pt = MagicMock(name="pt-child") + cmd._prepare_task = MagicMock(side_effect=[parent_pt, child_pt]) + + result = cmd._handle_collection( + [Role("parent", dependencies=[Role("child")])], **_collection_kwargs() + ) + + chain_mock.assert_called_once_with(parent_pt, group_mock.return_value) + assert group_mock.call_args_list == [ + call([child_pt]), + call([chain_mock.return_value]), + ] + assert result is group_mock.return_value + + +def test_handle_collection_dry_run_uses_noop_tasks(mocker, task_mocks): + group_mock = mocker.patch("celery.group") + cmd = make_command(apply.Run) + cmd._prepare_task = MagicMock() + + result = cmd._handle_collection( + [Role("a"), Role("b")], **_collection_kwargs(dry_run=True) + ) + + cmd._prepare_task.assert_not_called() + assert task_mocks.ansible_noop.si.call_count == 2 + task_mocks.ansible_noop.si.assert_called_with() + group_mock.assert_called_once_with([task_mocks.ansible_noop.si.return_value] * 2) + assert result is group_mock.return_value + + +def test_handle_collection_show_tree_only_logs(task_mocks, loguru_logs): + cmd = make_command(apply.Run) + cmd._prepare_task = MagicMock() + + result = cmd._handle_collection( + [Role("parent", dependencies=[Role("child")])], + **_collection_kwargs(show_tree=True), + ) + + assert result is None + cmd._prepare_task.assert_not_called() + task_mocks.ansible_noop.si.assert_not_called() + messages = [r["message"] for r in loguru_logs] + assert "A [0] - parent" in messages + assert "A [1] -- child" in messages + + +# handle_collection + + +def test_handle_collection_applies_prepared_group(loguru_logs): + cmd = make_command(apply.Run) + prepared = MagicMock() + cmd._handle_collection = MagicMock(return_value=prepared) + collection_roles = [Role("a")] + + with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": collection_roles}): + cmd.handle_collection(**_public_collection_kwargs()) + + args = cmd._handle_collection.call_args + assert args.args[0] is collection_roles + assert args.args[1] == 0 + prepared.apply_async.assert_called_once_with() + messages = [r["message"] for r in loguru_logs] + assert "Collection testcollection is prepared for execution" in messages + assert ( + "All tasks of the collection testcollection are prepared for execution" + in messages + ) + assert "Tasks are running in the background" in messages + + +def test_handle_collection_show_tree_does_not_apply(loguru_logs): + cmd = make_command(apply.Run) + prepared = MagicMock() + cmd._handle_collection = MagicMock(return_value=prepared) + + with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": [Role("a")]}): + cmd.handle_collection(**_public_collection_kwargs(show_tree=True)) + + prepared.apply_async.assert_not_called() + messages = [r["message"] for r in loguru_logs] + assert "Showing execution tree for collection testcollection" in messages + assert "Tasks are running in the background" not in messages + + +def test_handle_collection_dry_run_logs_but_still_applies(loguru_logs): + """Characterize current behavior: the dry-run log claims that no tasks + are scheduled (as does the ``--dry-run`` help text), but the prepared + (noop) group is still applied async and occupies the queue. Follow-up: + either skip ``apply_async`` under dry-run or fix log and help text.""" + cmd = make_command(apply.Run) + prepared = MagicMock() + cmd._handle_collection = MagicMock(return_value=prepared) + + with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": [Role("a")]}): + cmd.handle_collection(**_public_collection_kwargs(dry_run=True)) + + prepared.apply_async.assert_called_once_with() + messages = [r["message"] for r in loguru_logs] + assert "Dry run for collection testcollection. No tasks are scheduled." in messages + assert "Tasks are running in the background" not in messages + + +# handle_role / handle_loadbalancer_task + + +def test_handle_role_passes_rc_from_handle_task(mocker): + handle_task = mocker.patch("osism.tasks.handle_task", return_value=3) + cmd = make_command(apply.Run) + cmd._prepare_task = MagicMock() + task = cmd._prepare_task.return_value.apply_async.return_value + + rc = _handle_role(cmd) + + assert rc == 3 + assert cmd._prepare_task.call_args.args[4] == "testrole" + handle_task.assert_called_once_with(task, True, "log", 300) + + +def test_handle_role_group_result_routes_to_loadbalancer_handler(mocker): + from celery.result import GroupResult + + handle_task = mocker.patch("osism.tasks.handle_task") + cmd = make_command(apply.Run) + cmd._prepare_task = MagicMock() + task = MagicMock(spec=GroupResult) + task.task_id = "group-task-id" + cmd._prepare_task.return_value.apply_async.return_value = task + cmd.handle_loadbalancer_task = MagicMock(return_value=5) + + rc = _handle_role(cmd, role="loadbalancer") + + assert rc == 5 + cmd.handle_loadbalancer_task.assert_called_once_with(task, True, "log", 300) + handle_task.assert_not_called() + + +def test_handle_loadbalancer_task_wait_uses_parent_rc(mocker, loguru_logs): + handle_task = mocker.patch("osism.tasks.handle_task", return_value=2) + cmd = make_command(apply.Run) + t = MagicMock() + t.children = [MagicMock(task_id="child-1"), MagicMock(task_id="child-2")] + + rc = cmd.handle_loadbalancer_task(t, True, "log", 300) + + assert rc == 2 + handle_task.assert_called_once_with(t.parent, True, "log", 300) + t.parent.get.assert_not_called() + t.get.assert_called_once_with() + messages = [r["message"] for r in loguru_logs] + assert any( + "Task child-1 (loadbalancer) is running in background" in m for m in messages + ) + assert any( + "Task child-2 (loadbalancer) is running in background" in m for m in messages + ) + + +def test_handle_loadbalancer_task_no_wait_collects_parent(mocker): + handle_task = mocker.patch("osism.tasks.handle_task", return_value=0) + cmd = make_command(apply.Run) + t = MagicMock() + t.children = [] + + rc = cmd.handle_loadbalancer_task(t, False, "log", 300) + + assert rc == 0 + handle_task.assert_called_once_with(t.parent, False, "log", 300) + t.parent.get.assert_called_once_with() + t.get.assert_called_once_with() + + +@pytest.mark.xfail( + strict=True, + reason="handle_loadbalancer_task returns only the parent rc and reads the " + "children solely for logging, so a failed loadbalancer child playbook " + "still exits 0; the child rcs have to be read the same way as the parent " + "rc (handle_task / the task-output stream) and the worst one propagated", +) +def test_handle_loadbalancer_task_child_failure_propagates(mocker): + """A non-zero rc recorded for a child task must surface in the return + value. The rc lives in the task-output stream read by ``handle_task``, + not in the Celery result, so ``child.get()`` cannot supply it.""" + cmd = make_command(apply.Run) + t = MagicMock() + child = MagicMock(task_id="child-1") + t.children = [child] + + def fake_handle_task(task, wait, format, timeout): + return 99 if task is child else 0 + + mocker.patch("osism.tasks.handle_task", side_effect=fake_handle_task) + + rc = cmd.handle_loadbalancer_task(t, True, "log", 300) + + assert rc == 99 + + +def test_handle_loadbalancer_task_group_failure_propagates(mocker): + handle_task = mocker.patch("osism.tasks.handle_task", return_value=0) + cmd = make_command(apply.Run) + t = MagicMock() + t.children = [MagicMock(task_id="child-1")] + t.get.side_effect = RuntimeError("child task failed") + + with pytest.raises(RuntimeError, match="child task failed"): + cmd.handle_loadbalancer_task(t, True, "log", 300) + + handle_task.assert_called_once_with(t.parent, True, "log", 300) + + +# take_action + + +def test_take_action_checks_task_lock_first(take_action_mocks): + _set_playbook_maps(role2environment={"myrole": "generic"}) + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + take_action_mocks.lock.side_effect = assert_not_called_before_lock_check( + cmd.handle_role + ) + parsed_args = cmd.get_parser("test").parse_args(["myrole"]) + + rc = cmd.take_action(parsed_args) + + assert rc == 0 + take_action_mocks.lock.assert_called_once_with() + cmd.handle_role.assert_called_once() + + +def test_take_action_without_role_prints_table(capsys, take_action_mocks, loguru_logs): + _set_playbook_maps(role2environment={"rolea": "enva", "roleb": "envb"}) + cmd = make_command(apply.Run) + parsed_args = cmd.get_parser("test").parse_args([]) + + rc = cmd.take_action(parsed_args) + + assert rc == 0 + out = capsys.readouterr().out + assert "Role" in out and "Environment" in out + assert "rolea" in out and "enva" in out + assert "roleb" in out and "envb" in out + take_action_mocks.facts.assert_not_called() + messages = [r["message"] for r in loguru_logs] + assert ( + "No role given for execution. The roles listed in the table can be used." + in messages + ) + + +def test_take_action_runs_facts_check_when_stale(take_action_mocks): + _set_playbook_maps(role2environment={"myrole": "generic"}) + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["myrole"]) + + before = time.time() + cmd.take_action(parsed_args) + + take_action_mocks.facts.assert_called_once_with() + assert osism_utils.__dict__["_last_ansible_facts_check"] >= before + + +def test_take_action_skips_facts_check_when_recent(take_action_mocks): + _set_playbook_maps(role2environment={"myrole": "generic"}) + osism_utils._last_ansible_facts_check = time.time() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["myrole"]) + + cmd.take_action(parsed_args) + + take_action_mocks.facts.assert_not_called() + + +@pytest.mark.parametrize("role", ["gather-facts", "facts"]) +def test_take_action_skips_facts_check_for_facts_roles(take_action_mocks, role): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args([role]) + + cmd.take_action(parsed_args) + + take_action_mocks.facts.assert_not_called() + cmd.handle_role.assert_called_once() + + +def test_take_action_skips_facts_check_for_show_tree(take_action_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["--show-tree", "myrole"]) + + cmd.take_action(parsed_args) + + take_action_mocks.facts.assert_not_called() + + +def test_take_action_splits_roles_on_double_slash(take_action_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["rolea//roleb"]) + + rc = cmd.take_action(parsed_args) + + assert rc == 0 + assert [c.args[4] for c in cmd.handle_role.call_args_list] == ["rolea", "roleb"] + + +def test_take_action_retries_until_attempts_exhausted(take_action_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(return_value=1) + parsed_args = cmd.get_parser("test").parse_args(["--retry", "2", "myrole"]) + + rc = cmd.take_action(parsed_args) + + assert rc == 1 + assert cmd.handle_role.call_count == 3 + + +def test_take_action_retry_stops_after_success(take_action_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_role = MagicMock(side_effect=[1, 0]) + parsed_args = cmd.get_parser("test").parse_args(["--retry", "2", "myrole"]) + + rc = cmd.take_action(parsed_args) + + assert rc == 0 + assert cmd.handle_role.call_count == 2 + + +def test_take_action_routes_collection_to_handle_collection(take_action_mocks): + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd.handle_collection = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["testcollection"]) + + with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": [Role("a")]}): + rc = cmd.take_action(parsed_args) + + assert rc == 0 + cmd.handle_collection.assert_called_once() + assert cmd.handle_collection.call_args.args[4] == "testcollection" + + +@pytest.mark.xfail( + strict=True, + reason="handle_collection returns None instead of an exit code, so the " + "'//' loop treats a successful collection as failed, silently skips the " + "remaining segments and take_action returns None (exit 0)", +) +def test_take_action_collection_chain_continues_after_success(take_action_mocks): + """A successful collection segment must not swallow the following ``//`` + segments: ``osism apply testcollection//other`` has to schedule the + collection and then run the ``other`` role.""" + _set_playbook_maps() + cmd = make_command(apply.Run) + cmd._handle_collection = MagicMock(return_value=MagicMock()) + cmd.handle_role = MagicMock(return_value=0) + parsed_args = cmd.get_parser("test").parse_args(["testcollection//other"]) + + with patch.dict(enums.MAP_ROLE2ROLE, {"testcollection": [Role("a")]}): + rc = cmd.take_action(parsed_args) + + assert rc == 0 + cmd.handle_role.assert_called_once() + assert cmd.handle_role.call_args.args[4] == "other" diff --git a/tests/unit/commands/test_check.py b/tests/unit/commands/test_check.py new file mode 100644 index 000000000..b9b3e9c55 --- /dev/null +++ b/tests/unit/commands/test_check.py @@ -0,0 +1,702 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism check`` commands. + +``check mount`` verifies bind-mount integrity by comparing the container's +view of /opt/configuration with the view from a freshly started container; +``check inode`` prints a lightweight inode snapshot without spawning one. +These tests cover: + +- the pure helpers ``get_file_info``, ``collect_file_info`` and + ``parse_stat_output`` on real temporary trees; +- the ``Mount`` helpers ``_compare_file_info`` (mismatch classification) and + ``_get_container_id`` / ``_get_mount_source`` (procfs parsing); +- the ``Mount.take_action`` guard rails (missing path, missing Docker, + undeterminable mount source) and its exit-code / script-output contract; +- ``Inode.take_action`` for explicit file lists and random sampling. +""" + +import hashlib +import io +import os +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from osism.commands import check + +from ._helpers import make_command, parse_args + + +def _fake_open(payloads): + """Return an ``open`` replacement serving per-path text payloads. + + Paths missing from ``payloads`` raise ``IOError``, emulating procfs files + that are absent or unreadable outside a container. + """ + + def _open(path, *args, **kwargs): + if path in payloads: + return io.StringIO(payloads[path]) + raise IOError(f"unreadable: {path}") + + return _open + + +# --- get_file_info --- + + +def test_get_file_info_small_file_includes_metadata_and_hash(tmp_path): + filepath = tmp_path / "small.txt" + filepath.write_bytes(b"hello osism") + + info = check.get_file_info(str(filepath)) + + st = os.stat(filepath) + assert info["inode"] == st.st_ino + assert info["mtime"] == st.st_mtime + assert info["size"] == len(b"hello osism") + assert info["mode"] == st.st_mode + assert info["uid"] == st.st_uid + assert info["gid"] == st.st_gid + assert info["is_link"] is False + assert info["hash"] == hashlib.md5(b"hello osism").hexdigest() + + +def test_get_file_info_large_file_has_no_hash(tmp_path): + filepath = tmp_path / "large.bin" + filepath.write_bytes(b"\0" * 1024 * 1024) + + info = check.get_file_info(str(filepath)) + + assert info["size"] == 1024 * 1024 + assert info["hash"] is None + + +def test_get_file_info_directory_has_no_hash(tmp_path): + info = check.get_file_info(str(tmp_path)) + + assert info["hash"] is None + assert info["is_link"] is False + assert "error" not in info + + +def test_get_file_info_unreadable_file_omits_hash_only(tmp_path): + filepath = tmp_path / "secret.txt" + filepath.write_bytes(b"data") + + with patch("builtins.open", side_effect=IOError("denied")): + info = check.get_file_info(str(filepath)) + + assert info["hash"] is None + assert info["size"] == 4 + assert "error" not in info + + +def test_get_file_info_nonexistent_path_returns_error(tmp_path): + info = check.get_file_info(str(tmp_path / "missing.txt")) + + assert set(info) == {"error"} + assert "missing.txt" in info["error"] + + +# --- collect_file_info --- + + +def test_collect_file_info_walks_files_and_dirs_and_skips_noise(tmp_path): + (tmp_path / "a.txt").write_text("a") + sub = tmp_path / "sub" + sub.mkdir() + (sub / "b.txt").write_text("b") + for skip in [".git", "venv", "__pycache__"]: + skip_dir = tmp_path / skip + skip_dir.mkdir() + (skip_dir / "inner.txt").write_text("x") + (tmp_path / "link.txt").symlink_to(tmp_path / "a.txt") + + result = check.collect_file_info(str(tmp_path)) + + assert set(result) == {"a.txt", "sub", "sub/b.txt"} + assert result["a.txt"]["hash"] == hashlib.md5(b"a").hexdigest() + assert result["sub"]["hash"] is None + + +def test_collect_file_info_stops_at_max_files_and_warns(tmp_path, loguru_logs): + for name in ["a.txt", "b.txt", "c.txt", "d.txt"]: + (tmp_path / name).write_text(name) + + result = check.collect_file_info(str(tmp_path), max_files=2) + + assert len(result) == 2 + assert set(result) <= {"a.txt", "b.txt", "c.txt", "d.txt"} + assert any( + r["level"] == "WARNING" + and r["message"] == "Reached max file limit (2), stopping scan" + for r in loguru_logs + ) + + +# --- parse_stat_output --- + + +def test_parse_stat_output_parses_typed_fields(): + output = "\n".join( + [ + "FILE:environments/kolla/configuration.yml", + "INODE:12345", + "SIZE:678", + "MTIME:1700000000", + "HASH:d41d8cd98f00b204e9800998ecf8427e", + ] + ) + + result = check.parse_stat_output(output) + + info = result["environments/kolla/configuration.yml"] + assert info == { + "inode": 12345, + "size": 678, + "mtime": 1700000000.0, + "hash": "d41d8cd98f00b204e9800998ecf8427e", + } + assert isinstance(info["inode"], int) + assert isinstance(info["size"], int) + assert isinstance(info["mtime"], float) + + +def test_parse_stat_output_maps_hash_none_to_none(): + result = check.parse_stat_output("FILE:a.txt\nHASH:NONE\n") + + assert result == {"a.txt": {"hash": None}} + + +def test_parse_stat_output_captures_error_lines(): + result = check.parse_stat_output("FILE:a.txt\nERROR:Permission denied\n") + + assert result == {"a.txt": {"error": "Permission denied"}} + + +def test_parse_stat_output_ignores_noise_lines(): + output = "\n".join( + [ + "INODE:99", + "", + "FILE:a.txt", + "INODE:1", + "line without a colon", + "UNKNOWN:value", + ] + ) + + result = check.parse_stat_output(output) + + assert result == {"a.txt": {"inode": 1}} + + +def test_parse_stat_output_parses_multiple_files_independently(): + output = "\n".join( + [ + "FILE:a.txt", + "INODE:1", + "HASH:aaa", + "FILE:b.txt", + "INODE:2", + "HASH:NONE", + ] + ) + + result = check.parse_stat_output(output) + + assert result == { + "a.txt": {"inode": 1, "hash": "aaa"}, + "b.txt": {"inode": 2, "hash": None}, + } + + +def test_parse_stat_output_returns_empty_dict_for_empty_output(): + assert check.parse_stat_output("") == {} + + +# --- Mount._compare_file_info --- + + +def _compare(local, fresh, check_content=False): + cmd = make_command(check.Mount) + return cmd._compare_file_info(local, fresh, check_content) + + +def test_compare_file_info_reports_inode_mismatches(): + inode, content, missing_local, missing_fresh = _compare( + {"a": {"inode": 1}}, {"a": {"inode": 2}} + ) + + assert inode == [{"file": "a", "local_inode": 1, "fresh_inode": 2}] + assert content == [] + assert missing_local == [] + assert missing_fresh == [] + + +def test_compare_file_info_reports_content_mismatches_only_when_enabled(): + local = {"a": {"inode": 1, "hash": "aaa"}} + fresh = {"a": {"inode": 1, "hash": "bbb"}} + + _, content, _, _ = _compare(local, fresh, check_content=True) + assert content == [{"file": "a", "local_hash": "aaa", "fresh_hash": "bbb"}] + + assert _compare(local, fresh, check_content=False) == ([], [], [], []) + + +def test_compare_file_info_reports_files_missing_on_either_side(): + inode, content, missing_local, missing_fresh = _compare( + {"only-local": {"inode": 1}}, {"only-fresh": {"inode": 2}} + ) + + assert missing_local == ["only-fresh"] + assert missing_fresh == ["only-local"] + assert inode == [] + assert content == [] + + +def test_compare_file_info_skips_entries_with_errors(): + result = _compare( + {"a": {"error": "denied"}, "b": {"inode": 1}}, + {"a": {"inode": 99}, "b": {"error": "gone"}}, + check_content=True, + ) + + assert result == ([], [], [], []) + + +def test_compare_file_info_ignores_falsy_inodes(): + result = _compare( + {"a": {"inode": None}, "b": {"inode": 0}, "c": {"inode": 3}}, + {"a": {"inode": 1}, "b": {"inode": 2}, "c": {"inode": None}}, + ) + + assert result == ([], [], [], []) + + +def test_compare_file_info_ignores_missing_hashes(): + result = _compare( + {"a": {"inode": 1, "hash": None}}, + {"a": {"inode": 1, "hash": "bbb"}}, + check_content=True, + ) + + assert result == ([], [], [], []) + + +def test_compare_file_info_sorts_results_by_file_path(): + local = { + "z": {"inode": 1}, + "a": {"inode": 1}, + "n": {"inode": 5}, + "b": {"inode": 5}, + } + fresh = { + "z": {"inode": 2}, + "a": {"inode": 2}, + "y": {"inode": 7}, + "c": {"inode": 7}, + } + + inode, _, missing_local, missing_fresh = _compare(local, fresh) + + assert [m["file"] for m in inode] == ["a", "z"] + assert missing_local == ["c", "y"] + assert missing_fresh == ["b", "n"] + + +# --- Mount._get_container_id --- + + +def test_get_container_id_truncates_64_char_cgroup_id(): + payloads = {"/proc/self/cgroup": "12:memory:/docker/" + "a" * 64 + "\n"} + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open(payloads)): + assert cmd._get_container_id() == "a" * 12 + + +def test_get_container_id_returns_12_char_cgroup_id_as_is(): + payloads = {"/proc/self/cgroup": "12:cpu:/docker/abcdef123456\n"} + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open(payloads)): + assert cmd._get_container_id() == "abcdef123456" + + +def test_get_container_id_falls_back_to_12_char_hostname(): + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open({})), patch( + "osism.commands.check.os.uname", + return_value=SimpleNamespace(nodename="abcdef123456"), + ): + assert cmd._get_container_id() == "abcdef123456" + + +def test_get_container_id_falls_back_to_mountinfo(): + payloads = { + "/proc/self/mountinfo": ( + "1443 1442 8:1 /var/lib/docker/containers/" + + "b" * 64 + + "/resolv.conf /etc/resolv.conf rw - ext4 /dev/sda1 rw\n" + ) + } + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open(payloads)), patch( + "osism.commands.check.os.uname", + return_value=SimpleNamespace(nodename="not-a-container-id"), + ): + assert cmd._get_container_id() == "b" * 12 + + +def test_get_container_id_returns_none_when_undetectable(): + payloads = { + "/proc/self/cgroup": "0::/init.scope\n", + "/proc/self/mountinfo": "36 35 98:0 / / rw - ext4 /dev/sda1 rw\n", + } + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open(payloads)), patch( + "osism.commands.check.os.uname", + return_value=SimpleNamespace(nodename="host"), + ): + assert cmd._get_container_id() is None + + +# --- Mount._get_mount_source --- + + +def test_get_mount_source_returns_absolute_source_after_separator(): + payloads = { + "/proc/self/mountinfo": ( + "543 481 254:1 /cfg /opt/configuration rw,relatime - ext4 /dev/vda1 rw\n" + ) + } + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open(payloads)): + assert cmd._get_mount_source("/opt/configuration") == "/dev/vda1" + + +@pytest.mark.parametrize( + "line", + [ + # no " - " separator at all + "543 481 254:1 /cfg /opt/configuration rw,relatime\n", + # separator present but no source field after the fstype + "543 481 254:1 /cfg /opt/configuration rw - ext4\n", + # source is not an absolute path (e.g. overlayfs) + "543 481 254:1 /cfg /opt/configuration rw - overlay overlay rw\n", + # no entry for the requested mount point + "543 481 254:1 /cfg /other rw - ext4 /dev/vda1 rw\n", + ], +) +def test_get_mount_source_returns_none_for_unusable_lines(line): + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open({"/proc/self/mountinfo": line})): + assert cmd._get_mount_source("/opt/configuration") is None + + +def test_get_mount_source_returns_none_when_mountinfo_unreadable(): + cmd = make_command(check.Mount) + + with patch("builtins.open", side_effect=_fake_open({})): + assert cmd._get_mount_source("/opt/configuration") is None + + +# --- Mount.take_action --- + + +def _run_mount( + argv, + *, + docker_available=True, + path_exists=True, + socket_exists=True, + docker_mock=None, + container_id=None, + mount_info=None, + mountinfo_source=None, + local_info=None, + fresh_output="", + fresh_error=None, +): + """Drive ``Mount.take_action`` with all environment probes patched out. + + The instance helpers that talk to procfs and Docker are replaced with + mocks so the tests steer the control flow purely through return values. + Returns the ``(exit_code, command)`` pair; the command exposes the helper + mocks for call assertions. + """ + cmd, parsed_args = parse_args(check.Mount, argv) + cmd._get_container_id = MagicMock(return_value=container_id) + cmd._get_volume_mount_info = MagicMock(return_value=mount_info) + cmd._get_mount_source = MagicMock(return_value=mountinfo_source) + cmd._run_fresh_container = MagicMock( + return_value=fresh_output, side_effect=fresh_error + ) + + if docker_mock is None: + docker_mock = MagicMock() + + def fake_exists(path): + if path == check.DOCKER_SOCKET_PATH: + return socket_exists + return path_exists + + with patch("osism.commands.check.DOCKER_AVAILABLE", docker_available), patch( + "osism.commands.check.docker", docker_mock, create=True + ), patch("osism.commands.check.os.path.exists", side_effect=fake_exists), patch( + "osism.commands.check.collect_file_info", return_value=local_info or {} + ): + return cmd.take_action(parsed_args), cmd + + +def test_mount_fails_when_path_missing(capsys): + result, _ = _run_mount(["--format", "script"], path_exists=False) + + assert result == 1 + assert "FAILED: Path does not exist: /opt/configuration" in capsys.readouterr().out + + +def test_mount_fails_without_docker_library(capsys): + result, _ = _run_mount(["--format", "script"], docker_available=False) + + assert result == 1 + assert "FAILED: Docker Python library not available" in capsys.readouterr().out + + +def test_mount_fails_without_docker_socket(capsys): + result, _ = _run_mount(["--format", "script"], socket_exists=False) + + assert result == 1 + out = capsys.readouterr().out + assert "FAILED: Docker socket not found at /var/run/docker.sock" in out + + +def test_mount_fails_when_docker_connection_fails(capsys): + docker_mock = MagicMock() + docker_mock.from_env.side_effect = Exception("cannot connect") + + result, _ = _run_mount(["--format", "script"], docker_mock=docker_mock) + + assert result == 1 + out = capsys.readouterr().out + assert "FAILED: Failed to connect to Docker: cannot connect" in out + + +def test_mount_fails_when_mount_source_undeterminable(capsys): + result, cmd = _run_mount(["--format", "script"]) + + assert result == 1 + out = capsys.readouterr().out + assert "FAILED: Could not determine mount source for /opt/configuration" in out + # Without a container ID the Docker API is never queried. + cmd._get_volume_mount_info.assert_not_called() + cmd._get_mount_source.assert_called_once_with("/opt/configuration") + cmd._run_fresh_container.assert_not_called() + + +def test_mount_uses_bind_mount_source_and_passes(capsys): + result, cmd = _run_mount( + ["--format", "script"], + container_id="abcdef123456", + mount_info={ + "type": "bind", + "source": "/host/configuration", + "name": None, + "driver": None, + }, + ) + + assert result == 0 + assert "PASSED" in capsys.readouterr().out + args = cmd._run_fresh_container.call_args[0] + assert args[1] == "registry.osism.cloud/dockerhub/alpine:latest" + assert args[2] == "/host/configuration" + assert args[3] == "/opt/configuration" + + +def test_mount_volume_name_overrides_detected_volume(): + result, cmd = _run_mount( + ["--format", "script", "--volume-name", "configuration"], + container_id="abcdef123456", + mount_info={ + "type": "volume", + "source": "/var/lib/docker/volumes/auto/_data", + "name": "auto", + "driver": "local", + }, + ) + + assert result == 0 + assert cmd._run_fresh_container.call_args[0][2] == "configuration" + + +def test_mount_volume_uses_detected_name_without_override(): + result, cmd = _run_mount( + ["--format", "script"], + container_id="abcdef123456", + mount_info={ + "type": "volume", + "source": "/var/lib/docker/volumes/auto/_data", + "name": "auto", + "driver": "local", + }, + ) + + assert result == 0 + assert cmd._run_fresh_container.call_args[0][2] == "auto" + + +def test_mount_fails_when_fresh_container_fails(capsys): + result, _ = _run_mount( + ["--format", "script"], + container_id="abcdef123456", + mount_info={"type": "bind", "source": "/host/configuration"}, + fresh_error=RuntimeError("Failed to run fresh container: boom"), + ) + + assert result == 1 + assert "FAILED: Failed to run fresh container: boom" in capsys.readouterr().out + + +def test_mount_reports_inode_mismatches_in_script_format(capsys): + fresh_output = "\n".join( + ["FILE:a.txt", "INODE:2", "SIZE:5", "MTIME:1.0", "HASH:NONE"] + ) + + result, _ = _run_mount( + ["--format", "script"], + container_id="abcdef123456", + mount_info={"type": "bind", "source": "/host/configuration"}, + local_info={"a.txt": {"inode": 1, "size": 5, "hash": None}}, + fresh_output=fresh_output, + ) + + assert result == 1 + out = capsys.readouterr().out + assert "FAILED" in out + assert "INODE_MISMATCHES:1" in out + + +def test_mount_passes_in_log_format_with_mountinfo_fallback(loguru_logs): + result, cmd = _run_mount([], mountinfo_source="/host/configuration") + + assert result == 0 + messages = [r["message"] for r in loguru_logs] + assert "Could not determine current container ID" in messages + assert "Using mount source: /host/configuration" in messages + assert "Mount integrity check PASSED" in messages + + +# --- Inode.take_action --- + + +def test_inode_script_reports_explicit_files_and_skips_symlinks(tmp_path, capsys): + (tmp_path / "a.txt").write_text("content") + (tmp_path / "subdir").mkdir() + (tmp_path / "link.txt").symlink_to(tmp_path / "a.txt") + + cmd, parsed_args = parse_args( + check.Inode, + [ + "a.txt", + "subdir", + "link.txt", + "ghost.txt", + "--path", + str(tmp_path), + "--format", + "script", + ], + ) + result = cmd.take_action(parsed_args) + + assert result == 0 + inode_file = os.stat(tmp_path / "a.txt").st_ino + inode_dir = os.stat(tmp_path / "subdir").st_ino + assert capsys.readouterr().out.splitlines() == [ + f"File:a.txt:{inode_file}", + f"Dir:subdir:{inode_dir}", + ] + + +def test_inode_table_format_prints_snapshot(tmp_path, capsys): + (tmp_path / "a.txt").write_text("content") + + cmd, parsed_args = parse_args(check.Inode, ["a.txt", "--path", str(tmp_path)]) + result = cmd.take_action(parsed_args) + + assert result == 0 + out = capsys.readouterr().out + assert "Inode snapshot of random files" in out + for header in ["Path", "Type", "Inode", "Size"]: + assert header in out + assert "a.txt" in out + + +def test_inode_log_format_logs_rows(tmp_path, loguru_logs): + (tmp_path / "a.txt").write_text("content") + + cmd, parsed_args = parse_args( + check.Inode, ["a.txt", "--path", str(tmp_path), "--format", "log"] + ) + result = cmd.take_action(parsed_args) + + assert result == 0 + st = os.stat(tmp_path / "a.txt") + assert any( + r["message"] == f"a.txt: type=File, inode={st.st_ino}, size={st.st_size}" + for r in loguru_logs + ) + + +def test_inode_samples_random_entries_when_no_files_given(tmp_path, capsys): + env_sub = tmp_path / "environments" / "kolla" + env_sub.mkdir(parents=True) + for name in ["a.yml", "b.yml", "c.yml"]: + (env_sub / name).write_text(name) + (tmp_path / "environments" / "site.yml").write_text("site") + # Symlinked subdirectories are skipped by the sampling. + (tmp_path / "environments" / "linked").symlink_to(env_sub) + + inv_sub = tmp_path / "inventory" / "group_vars" + inv_sub.mkdir(parents=True) + (inv_sub / "all.yml").write_text("all") + + cmd, parsed_args = parse_args( + check.Inode, ["--path", str(tmp_path), "--format", "script"] + ) + with patch( + "osism.commands.check.random.sample", + side_effect=lambda population, k: sorted(population)[:k], + ): + result = cmd.take_action(parsed_args) + + assert result == 0 + lines = capsys.readouterr().out.splitlines() + reported = {":".join(line.split(":")[:2]) for line in lines} + assert reported == { + "Dir:environments/kolla", + "File:environments/kolla/a.yml", + "File:environments/site.yml", + "Dir:inventory/group_vars", + "File:inventory/group_vars/all.yml", + } + + +def test_inode_returns_zero_without_configuration_directories(tmp_path, capsys): + cmd, parsed_args = parse_args( + check.Inode, ["--path", str(tmp_path), "--format", "script"] + ) + result = cmd.take_action(parsed_args) + + assert result == 0 + assert capsys.readouterr().out == "" diff --git a/tests/unit/commands/test_compose.py b/tests/unit/commands/test_compose.py new file mode 100644 index 000000000..ea5e38ef4 --- /dev/null +++ b/tests/unit/commands/test_compose.py @@ -0,0 +1,67 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism compose`` command. + +``Run`` wraps a remote ``docker compose`` invocation in an SSH call to the +operator user on the target host. These tests pin the exact command string +that is executed - including the ``UserKnownHostsFile`` option and the +``/opt/`` project directory. A strict xfail pins the corrected +space-separated joining of multi-token remainder arguments, which is +currently mangled by ``"".join`` (``up -d`` becomes ``up-d``). A failed +known_hosts initialization must only warn; the SSH attempt is still made. +""" + +from unittest.mock import patch + +import pytest + +from osism import settings +from osism.commands import compose +from osism.commands.compose import KNOWN_HOSTS_PATH + +from ._helpers import parse_args + + +def _run_compose(args, *, known_hosts=True): + cmd, parsed_args = parse_args(compose.Run, args) + + with patch("osism.commands.compose.subprocess.call") as mock_call, patch( + "osism.commands.compose.ensure_known_hosts_file", return_value=known_hosts + ): + cmd.take_action(parsed_args) + + return mock_call + + +def test_run_builds_ssh_docker_compose_command(): + mock_call = _run_compose(["testhost", "production", "ps"]) + + expected = ( + "/usr/bin/ssh -i /ansible/secrets/id_rsa.operator " + "-o StrictHostKeyChecking=no -o LogLevel=ERROR " + f"-o UserKnownHostsFile={KNOWN_HOSTS_PATH} " + f"{settings.OPERATOR_USER}@testhost " + "'docker compose --project-directory=/opt/production ps'" + ) + mock_call.assert_called_once_with(expected, shell=True) + + +@pytest.mark.xfail( + strict=True, + reason="compose.py joins the remainder arguments with '\"\".join', so any " + "multi-token subcommand is mangled into an unrunnable command ('up -d' " + "becomes 'up-d'); needs '\" \".join'", +) +def test_run_joins_remainder_arguments_with_spaces(): + mock_call = _run_compose(["testhost", "production", "up", "-d"]) + + command = mock_call.call_args[0][0] + assert "'docker compose --project-directory=/opt/production up -d'" in command + + +def test_run_warns_but_still_connects_when_known_hosts_init_fails(loguru_logs): + mock_call = _run_compose(["testhost", "production", "ps"], known_hosts=False) + + mock_call.assert_called_once() + warnings = [r["message"] for r in loguru_logs if r["level"] == "WARNING"] + assert any(f"Could not initialize {KNOWN_HOSTS_PATH}" in m for m in warnings) diff --git a/tests/unit/commands/test_console.py b/tests/unit/commands/test_console.py new file mode 100644 index 000000000..da7064937 --- /dev/null +++ b/tests/unit/commands/test_console.py @@ -0,0 +1,346 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism console`` command and its resolution helpers. + +The module-level helpers resolve a target host step by step: DNS first +(``resolve_hostname_to_ip``), then the Netbox primary IPv4 address +(``get_primary_ipv4_from_netbox``), falling back to the original hostname +(``resolve_host_with_fallback``). ``get_hosts_from_group`` expands an +inventory group into a sorted host list and swallows any resolution error, +and ``select_host_from_list`` runs the interactive host picker. + +``Run.take_action`` routes on the host syntax (``host/`` container prompt, +``host/container`` container console, ``.host`` ansible console, ``:group`` +clush) and otherwise opens an SSH session, expanding inventory groups and +resolving the host first. The tests pin the constructed command lines, +including shell quoting and the ``UserKnownHostsFile`` SSH option. +""" + +import json +import socket +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from osism import settings +from osism.commands import console +from osism.commands.console import KNOWN_HOSTS_PATH + +from ._helpers import parse_args + +# --- resolve_hostname_to_ip --- + + +def test_resolve_hostname_to_ip_returns_ip_on_success(): + with patch("osism.commands.console.socket.gethostbyname", return_value="192.0.2.1"): + assert console.resolve_hostname_to_ip("testhost") == "192.0.2.1" + + +def test_resolve_hostname_to_ip_returns_none_on_dns_failure(): + with patch( + "osism.commands.console.socket.gethostbyname", + side_effect=socket.gaierror("Name or service not known"), + ): + assert console.resolve_hostname_to_ip("testhost") is None + + +# --- get_primary_ipv4_from_netbox --- + + +def test_get_primary_ipv4_returns_none_without_netbox_connection(): + with patch("osism.commands.console.utils.nb", None): + assert console.get_primary_ipv4_from_netbox("testhost") is None + + +def test_get_primary_ipv4_strips_prefix_from_netbox_address(): + device = MagicMock() + device.primary_ip4.address = "10.0.0.1/24" + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + with patch("osism.commands.console.utils.nb", nb): + assert console.get_primary_ipv4_from_netbox("testhost") == "10.0.0.1" + + nb.dcim.devices.get.assert_called_once_with(name="testhost") + + +def test_get_primary_ipv4_returns_none_when_device_missing(): + nb = MagicMock() + nb.dcim.devices.get.return_value = None + + with patch("osism.commands.console.utils.nb", nb): + assert console.get_primary_ipv4_from_netbox("testhost") is None + + +def test_get_primary_ipv4_returns_none_without_primary_ip(): + device = MagicMock() + device.primary_ip4 = None + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + with patch("osism.commands.console.utils.nb", nb): + assert console.get_primary_ipv4_from_netbox("testhost") is None + + +def test_get_primary_ipv4_warns_and_returns_none_on_netbox_error(loguru_logs): + nb = MagicMock() + nb.dcim.devices.get.side_effect = RuntimeError("connection refused") + + with patch("osism.commands.console.utils.nb", nb): + assert console.get_primary_ipv4_from_netbox("testhost") is None + + warnings = [r["message"] for r in loguru_logs if r["level"] == "WARNING"] + assert any("Error querying Netbox for testhost" in m for m in warnings) + + +# --- resolve_host_with_fallback --- + + +def test_resolve_host_with_fallback_prefers_dns_result(): + with patch( + "osism.commands.console.resolve_hostname_to_ip", return_value="192.0.2.1" + ), patch("osism.commands.console.get_primary_ipv4_from_netbox") as mock_netbox: + assert console.resolve_host_with_fallback("testhost") == "192.0.2.1" + + mock_netbox.assert_not_called() + + +def test_resolve_host_with_fallback_uses_netbox_when_dns_fails(): + with patch( + "osism.commands.console.resolve_hostname_to_ip", return_value=None + ), patch( + "osism.commands.console.get_primary_ipv4_from_netbox", return_value="10.0.0.1" + ): + assert console.resolve_host_with_fallback("testhost") == "10.0.0.1" + + +def test_resolve_host_with_fallback_returns_hostname_when_all_fail(loguru_logs): + with patch( + "osism.commands.console.resolve_hostname_to_ip", return_value=None + ), patch("osism.commands.console.get_primary_ipv4_from_netbox", return_value=None): + assert console.resolve_host_with_fallback("testhost") == "testhost" + + warnings = [r["message"] for r in loguru_logs if r["level"] == "WARNING"] + assert any("Could not resolve testhost via DNS or Netbox" in m for m in warnings) + + +# --- get_hosts_from_group --- + + +def test_get_hosts_from_group_returns_sorted_hosts(): + inventory = {"_meta": {"hostvars": {}}} + + with patch( + "osism.commands.console.get_inventory_path", + return_value="/ansible/inventory/hosts.yml", + ), patch( + "osism.commands.console.subprocess.check_output", + return_value=json.dumps(inventory).encode(), + ) as mock_check_output, patch( + "osism.commands.console.get_hosts_from_inventory", + return_value=["ctl002", "ctl001"], + ): + assert console.get_hosts_from_group("ctl") == ["ctl001", "ctl002"] + + mock_check_output.assert_called_once_with( + [ + "ansible-inventory", + "-i", + "/ansible/inventory/hosts.yml", + "--list", + "--limit", + "ctl", + ], + stderr=subprocess.DEVNULL, + ) + + +def test_get_hosts_from_group_returns_empty_list_on_error(): + with patch( + "osism.commands.console.get_inventory_path", + return_value="/ansible/inventory/hosts.yml", + ), patch( + "osism.commands.console.subprocess.check_output", + side_effect=subprocess.CalledProcessError(1, "ansible-inventory"), + ): + assert console.get_hosts_from_group("ctl") == [] + + +# --- select_host_from_list --- + + +def test_select_host_from_list_returns_chosen_host(capsys): + with patch("osism.commands.console.prompt", side_effect=["2"]): + assert console.select_host_from_list(["host1", "host2"]) == "host2" + + captured = capsys.readouterr() + assert "Group contains 2 hosts" in captured.out + assert "1) host1" in captured.out + + +@pytest.mark.parametrize("answer", ["q", "quit", "exit"]) +def test_select_host_from_list_returns_none_on_cancel(answer): + with patch("osism.commands.console.prompt", side_effect=[answer]): + assert console.select_host_from_list(["host1", "host2"]) is None + + +def test_select_host_from_list_retries_on_non_numeric_input(capsys): + with patch( + "osism.commands.console.prompt", side_effect=["abc", "1"] + ) as mock_prompt: + assert console.select_host_from_list(["host1", "host2"]) == "host1" + + assert mock_prompt.call_count == 2 + assert "Please enter a number between 1 and 2" in capsys.readouterr().out + + +def test_select_host_from_list_retries_on_out_of_range_input(): + with patch( + "osism.commands.console.prompt", side_effect=["0", "3", "2"] + ) as mock_prompt: + assert console.select_host_from_list(["host1", "host2"]) == "host2" + + assert mock_prompt.call_count == 3 + + +# --- Run.take_action --- + + +def _invoke_run( + args, + *, + known_hosts=True, + group_hosts=None, + selected=None, + resolved=None, + prompts=None, +): + cmd, parsed_args = parse_args(console.Run, args) + resolved_map = resolved or {} + + with patch( + "osism.commands.console.subprocess.call", return_value=0 + ) as mock_call, patch( + "osism.commands.console.ensure_known_hosts_file", return_value=known_hosts + ), patch( + "osism.commands.console.get_hosts_from_group", return_value=group_hosts or [] + ), patch( + "osism.commands.console.resolve_host_with_fallback", + side_effect=lambda host: resolved_map.get(host, host), + ) as mock_resolve, patch( + "osism.commands.console.select_host_from_list", return_value=selected + ) as mock_select, patch( + "osism.commands.console.prompt", side_effect=prompts or ["exit"] + ) as mock_prompt: + result = cmd.take_action(parsed_args) + + return result, { + "call": mock_call, + "resolve": mock_resolve, + "select": mock_select, + "prompt": mock_prompt, + } + + +def test_run_trailing_slash_routes_to_container_prompt(): + _, mocks = _invoke_run(["ctl001/"], prompts=["ps -a", "exit"]) + + mocks["prompt"].assert_called_with("ctl001>>> ") + mocks["resolve"].assert_called_once_with("ctl001") + mocks["call"].assert_called_once() + call_args = mocks["call"].call_args[0][0] + assert call_args[-1] == "docker 'ps -a'" + assert f"{settings.OPERATOR_USER}@ctl001" in call_args + + +def test_run_container_prompt_exit_immediately_makes_no_ssh_call(): + _, mocks = _invoke_run(["ctl001/"], prompts=["exit"]) + + mocks["call"].assert_not_called() + + +def test_run_slash_routes_to_container_console(): + _, mocks = _invoke_run(["ctl001/rabbitmq"]) + + mocks["prompt"].assert_not_called() + mocks["resolve"].assert_called_once_with("ctl001") + call_args = mocks["call"].call_args[0][0] + assert call_args[-1] == "docker exec -it rabbitmq bash" + assert "RequestTTY=force" in call_args + assert f"UserKnownHostsFile={KNOWN_HOSTS_PATH}" in call_args + assert f"{settings.OPERATOR_USER}@ctl001" in call_args + + +def test_run_container_console_shell_quotes_container_name(): + _, mocks = _invoke_run(["ctl001/rabbit mq"]) + + call_args = mocks["call"].call_args[0][0] + assert call_args[-1] == "docker exec -it 'rabbit mq' bash" + + +def test_run_leading_dot_routes_to_ansible_console(): + _, mocks = _invoke_run([".ctl001"]) + + mocks["call"].assert_called_once_with(["/run-ansible-console.sh", "ctl001"]) + + +def test_run_leading_colon_routes_to_clush(): + _, mocks = _invoke_run([":ctl"]) + + mocks["call"].assert_called_once_with( + ["/usr/local/bin/clush", "-l", settings.OPERATOR_USER, "-g", "ctl"] + ) + + +def test_run_ssh_uses_single_host_from_group(): + _, mocks = _invoke_run(["ctl"], group_hosts=["ctl001"]) + + mocks["select"].assert_not_called() + mocks["resolve"].assert_called_once_with("ctl001") + assert mocks["call"].call_args[0][0][-1] == f"{settings.OPERATOR_USER}@ctl001" + + +def test_run_ssh_prompts_selection_for_multi_host_group(): + _, mocks = _invoke_run(["ctl"], group_hosts=["ctl001", "ctl002"], selected="ctl002") + + mocks["select"].assert_called_once_with(["ctl001", "ctl002"]) + mocks["resolve"].assert_called_once_with("ctl002") + assert mocks["call"].call_args[0][0][-1] == f"{settings.OPERATOR_USER}@ctl002" + + +def test_run_ssh_cancelled_selection_skips_ssh_call(): + result, mocks = _invoke_run( + ["ctl"], group_hosts=["ctl001", "ctl002"], selected=None + ) + + assert result is None + mocks["resolve"].assert_not_called() + mocks["call"].assert_not_called() + + +def test_run_ssh_call_uses_resolved_host_and_known_hosts_file(): + _, mocks = _invoke_run(["testhost"], resolved={"testhost": "192.0.2.1"}) + + mocks["call"].assert_called_once_with( + [ + "/usr/bin/ssh", + "-i", + "/ansible/secrets/id_rsa.operator", + "-o", + "StrictHostKeyChecking=no", + "-o", + "LogLevel=ERROR", + "-o", + f"UserKnownHostsFile={KNOWN_HOSTS_PATH}", + f"{settings.OPERATOR_USER}@192.0.2.1", + ] + ) + + +def test_run_warns_but_continues_when_known_hosts_init_fails(loguru_logs): + _, mocks = _invoke_run(["testhost"], known_hosts=False) + + mocks["call"].assert_called_once() + warnings = [r["message"] for r in loguru_logs if r["level"] == "WARNING"] + assert any(f"Could not initialize {KNOWN_HOSTS_PATH}" in m for m in warnings) diff --git a/tests/unit/commands/test_get.py b/tests/unit/commands/test_get.py index c4996a3be..76542ebff 100644 --- a/tests/unit/commands/test_get.py +++ b/tests/unit/commands/test_get.py @@ -2,16 +2,25 @@ """Tests for the ``osism get`` commands. -These focus on the exit-code contract: a command must return a non-zero exit -status when the underlying query cannot be run (e.g. the inventory cannot be -loaded), but must keep returning success when the query runs fine and simply -yields an empty result. +The first tests focus on the exit-code contract: a command must return a +non-zero exit status when the underlying query cannot be run (e.g. the +inventory cannot be loaded), but must keep returning success when the query +runs fine and simply yields an empty result. + +The later tests cover the rendering paths: container versions read via the +Docker API, active and scheduled Celery tasks, cached Ansible facts and local +state facts from Redis, and the inventory-backed host and hostvars listings. """ import json +import pprint import subprocess +from datetime import datetime from unittest.mock import MagicMock, patch +import docker + +from osism import utils from osism.commands import get @@ -89,3 +98,325 @@ def test_hostvars_returns_success_when_variable_absent_from_result(): result = cmd.take_action(parsed_args) assert not result + + +def test_hostvars_prints_requested_variable(capsys): + cmd = _make(get.Hostvars) + parsed_args = cmd.get_parser("test").parse_args(["somehost", "myvar"]) + + with patch( + "osism.commands.get.get_inventory_path", + return_value="/ansible/inventory/hosts.yml", + ), patch( + "osism.commands.get.subprocess.check_output", + return_value=json.dumps({"myvar": "value1", "other": "x"}).encode(), + ): + result = cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert not result + assert "somehost" in out + assert "myvar" in out + assert "'value1'" in out + assert "other" not in out + + +def test_hostvars_lists_all_variables_without_variable_argument(capsys): + cmd = _make(get.Hostvars) + parsed_args = cmd.get_parser("test").parse_args(["somehost"]) + + with patch( + "osism.commands.get.get_inventory_path", + return_value="/ansible/inventory/hosts.yml", + ), patch( + "osism.commands.get.subprocess.check_output", + return_value=json.dumps({"var_a": 1, "var_b": "two"}).encode(), + ): + result = cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert not result + assert "var_a" in out + assert "var_b" in out + assert "'two'" in out + + +def test_hosts_prints_host_table(capsys): + cmd = _make(get.Hosts) + parsed_args = cmd.get_parser("test").parse_args([]) + + inventory = {"_meta": {"hostvars": {"node2": {}, "node1": {}}}} + with patch( + "osism.commands.get.get_inventory_path", + return_value="/ansible/inventory/hosts.yml", + ), patch( + "osism.commands.get.subprocess.check_output", + return_value=json.dumps(inventory).encode(), + ): + result = cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert not result + assert "Host" in out + assert "node1" in out + assert "node2" in out + + +# --- VersionsManager.take_action --- + + +def _container(labels): + container = MagicMock() + container.labels = labels + return container + + +def test_versions_manager_lists_module_versions_and_releases(capsys): + cmd = _make(get.VersionsManager) + parsed_args = cmd.get_parser("test").parse_args([]) + + containers = { + "osism-ansible": _container({"org.opencontainers.image.version": "9.0.0"}), + "ceph-ansible": _container( + { + "org.opencontainers.image.version": "9.1.0", + "de.osism.release.ceph": "18.2.4", + } + ), + "kolla-ansible": _container( + { + "org.opencontainers.image.version": "9.2.0", + "de.osism.release.openstack": "2025.1", + } + ), + } + client = MagicMock() + client.containers.get.side_effect = lambda name: containers[name] + + with patch("docker.from_env", return_value=client): + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + for name in containers: + assert name in out + assert "18.2.4" in out + assert "2025.1" in out + + +def test_versions_manager_skips_missing_containers(capsys): + cmd = _make(get.VersionsManager) + parsed_args = cmd.get_parser("test").parse_args([]) + + labels = { + "osism-ansible": {"org.opencontainers.image.version": "9.0.0"}, + "kolla-ansible": { + "org.opencontainers.image.version": "9.2.0", + "de.osism.release.openstack": "2025.1", + }, + } + + def get_container(name): + if name == "ceph-ansible": + raise docker.errors.NotFound("no such container") + return _container(labels[name]) + + client = MagicMock() + client.containers.get.side_effect = get_container + + with patch("docker.from_env", return_value=client): + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert "osism-ansible" in out + assert "kolla-ansible" in out + assert "ceph-ansible" not in out + + +# --- Tasks.take_action --- + + +def _run_tasks(active, scheduled): + cmd = _make(get.Tasks) + parsed_args = cmd.get_parser("test").parse_args([]) + + inspect = MagicMock() + inspect.active.return_value = active + inspect.scheduled.return_value = scheduled + + with patch("celery.Celery") as mock_celery: + mock_celery.return_value.control.inspect.return_value = inspect + cmd.take_action(parsed_args) + + +def test_tasks_lists_active_and_scheduled_tasks(capsys): + start = 1750000000.0 + active = { + "worker-a": [ + { + "id": "task-1", + "name": "osism.tasks.ansible.run", + "time_start": start, + "args": ["generic", "facts"], + } + ] + } + scheduled = { + "worker-b": [ + { + "id": "task-2", + "name": "osism.tasks.conductor.sync_sonic", + "time_start": start, + "args": [], + } + ] + } + + _run_tasks(active, scheduled) + + out = capsys.readouterr().out + assert "worker-a" in out + assert "task-1" in out + assert "osism.tasks.ansible.run" in out + assert "ACTIVE" in out + assert "worker-b" in out + assert "task-2" in out + assert "SCHEDULED" in out + assert str(datetime.fromtimestamp(start)) in out + assert "generic" in out + + +def test_tasks_renders_empty_table_when_no_tasks_are_reported(capsys): + _run_tasks({}, {}) + + out = capsys.readouterr().out + assert "Worker" in out + assert "ACTIVE" not in out + assert "SCHEDULED" not in out + + +# --- Facts.take_action / States.take_action --- + + +def _fake_redis(monkeypatch, value): + """Install a fake Redis client as the ``utils.redis`` module attribute. + + ``utils.redis`` is created lazily by ``osism.utils.__getattr__`` (which + would open a real connection) and then cached in the module globals. + Seeding the module dict directly avoids the connection attempt and is + undone by ``monkeypatch`` afterwards. + """ + fake = MagicMock() + fake.get.return_value = value + monkeypatch.setitem(vars(utils), "redis", fake) + return fake + + +def test_facts_logs_error_when_no_facts_cached(monkeypatch, loguru_logs, capsys): + _fake_redis(monkeypatch, None) + cmd = _make(get.Facts) + parsed_args = cmd.get_parser("test").parse_args(["testhost"]) + + cmd.take_action(parsed_args) + + assert any( + record["level"] == "ERROR" + and "No facts found in cache for testhost." in record["message"] + for record in loguru_logs + ) + assert capsys.readouterr().out == "" + + +def test_facts_prints_single_requested_fact(monkeypatch, capsys): + fake = _fake_redis( + monkeypatch, json.dumps({"ansible_hostname": "testhost", "other": 1}) + ) + cmd = _make(get.Facts) + parsed_args = cmd.get_parser("test").parse_args(["testhost", "ansible_hostname"]) + + cmd.take_action(parsed_args) + + fake.get.assert_called_once_with("ansible_factstesthost") + out = capsys.readouterr().out + assert "ansible_hostname" in out + assert "'testhost'" in out + assert "other" not in out + + +def test_facts_logs_error_for_unknown_fact(monkeypatch, loguru_logs, capsys): + _fake_redis(monkeypatch, json.dumps({"ansible_hostname": "testhost"})) + cmd = _make(get.Facts) + parsed_args = cmd.get_parser("test").parse_args(["testhost", "missing_fact"]) + + cmd.take_action(parsed_args) + + assert any( + record["level"] == "ERROR" + and "Fact missing_fact not found in cache for testhost." in record["message"] + for record in loguru_logs + ) + assert capsys.readouterr().out == "" + + +def test_facts_listing_truncates_ssh_host_key_facts(monkeypatch, capsys): + key_facts = { + "ansible_ssh_host_key_dsa_public": "ssh-dss " + "D" * 45, + "ansible_ssh_host_key_ecdsa_public": "ecdsa-sha2 " + "E" * 45, + "ansible_ssh_host_key_ed25519_public": "ssh-ed25519 " + "F" * 45, + "ansible_ssh_host_key_rsa_public": "ssh-rsa " + "R" * 45, + } + _fake_redis(monkeypatch, json.dumps({"ansible_hostname": "testhost", **key_facts})) + cmd = _make(get.Facts) + parsed_args = cmd.get_parser("test").parse_args(["testhost"]) + + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + for value in key_facts.values(): + formatted = pprint.pformat(value, indent=2, width=60, compact=True) + assert f"{formatted[0:40]}..." in out + assert formatted not in out + assert "'testhost'" in out + + +def test_states_lists_roles_and_skips_bootstrap(monkeypatch, capsys): + facts = { + "ansible_local": { + "osism": { + "bootstrap": {"state": "ok", "timestamp": "2026-01-01 00:00:00"}, + "docker": {"state": "ok", "timestamp": "2026-01-02 00:00:00"}, + "frr": {"state": "configured", "timestamp": "2026-01-03 00:00:00"}, + } + } + } + _fake_redis(monkeypatch, json.dumps(facts)) + cmd = _make(get.States) + parsed_args = cmd.get_parser("test").parse_args(["testhost"]) + + cmd.take_action(parsed_args) + + out = capsys.readouterr().out + assert "docker" in out + assert "frr" in out + assert "configured" in out + assert "2026-01-02 00:00:00" in out + assert "bootstrap" not in out + + +def test_states_prints_nothing_without_local_osism_facts(monkeypatch, capsys): + _fake_redis(monkeypatch, json.dumps({"ansible_local": {}})) + cmd = _make(get.States) + parsed_args = cmd.get_parser("test").parse_args(["testhost"]) + + cmd.take_action(parsed_args) + + assert capsys.readouterr().out == "" + + +def test_states_prints_nothing_without_cache_entry(monkeypatch, capsys): + _fake_redis(monkeypatch, None) + cmd = _make(get.States) + parsed_args = cmd.get_parser("test").parse_args(["testhost"]) + + cmd.take_action(parsed_args) + + assert capsys.readouterr().out == "" diff --git a/tests/unit/commands/test_log.py b/tests/unit/commands/test_log.py new file mode 100644 index 000000000..e621bae8f --- /dev/null +++ b/tests/unit/commands/test_log.py @@ -0,0 +1,272 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism log`` commands. + +``Ansible`` shells out to ``ara``, ``Container`` wraps ``docker logs`` in an +SSH call, and ``File`` tails a file below ``/var/log`` either via clush (for +inventory groups with multiple hosts) or via a direct SSH connection. These +tests pin the constructed command lines, the path-traversal guard and its +exit code, the shell quoting of the resolved path, and the pass-through of +non-zero clush/ssh return codes. ``Opensearch`` runs an interactive query +loop against the OpenSearch SQL plugin; its output formats (payload-only, +verbose with ``@timestamp`` fallback, raw response without hits) are pinned +via ``capsys``. +""" + +import json +from unittest.mock import patch + +from osism import settings +from osism.commands import log +from osism.commands.log import KNOWN_HOSTS_PATH + +from ._helpers import parse_args + +# --- Ansible.take_action --- + + +def test_ansible_appends_joined_parameters_to_ara(): + cmd, parsed_args = parse_args(log.Ansible, ["result", "list"]) + + with patch("osism.commands.log.subprocess.call") as mock_call: + cmd.take_action(parsed_args) + + mock_call.assert_called_once_with("/usr/local/bin/ara result list", shell=True) + + +# --- Container.take_action --- + + +def _run_container(args, *, known_hosts=True): + cmd, parsed_args = parse_args(log.Container, args) + + with patch("osism.commands.log.subprocess.call") as mock_call, patch( + "osism.commands.log.ensure_known_hosts_file", return_value=known_hosts + ): + cmd.take_action(parsed_args) + + return mock_call + + +def test_container_builds_docker_logs_ssh_command(): + mock_call = _run_container(["testhost", "nova_compute", "--tail", "50"]) + + command = mock_call.call_args[0][0] + assert mock_call.call_args.kwargs == {"shell": True} + assert f"{settings.OPERATOR_USER}@testhost" in command + assert f"UserKnownHostsFile={KNOWN_HOSTS_PATH}" in command + assert command.endswith("docker logs --tail 50 nova_compute") + + +def test_container_warns_but_still_connects_when_known_hosts_init_fails(loguru_logs): + mock_call = _run_container(["testhost", "nova_compute"], known_hosts=False) + + mock_call.assert_called_once() + warnings = [r["message"] for r in loguru_logs if r["level"] == "WARNING"] + assert any(f"Could not initialize {KNOWN_HOSTS_PATH}" in m for m in warnings) + + +# --- File.take_action --- + + +def _run_file(args, *, group_hosts, call_rc=0, resolved="192.0.2.1"): + cmd, parsed_args = parse_args(log.File, args) + + with patch( + "osism.commands.log.get_hosts_from_group", return_value=group_hosts + ), patch( + "osism.commands.log.resolve_host_with_fallback", return_value=resolved + ) as mock_resolve, patch( + "osism.commands.log.subprocess.call", return_value=call_rc + ) as mock_call, patch( + "osism.commands.log.ensure_known_hosts_file", return_value=True + ): + result = cmd.take_action(parsed_args) + + return result, mock_call, mock_resolve + + +def test_file_rejects_path_traversal_outside_var_log(loguru_logs): + result, mock_call, _ = _run_file(["testhost", "../../etc/passwd"], group_hosts=[]) + + assert result == 1 + mock_call.assert_not_called() + errors = [r["message"] for r in loguru_logs if r["level"] == "ERROR"] + assert any("must stay within /var/log" in m for m in errors) + + +def test_file_accepts_nested_path_below_var_log(): + result, mock_call, _ = _run_file( + ["testhost", "kolla/nova/nova-compute.log"], group_hosts=[] + ) + + assert result == 0 + tail_command = mock_call.call_args[0][0][-1] + assert tail_command == "tail -n 100 /var/log/kolla/nova/nova-compute.log" + + +def test_file_tail_command_includes_lines_and_follow_flag(): + _, mock_call, _ = _run_file( + ["testhost", "syslog", "--follow", "--lines", "500"], group_hosts=[] + ) + + tail_command = mock_call.call_args[0][0][-1] + assert tail_command == "tail -n 500 -f /var/log/syslog" + + +def test_file_shell_quotes_resolved_path(): + _, mock_call, _ = _run_file(["testhost", "nova/instance name.log"], group_hosts=[]) + + tail_command = mock_call.call_args[0][0][-1] + assert tail_command == "tail -n 100 '/var/log/nova/instance name.log'" + + +def test_file_uses_clush_for_group_with_multiple_hosts(): + result, mock_call, mock_resolve = _run_file( + ["ctl", "syslog"], group_hosts=["host1", "host2"] + ) + + assert result == 0 + mock_resolve.assert_not_called() + mock_call.assert_called_once_with( + [ + "/usr/local/bin/clush", + "-l", + settings.OPERATOR_USER, + "-w", + "host1,host2", + "tail -n 100 /var/log/syslog", + ] + ) + + +def test_file_passes_through_clush_return_code(loguru_logs): + result, _, _ = _run_file( + ["ctl", "syslog"], group_hosts=["host1", "host2"], call_rc=3 + ) + + assert result == 3 + errors = [r["message"] for r in loguru_logs if r["level"] == "ERROR"] + assert any("clush log tailing failed" in m for m in errors) + + +def test_file_substitutes_single_group_host_and_uses_ssh(): + result, mock_call, mock_resolve = _run_file( + ["ctl", "syslog"], group_hosts=["ctl001"] + ) + + assert result == 0 + mock_resolve.assert_called_once_with("ctl001") + assert mock_call.call_args[0][0][0] == "/usr/bin/ssh" + + +def test_file_resolves_non_group_host_for_ssh(): + result, mock_call, mock_resolve = _run_file( + ["testhost", "syslog"], group_hosts=[], resolved="192.0.2.10" + ) + + assert result == 0 + mock_resolve.assert_called_once_with("testhost") + mock_call.assert_called_once_with( + [ + "/usr/bin/ssh", + "-i", + "/ansible/secrets/id_rsa.operator", + "-o", + "StrictHostKeyChecking=no", + "-o", + "LogLevel=ERROR", + "-o", + f"UserKnownHostsFile={KNOWN_HOSTS_PATH}", + f"{settings.OPERATOR_USER}@192.0.2.10", + "tail -n 100 /var/log/syslog", + ] + ) + + +def test_file_passes_through_ssh_return_code(loguru_logs): + result, _, _ = _run_file(["testhost", "syslog"], group_hosts=[], call_rc=5) + + assert result == 5 + errors = [r["message"] for r in loguru_logs if r["level"] == "ERROR"] + assert any("ssh log tailing failed" in m for m in errors) + + +# --- Opensearch.take_action --- + + +def _run_opensearch(args, prompts, response_data=None): + cmd, parsed_args = parse_args(log.Opensearch, args) + + with patch("osism.commands.log.PromptSession") as mock_session_cls, patch( + "osism.commands.log.requests.post" + ) as mock_post: + mock_session_cls.return_value.prompt.side_effect = prompts + if response_data is not None: + mock_post.return_value.json.return_value = response_data + cmd.take_action(parsed_args) + + return mock_post + + +def test_opensearch_exit_breaks_loop_without_query(): + mock_post = _run_opensearch([], ["exit"]) + + mock_post.assert_not_called() + + +def test_opensearch_prints_payload_for_each_hit(capsys): + data = { + "hits": { + "hits": [ + {"_source": {"Payload": "line one"}}, + {"_source": {"Payload": "line two"}}, + ] + } + } + + mock_post = _run_opensearch([], ["SELECT * FROM logs", "exit"], data) + + assert capsys.readouterr().out == "line one\nline two\n" + request_body = json.loads(mock_post.call_args.kwargs["data"]) + assert request_body == {"query": "SELECT * FROM logs"} + assert mock_post.call_args.kwargs["verify"] is False + + +def test_opensearch_verbose_prints_metadata_with_timestamp_fallback(capsys): + data = { + "hits": { + "hits": [ + { + "_source": { + "timestamp": "2026-01-01T00:00:00", + "Hostname": "ctl001", + "programname": "nova-api", + "Payload": "request done", + } + }, + { + "_source": { + "@timestamp": "2026-01-02T00:00:00", + "Hostname": "ctl002", + "Payload": "booted", + } + }, + ] + } + } + + _run_opensearch(["--verbose"], ["SELECT * FROM logs", "exit"], data) + + assert capsys.readouterr().out == ( + "2026-01-01T00:00:00 | ctl001 | nova-api | request done\n" + "2026-01-02T00:00:00 | ctl002 | booted\n" + ) + + +def test_opensearch_prints_raw_response_without_hits(capsys): + data = {"error": "no permissions"} + + _run_opensearch([], ["SELECT 1", "exit"], data) + + assert capsys.readouterr().out == "{'error': 'no permissions'}\n" diff --git a/tests/unit/commands/test_sync.py b/tests/unit/commands/test_sync.py new file mode 100644 index 000000000..b2205259a --- /dev/null +++ b/tests/unit/commands/test_sync.py @@ -0,0 +1,476 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the ``osism sync`` commands. + +``Facts``, ``CephKeys`` and ``Sonic`` are thin Celery dispatchers: they must +consult the task lock before scheduling anything, forward the right playbook +or conductor arguments and propagate ``handle_task``'s exit code. ``Versions`` +syncs Kolla image versions from an SBOM container image into the configuration +repository; those tests cover the SBOM image derivation, the release lookup +against the release repository, the skopeo/OCI extraction and the rendering of +``versions.yml``. +""" + +import contextlib +import io +import json +import subprocess +import tarfile +from unittest.mock import MagicMock, patch + +import pytest +import requests +from yaml import YAMLError + +from osism.commands import sync + +from ._helpers import assert_not_called_before_lock_check, make_command, parse_args + +# --- Facts.take_action --- + + +@pytest.mark.parametrize("rc", [0, 2]) +def test_facts_schedules_gather_facts_and_returns_rc(rc): + cmd, parsed_args = parse_args(sync.Facts, []) + + with patch( + "osism.commands.sync.utils.check_task_lock_and_exit" + ) as mock_check, patch("osism.tasks.ansible.run.delay") as mock_delay, patch( + "osism.tasks.handle_task", return_value=rc + ) as mock_handle: + mock_check.side_effect = assert_not_called_before_lock_check(mock_delay) + result = cmd.take_action(parsed_args) + + mock_check.assert_called_once() + mock_delay.assert_called_once_with( + "generic", "gather-facts", [], auto_release_time=3600 + ) + mock_handle.assert_called_once_with(mock_delay.return_value) + assert result == rc + + +# --- CephKeys.take_action --- + + +def test_ceph_keys_schedules_copy_ceph_keys_and_waits(loguru_logs): + cmd, parsed_args = parse_args(sync.CephKeys, []) + + with patch( + "osism.commands.sync.utils.check_task_lock_and_exit" + ) as mock_check, patch("osism.tasks.ansible.run.delay") as mock_delay, patch( + "osism.tasks.handle_task", return_value=0 + ) as mock_handle: + mock_check.side_effect = assert_not_called_before_lock_check(mock_delay) + result = cmd.take_action(parsed_args) + + mock_check.assert_called_once() + mock_delay.assert_called_once_with( + "manager", "copy-ceph-keys", [], auto_release_time=3600 + ) + mock_handle.assert_called_once_with(mock_delay.return_value, True) + assert result == 0 + assert any( + "(sync ceph-keys) started" in record["message"] for record in loguru_logs + ) + + +def test_ceph_keys_no_wait_disables_waiting(): + cmd, parsed_args = parse_args(sync.CephKeys, ["--no-wait"]) + + with patch("osism.commands.sync.utils.check_task_lock_and_exit"), patch( + "osism.tasks.ansible.run.delay" + ) as mock_delay, patch("osism.tasks.handle_task", return_value=0) as mock_handle: + cmd.take_action(parsed_args) + + mock_handle.assert_called_once_with(mock_delay.return_value, False) + + +# --- Sonic.take_action --- + + +def _run_sonic(args): + cmd, parsed_args = parse_args(sync.Sonic, args) + + with patch( + "osism.commands.sync.utils.check_task_lock_and_exit" + ) as mock_check, patch( + "osism.tasks.conductor.sync_sonic.delay" + ) as mock_delay, patch( + "osism.tasks.handle_task", return_value=0 + ) as mock_handle: + mock_check.side_effect = assert_not_called_before_lock_check(mock_delay) + result = cmd.take_action(parsed_args) + + return result, mock_check, mock_delay, mock_handle + + +def test_sonic_syncs_specific_device_and_logs_device_name(loguru_logs): + result, mock_check, mock_delay, mock_handle = _run_sonic(["switch1"]) + + mock_check.assert_called_once() + mock_delay.assert_called_once_with("switch1", True) + mock_handle.assert_called_once_with(mock_delay.return_value, wait=True) + assert result == 0 + assert any( + "(sync sonic for device switch1) started" in record["message"] + for record in loguru_logs + ) + + +def test_sonic_without_device_logs_generic_message(loguru_logs): + _, _, mock_delay, _ = _run_sonic([]) + + mock_delay.assert_called_once_with(None, True) + assert any("(sync sonic) started" in record["message"] for record in loguru_logs) + assert not any("for device" in record["message"] for record in loguru_logs) + + +def test_sonic_no_diff_and_no_wait_are_forwarded(): + _, _, mock_delay, mock_handle = _run_sonic(["switch1", "--no-diff", "--no-wait"]) + + mock_delay.assert_called_once_with("switch1", False) + mock_handle.assert_called_once_with(mock_delay.return_value, wait=False) + + +# --- Versions._get_kolla_version_from_release --- + + +def _release_response(text): + response = MagicMock() + response.text = text + return response + + +def test_get_kolla_version_from_release_returns_version_from_base_yml(): + cmd = make_command(sync.Versions) + response = _release_response('docker_images:\n kolla: "0.20250928.0"\n') + + with patch("requests.get", return_value=response) as mock_get: + version = cmd._get_kolla_version_from_release( + "9.4.0", "https://example.com/release" + ) + + mock_get.assert_called_once_with( + "https://example.com/release/9.4.0/base.yml", timeout=30 + ) + response.raise_for_status.assert_called_once_with() + assert version == "0.20250928.0" + + +def test_get_kolla_version_from_release_raises_on_http_error(): + cmd = make_command(sync.Versions) + response = MagicMock() + response.raise_for_status.side_effect = requests.exceptions.RequestException("404") + + with patch("requests.get", return_value=response): + with pytest.raises(RuntimeError, match="Failed to fetch release configuration"): + cmd._get_kolla_version_from_release("9.4.0", "https://example.com/release") + + +def test_get_kolla_version_from_release_raises_on_invalid_yaml(): + cmd = make_command(sync.Versions) + response = _release_response("\t") + + with patch("requests.get", return_value=response): + with pytest.raises(RuntimeError, match="Failed to parse release configuration"): + cmd._get_kolla_version_from_release("9.4.0", "https://example.com/release") + + +def test_get_kolla_version_from_release_raises_when_kolla_version_missing(): + cmd = make_command(sync.Versions) + response = _release_response("docker_images: {}\n") + + with patch("requests.get", return_value=response): + with pytest.raises(RuntimeError, match="Kolla version not found"): + cmd._get_kolla_version_from_release("9.4.0", "https://example.com/release") + + +# --- Versions.take_action / _sync_kolla_versions --- + + +def _run_versions( + tmp_path, args, *, sbom=None, extract=None, release=None, config=None +): + """Run ``sync versions`` with the SBOM extraction and release lookup stubbed.""" + config_path = str(config) if config is not None else str(tmp_path) + cmd, parsed_args = parse_args( + sync.Versions, ["--configuration-path", config_path, *args] + ) + + extract_kwargs = extract or {"return_value": sbom if sbom is not None else {}} + release_kwargs = release or {"return_value": "0.20250928.0"} + + with patch.object( + sync.Versions, "_extract_sbom_with_skopeo", **extract_kwargs + ) as mock_extract, patch.object( + sync.Versions, "_get_kolla_version_from_release", **release_kwargs + ) as mock_release: + result = cmd.take_action(parsed_args) + + return result, mock_extract, mock_release + + +def test_versions_release_derives_sbom_image_from_release_repository(tmp_path): + result, mock_extract, mock_release = _run_versions( + tmp_path, + [ + "--release", + "9.4.0", + "--release-repository-url", + "https://example.com/release/", + "--dry-run", + ], + ) + + # A trailing slash on the repository URL is stripped before the lookup. + mock_release.assert_called_once_with("9.4.0", "https://example.com/release") + mock_extract.assert_called_once_with( + "registry.osism.cloud/kolla/release/sbom:0.20250928.0" + ) + assert result == 0 + + +def test_versions_release_lookup_failure_returns_error(tmp_path, loguru_logs): + result, mock_extract, _ = _run_versions( + tmp_path, + ["--release", "9.4.0", "--dry-run"], + release={"side_effect": RuntimeError("Failed to fetch release configuration")}, + ) + + assert result == 1 + mock_extract.assert_not_called() + assert any( + record["level"] == "ERROR" + and "Failed to fetch release configuration" in record["message"] + for record in loguru_logs + ) + + +def test_versions_date_tag_uses_release_sbom_image(tmp_path): + _, mock_extract, _ = _run_versions( + tmp_path, ["--openstack-version", "0.20251128.0", "--dry-run"] + ) + mock_extract.assert_called_once_with( + "registry.osism.cloud/kolla/release/sbom:0.20251128.0" + ) + + +def test_versions_v_prefixed_date_tag_is_stripped(tmp_path): + _, mock_extract, _ = _run_versions( + tmp_path, ["--openstack-version", "v0.20251128.0", "--dry-run"] + ) + mock_extract.assert_called_once_with( + "registry.osism.cloud/kolla/release/sbom:0.20251128.0" + ) + + +def test_versions_openstack_version_uses_plain_sbom_image(tmp_path): + _, mock_extract, _ = _run_versions( + tmp_path, ["--openstack-version", "2025.1", "--dry-run"] + ) + mock_extract.assert_called_once_with("registry.osism.cloud/kolla/sbom:2025.1") + + +def test_versions_explicit_sbom_image_is_used_verbatim(tmp_path): + _, mock_extract, mock_release = _run_versions( + tmp_path, + [ + "--sbom-image", + "example.com/custom/sbom:1.2.3", + "--release", + "9.4.0", + "--dry-run", + ], + ) + mock_extract.assert_called_once_with("example.com/custom/sbom:1.2.3") + mock_release.assert_not_called() + + +def test_versions_missing_configuration_path_returns_error(tmp_path, loguru_logs): + result, mock_extract, _ = _run_versions(tmp_path, [], config=tmp_path / "missing") + + assert result == 1 + mock_extract.assert_not_called() + assert any( + record["level"] == "ERROR" + and "Configuration path does not exist" in record["message"] + for record in loguru_logs + ) + + +def test_versions_extraction_runtime_error_returns_error(tmp_path, loguru_logs): + result, _, _ = _run_versions( + tmp_path, + ["--dry-run"], + extract={"side_effect": RuntimeError("skopeo copy failed: denied")}, + ) + + assert result == 1 + assert any( + record["level"] == "ERROR" and "skopeo copy failed" in record["message"] + for record in loguru_logs + ) + + +def test_versions_extraction_yaml_error_returns_error(tmp_path, loguru_logs): + result, _, _ = _run_versions( + tmp_path, + ["--dry-run"], + extract={"side_effect": YAMLError("bad yaml")}, + ) + + assert result == 1 + assert any( + record["level"] == "ERROR" and "Failed to parse SBOM YAML" in record["message"] + for record in loguru_logs + ) + + +def test_versions_sbom_openstack_version_overrides_cli_value(tmp_path, capsys): + result, _, _ = _run_versions( + tmp_path, + ["--openstack-version", "2025.1", "--dry-run"], + sbom={"openstack_version": "2024.2", "versions": {}}, + ) + + out = capsys.readouterr().out + assert result == 0 + assert 'kolla_aodh_version: "2024.2"' in out + assert '"2025.1"' not in out + + +def test_versions_dry_run_prints_rendered_versions_without_writing(tmp_path, capsys): + result, _, _ = _run_versions( + tmp_path, + ["--dry-run"], + sbom={"openstack_version": "2025.1", "versions": {"aodh": "20.0.1"}}, + ) + + out = capsys.readouterr().out + assert result == 0 + assert 'kolla_aodh_version: "20.0.1"' in out + assert 'kolla_barbican_version: "2025.1"' in out + assert not (tmp_path / "environments").exists() + + +def test_versions_writes_versions_yml_and_creates_directories(tmp_path, loguru_logs): + result, _, _ = _run_versions( + tmp_path, + [], + sbom={"openstack_version": "2025.1", "versions": {"aodh": "20.0.1"}}, + ) + + output_path = tmp_path / "environments" / "kolla" / "versions.yml" + assert result == 0 + assert output_path.exists() + content = output_path.read_text() + assert content.startswith("---") + assert 'kolla_aodh_version: "20.0.1"' in content + assert 'kolla_barbican_version: "2025.1"' in content + assert any( + record["level"] == "SUCCESS" and "Versions written to" in record["message"] + for record in loguru_logs + ) + + +# --- Versions._extract_sbom_with_skopeo --- + + +def _tar_bytes(files): + """Build an uncompressed tar archive with the given ``{name: content}``.""" + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w") as tar: + for name, content in files.items(): + data = content.encode() + info = tarfile.TarInfo(name=name) + info.size = len(data) + tar.addfile(info, io.BytesIO(data)) + return buffer.getvalue() + + +def _build_oci_layout(tmp_path, layer_blobs): + """Create the OCI layout skopeo would have produced under ``tmp_path``. + + The code only ever splits digests on ``:`` to derive blob file names, so + placeholder digests are sufficient. + """ + blobs = tmp_path / "oci" / "blobs" / "sha256" + blobs.mkdir(parents=True) + + layers = [] + for index, blob in enumerate(layer_blobs): + (blobs / f"layer{index}").write_bytes(blob) + layers.append({"digest": f"sha256:layer{index}"}) + + (blobs / "manifest").write_text(json.dumps({"layers": layers})) + (tmp_path / "oci" / "index.json").write_text( + json.dumps({"manifests": [{"digest": "sha256:manifest"}]}) + ) + + +@contextlib.contextmanager +def _fake_tempdir(path): + yield str(path) + + +def _extract_sbom(tmp_path, *, run_kwargs=None): + """Call ``_extract_sbom_with_skopeo`` against a fake OCI layout.""" + cmd = make_command(sync.Versions) + + with patch( + "osism.commands.sync.tempfile.TemporaryDirectory", + return_value=_fake_tempdir(tmp_path), + ), patch("osism.commands.sync.subprocess.run", **(run_kwargs or {})) as mock_run: + sbom = cmd._extract_sbom_with_skopeo("registry.example.com/sbom:1") + + return sbom, mock_run + + +def test_extract_sbom_returns_parsed_images_yml(tmp_path): + layer = _tar_bytes({"images.yml": 'versions:\n aodh: "1.2.3"\n'}) + _build_oci_layout(tmp_path, [layer]) + + sbom, mock_run = _extract_sbom(tmp_path) + + assert sbom == {"versions": {"aodh": "1.2.3"}} + mock_run.assert_called_once() + command = mock_run.call_args[0][0] + assert command[:3] == ["skopeo", "copy", "docker://registry.example.com/sbom:1"] + assert command[3] == f"oci:{tmp_path / 'oci'}:latest" + assert mock_run.call_args[1]["check"] is True + + +def test_extract_sbom_skopeo_failure_raises_runtime_error(tmp_path): + with pytest.raises(RuntimeError, match="skopeo copy failed: denied"): + _extract_sbom( + tmp_path, + run_kwargs={ + "side_effect": subprocess.CalledProcessError( + 1, ["skopeo"], stderr="denied" + ) + }, + ) + + +def test_extract_sbom_missing_skopeo_binary_raises_runtime_error(tmp_path): + with pytest.raises(RuntimeError, match="skopeo not found"): + _extract_sbom(tmp_path, run_kwargs={"side_effect": FileNotFoundError}) + + +def test_extract_sbom_skips_layers_that_are_not_tar_archives(tmp_path): + layers = [ + b"this is not a tar archive", + _tar_bytes({"sbom/images.yml": "versions: {}\n"}), + ] + _build_oci_layout(tmp_path, layers) + + sbom, _ = _extract_sbom(tmp_path) + + assert sbom == {"versions": {}} + + +def test_extract_sbom_without_images_yml_raises_runtime_error(tmp_path): + _build_oci_layout(tmp_path, [_tar_bytes({"other.yml": "foo: bar\n"})]) + + with pytest.raises(RuntimeError, match="images.yml not found"): + _extract_sbom(tmp_path) diff --git a/tests/unit/commands/test_validate.py b/tests/unit/commands/test_validate.py index add9a1648..1b5da34d2 100644 --- a/tests/unit/commands/test_validate.py +++ b/tests/unit/commands/test_validate.py @@ -2,15 +2,29 @@ """Tests for the ``osism validate`` commands. -These focus on the exit-code contract: ``_handle_task`` is what ``take_action`` -returns, so a timeout while waiting for task output must yield a non-zero exit -status rather than an implicit ``None`` (exit 0). +Covered here: + +- the exit-code contract: ``_handle_task`` is what ``take_action`` returns, so + a timeout while waiting for task output must yield a non-zero exit status + rather than an implicit ``None`` (exit 0); +- ``Run.take_action``: dispatch of validators to the matching runtime + (kolla-ansible, ceph-ansible, osism-ansible), the ``--environment`` + override, and the task-lock check before any task is scheduled; +- ``Run._handle_task``: rc passthrough when waiting and the log/script output + modes when not waiting; +- ``Scs.take_action``: compliance-check command construction, error handling, + and cloud environment cleanup on every post-setup path. """ +from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + from osism.commands import validate +from ._helpers import assert_not_called_before_lock_check, parse_args + def test_handle_task_returns_nonzero_on_timeout(): cmd = validate.Run(MagicMock(), MagicMock()) @@ -25,3 +39,234 @@ def test_handle_task_returns_nonzero_on_timeout(): ) assert result == 1 + + +def _run_take_action(args): + """Drive ``Run.take_action`` with the task backends and helpers mocked.""" + cmd, parsed_args = parse_args(validate.Run, args) + + with patch("osism.tasks.ansible.run") as mock_ansible, patch( + "osism.tasks.ceph.run" + ) as mock_ceph, patch("osism.tasks.kolla.run") as mock_kolla, patch( + "osism.commands.validate.utils.check_task_lock_and_exit" + ) as mock_check, patch.object( + validate.Run, "_handle_task", return_value=0 + ) as mock_handle: + result = cmd.take_action(parsed_args) + + return SimpleNamespace( + result=result, + ansible=mock_ansible, + ceph=mock_ceph, + kolla=mock_kolla, + check=mock_check, + handle=mock_handle, + ) + + +def test_take_action_kolla_validator_appends_config_validate_action(): + mocks = _run_take_action(["keystone-config", "-e", "foo=1"]) + + mocks.kolla.delay.assert_called_once_with( + "kolla", "keystone", ["-e", "foo=1", "-e kolla_action=config_validate"] + ) + mocks.ansible.delay.assert_not_called() + mocks.ceph.delay.assert_not_called() + mocks.handle.assert_called_once_with( + mocks.kolla.delay.return_value, True, "log", 300, "keystone" + ) + assert mocks.result == 0 + + +@pytest.mark.parametrize( + ("args", "backend", "expected"), + [ + ( + ["--environment", "custom", "keystone-config"], + "kolla", + ("custom", "keystone", ["-e kolla_action=config_validate"]), + ), + ( + ["--environment", "custom", "ceph-config"], + "ceph", + ("custom", "validate", []), + ), + ], +) +def test_take_action_honors_environment_for_ceph_and_kolla(args, backend, expected): + mocks = _run_take_action(args) + + getattr(mocks, backend).delay.assert_called_once_with(*expected) + + +def test_take_action_ceph_validator_rewrites_playbook(): + mocks = _run_take_action(["ceph-config"]) + + mocks.ceph.delay.assert_called_once_with("ceph", "validate", []) + mocks.handle.assert_called_once_with( + mocks.ceph.delay.return_value, True, "log", 300, "validate" + ) + assert mocks.result == 0 + + +def test_take_action_osism_validator_derives_playbook_name(): + mocks = _run_take_action(["ntp"]) + + mocks.ansible.delay.assert_called_once_with("generic", "validate-ntp", []) + mocks.handle.assert_called_once_with( + mocks.ansible.delay.return_value, True, "log", 300, "validate-ntp" + ) + assert mocks.result == 0 + + +def test_take_action_checks_task_lock_before_scheduling(): + cmd, parsed_args = parse_args(validate.Run, ["ntp"]) + + with patch("osism.tasks.ansible.run") as mock_ansible, patch( + "osism.commands.validate.utils.check_task_lock_and_exit" + ) as mock_check, patch.object(validate.Run, "_handle_task", return_value=0): + mock_check.side_effect = assert_not_called_before_lock_check(mock_ansible.delay) + cmd.take_action(parsed_args) + + mock_check.assert_called_once() + + +def test_handle_task_passes_through_rc_when_waiting(): + cmd = validate.Run(MagicMock(), MagicMock()) + task = MagicMock() + + with patch( + "osism.commands.validate.utils.fetch_task_output", return_value=3 + ) as mock_fetch: + result = cmd._handle_task( + task, wait=True, format="log", timeout=5, playbook="validate-x" + ) + + mock_fetch.assert_called_once_with(task.id, timeout=5) + assert result == 3 + + +def test_handle_task_no_wait_log_format_logs_and_returns_zero(loguru_logs): + cmd = validate.Run(MagicMock(), MagicMock()) + task = MagicMock() + task.task_id = "taskid1" + + result = cmd._handle_task( + task, wait=False, format="log", timeout=5, playbook="validate-ntp" + ) + + assert result == 0 + assert any( + record["level"] == "INFO" + and "Task taskid1 (validate validate-ntp) is running in background" + in record["message"] + for record in loguru_logs + ) + + +def test_handle_task_no_wait_script_format_prints_task_id(capsys): + cmd = validate.Run(MagicMock(), MagicMock()) + task = MagicMock() + task.task_id = "taskid1" + + result = cmd._handle_task( + task, wait=False, format="script", timeout=5, playbook="validate-ntp" + ) + + assert result == 0 + assert capsys.readouterr().out == "taskid1\n" + + +def _run_scs(args, *, setup=("secret", ["/tmp/f1"], "/orig", True), run=None): + """Drive ``Scs.take_action`` with the cloud setup and subprocess mocked.""" + cmd, parsed_args = parse_args(validate.Scs, args) + + if run is None: + run = {"return_value": MagicMock(returncode=0)} + + with patch( + "osism.tasks.openstack.setup_cloud_environment", return_value=setup + ) as mock_setup, patch( + "osism.tasks.openstack.cleanup_cloud_environment" + ) as mock_cleanup, patch( + "osism.commands.validate.subprocess.run", **run + ) as mock_run: + result = cmd.take_action(parsed_args) + + return SimpleNamespace( + result=result, setup=mock_setup, cleanup=mock_cleanup, run=mock_run + ) + + +def _contains_slice(command, expected): + """Whether ``expected`` occurs in ``command`` as a contiguous slice.""" + width = len(expected) + return any( + command[i : i + width] == expected for i in range(len(command) - width + 1) + ) + + +def test_scs_returns_one_when_cloud_setup_fails(): + mocks = _run_scs([], setup=(None, [], "/orig", False)) + + assert mocks.result == 1 + mocks.run.assert_not_called() + mocks.cleanup.assert_not_called() + + +def test_scs_builds_command_and_passes_returncode_through(): + mocks = _run_scs( + ["--cloud", "mycloud", "--version", "v9.9"], + run={"return_value": MagicMock(returncode=7)}, + ) + + mocks.setup.assert_called_once_with("mycloud") + + args, kwargs = mocks.run.call_args + command = args[0] + assert command[:2] == ["python3", "/scs-tests/scs-compliance-check.py"] + assert _contains_slice(command, ["-s", "mycloud"]) + assert _contains_slice(command, ["-a", "os_cloud=mycloud"]) + assert _contains_slice(command, ["-V", "v9.9"]) + assert command[-1] == "scs-compatible-iaas.yaml" + assert kwargs["cwd"] == "/scs-tests/" + assert kwargs["check"] is False + assert kwargs["env"]["OS_CLIENT_CONFIG_FILE"] == "/tmp/clouds.yaml" + + mocks.cleanup.assert_called_once_with(["/tmp/f1"], "/orig") + assert mocks.result == 7 + + +@pytest.mark.parametrize( + ("args", "expected"), + [ + (["--verbose"], ["-v"]), + (["--debug"], ["--debug"]), + (["--tests", "scs-0100|scs-0103"], ["-t", "scs-0100|scs-0103"]), + (["--output", "report.yaml"], ["-o", "report.yaml"]), + (["--sections", "standard,iaas"], ["-S", "standard,iaas"]), + ], +) +def test_scs_optional_arguments_extend_command(args, expected): + mocks = _run_scs(args) + + command = mocks.run.call_args[0][0] + assert _contains_slice(command, expected) + + +def test_scs_returns_one_when_check_tool_is_missing(loguru_logs): + mocks = _run_scs([], run={"side_effect": FileNotFoundError}) + + assert mocks.result == 1 + assert any( + "SCS compliance check tool not found" in record["message"] + for record in loguru_logs + ) + mocks.cleanup.assert_called_once_with(["/tmp/f1"], "/orig") + + +def test_scs_returns_one_on_unexpected_subprocess_error(): + mocks = _run_scs([], run={"side_effect": RuntimeError("boom")}) + + assert mocks.result == 1 + mocks.cleanup.assert_called_once_with(["/tmp/f1"], "/orig") diff --git a/tests/unit/commands/test_wait.py b/tests/unit/commands/test_wait.py index 869042aa9..c941b3b87 100644 --- a/tests/unit/commands/test_wait.py +++ b/tests/unit/commands/test_wait.py @@ -13,8 +13,13 @@ The pre-fix code only returned an exit code under a ``len(task_ids) == 1`` guard, which never fired for a single task (so a timeout was ignored) and raised ``UnboundLocalError`` on a timeout with two tasks. + +The remaining tests characterize the non-``--live`` loop: task-id discovery +via the Celery inspect API, the PENDING/STARTED re-queue behaviour, the +``--output`` and ``--refresh`` options, and the script output format. """ +from types import SimpleNamespace from unittest.mock import MagicMock, patch from osism.commands import wait @@ -69,3 +74,164 @@ def test_live_returns_zero_when_task_succeeds(): fetch={"return_value": 0}, ) assert result == 0 + + +def _make_result(state, output=None): + """Build an ``AsyncResult`` stand-in reporting ``state``.""" + result = MagicMock() + result.state = state + if output is not None: + result.get.return_value = output + return result + + +def _run_states(args, *, results, query=None, scheduled=None, active=None): + """Drive ``take_action`` through the non-``--live`` loop. + + ``results`` is consumed one entry per ``AsyncResult`` construction, so + state transitions between loop iterations are modelled by consecutive + entries. ``query``, ``scheduled`` and ``active`` configure the mocked + Celery inspect API. + """ + cmd = wait.Run(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + + with patch("celery.Celery") as mock_celery, patch( + "celery.result.AsyncResult", side_effect=results + ) as mock_async, patch("osism.commands.wait.time.sleep") as mock_sleep, patch( + "osism.utils._init_redis", return_value=MagicMock() + ): + inspect = mock_celery.return_value.control.inspect.return_value + inspect.scheduled.return_value = scheduled if scheduled is not None else {} + inspect.active.return_value = active if active is not None else {} + if query is not None: + inspect.query_task.return_value = query + rc = cmd.take_action(parsed_args) + + return SimpleNamespace( + rc=rc, async_result=mock_async, sleep=mock_sleep, inspect=inspect + ) + + +def test_get_all_task_ids_merges_scheduled_and_active_sorted(): + cmd = wait.Run(MagicMock(), MagicMock()) + inspect = MagicMock() + inspect.scheduled.return_value = { + "worker1": [{"id": "task-c"}], + "worker2": [{"id": "task-a"}], + } + inspect.active.return_value = {"worker1": [{"id": "task-b"}]} + + assert cmd.get_all_task_ids(inspect) == ["task-a", "task-b", "task-c"] + + +def test_no_task_ids_on_cli_waits_for_all_running_tasks(loguru_logs): + mocks = _run_states( + [], + results=[_make_result("SUCCESS"), _make_result("SUCCESS")], + scheduled={"worker1": [{"id": "taskid2"}]}, + active={"worker1": [{"id": "taskid1"}]}, + ) + + assert mocks.rc == 0 + waited = [call.args[0] for call in mocks.async_result.call_args_list] + assert sorted(waited) == ["taskid1", "taskid2"] + assert any("No task IDs specified" in record["message"] for record in loguru_logs) + + +def test_pending_task_unknown_to_any_worker_is_not_requeued(loguru_logs): + mocks = _run_states( + ["taskid1"], + results=[_make_result("PENDING")], + query={"worker1": []}, + ) + + assert mocks.rc == 0 + assert mocks.async_result.call_count == 1 + mocks.inspect.query_task.assert_called_once_with("taskid1") + mocks.sleep.assert_not_called() + assert any( + record["message"] == "Task taskid1 is unavailable" for record in loguru_logs + ) + + +def test_pending_task_known_to_worker_is_requeued_until_success(loguru_logs): + mocks = _run_states( + ["taskid1"], + results=[_make_result("PENDING"), _make_result("SUCCESS")], + query={"worker1": [["taskid1", {}]]}, + ) + + assert mocks.rc == 0 + assert mocks.async_result.call_count == 2 + mocks.sleep.assert_called_once_with(1) + messages = [record["message"] for record in loguru_logs] + assert "Task taskid1 is in state PENDING" in messages + assert "Task taskid1 is in state SUCCESS" in messages + + +def test_success_with_output_prints_task_result(capsys): + mocks = _run_states( + ["taskid1", "--output"], + results=[_make_result("SUCCESS", output="task output")], + ) + + assert mocks.rc == 0 + assert "task output" in capsys.readouterr().out + + +def test_started_task_without_live_is_requeued_until_success(loguru_logs): + mocks = _run_states( + ["taskid1"], + results=[_make_result("STARTED"), _make_result("SUCCESS")], + ) + + assert mocks.rc == 0 + assert mocks.async_result.call_count == 2 + mocks.sleep.assert_called_once_with(1) + messages = [record["message"] for record in loguru_logs] + assert "Task taskid1 is in state STARTED" in messages + assert "Task taskid1 is in state SUCCESS" in messages + + +def test_refresh_consults_task_list_again_after_queue_drains(): + # NOTE: ``--refresh`` only takes effect when no task IDs are given on the + # command line: with explicit IDs ``do_refresh`` stays False and the loop + # exits as soon as the queue drains, so the refresh branch never runs. + cmd = wait.Run(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(["--refresh", "1"]) + + with patch("celery.Celery"), patch( + "celery.result.AsyncResult", side_effect=[_make_result("SUCCESS")] + ), patch("osism.commands.wait.time.sleep") as mock_sleep, patch( + "osism.utils._init_redis", return_value=MagicMock() + ), patch.object( + cmd, "get_all_task_ids", side_effect=[["taskid1"], []] + ) as mock_ids: + rc = cmd.take_action(parsed_args) + + assert rc == 0 + assert mock_ids.call_count == 2 + mock_sleep.assert_called_once_with(1) + + +def test_script_format_prints_state_lines_instead_of_log_output(capsys, loguru_logs): + mocks = _run_states( + ["taskid1", "--format", "script"], + results=[_make_result("SUCCESS")], + ) + + assert mocks.rc == 0 + assert capsys.readouterr().out == "taskid1 = SUCCESS\n" + assert not any("taskid1" in record["message"] for record in loguru_logs) + + +def test_script_format_prints_unavailable_for_unknown_pending_task(capsys): + mocks = _run_states( + ["taskid1", "--format", "script"], + results=[_make_result("PENDING")], + query={"worker1": []}, + ) + + assert mocks.rc == 0 + assert capsys.readouterr().out == "taskid1 = UNAVAILABLE\n"