Add unit tests for task lifecycle commands

osism/commands/configuration.py

@@ -8,11 +8,30 @@
 from osism import utils
 
 
+class _PassthroughParser(argparse.ArgumentParser):
+    """Argument parser that forwards unrecognized arguments to Ansible.
+
+    ``argparse.REMAINDER`` only starts capturing at the first *positional*
+    token, but everything forwarded to Ansible here is option-like (``-e``,
+    ``--limit``, ...) and there is no natural leading positional, so a
+    remainder would reject exactly the arguments it exists to forward.
+    Delegating ``parse_args()`` to ``parse_known_args()`` lets the parser
+    consume only its own options and forward everything else verbatim,
+    preserving order.
+    """
+
+    def parse_args(self, args=None, namespace=None):
+        namespace, arguments = self.parse_known_args(args, namespace)
+        namespace.arguments = arguments
+        return namespace
+
+
 class Sync(Command):
     def get_parser(self, prog_name):
-        parser = super(Sync, self).get_parser(prog_name)
-        parser.add_argument(
-            "arguments", nargs=argparse.REMAINDER, help="Other arguments for Ansible"
+        parser = _PassthroughParser(
+            prog=prog_name,
+            description=self.get_description(),
+            allow_abbrev=False,
         )
         return parser
 
@@ -35,5 +54,5 @@ def take_action(self, parsed_args):
             f"It takes a moment until task {t.task_id} (sync configuration) has been started and output is visible here."
         )
 
-        rc = handle_task(t, True, format, 60)
+        rc = handle_task(t, True, "log", 60)
         return rc

osism/commands/reconciler.py

@@ -15,6 +15,10 @@ def take_action(self, parsed_args):
         logger.info(
             "The osism reconciler command is deprecated and will be removed. Use osism service reconciler."
         )
+
+        # Check if tasks are locked before starting the worker
+        utils.check_task_lock_and_exit()
+
         # NOTE: use python interface in the future, works for the moment
         default_concurrency = min(multiprocessing.cpu_count(), 4)
         concurrency = int(

osism/commands/service.py

@@ -6,6 +6,7 @@
 import time
 
 from cliff.command import Command
+from loguru import logger
 
 from osism import utils
 
@@ -101,6 +102,10 @@ def take_action(self, parsed_args):
                 observer.stop()
                 observer.join()
 
+        else:
+            logger.error(f"Unknown service type: {service}")
+            return 1
+
     def watchdog_inventory_on_any_event(self, event):
         from osism.tasks import reconciler
 

osism/commands/task.py

@@ -13,7 +13,7 @@ def take_action(self, parsed_args):
         from celery import Celery
         from osism.tasks import Config
 
-        task = parsed_args.task
+        task = parsed_args.task[0]
 
         app = Celery("task")
         app.config_from_object(Config)

osism/commands/worker.py

@@ -5,6 +5,8 @@
 import subprocess
 
 from cliff.command import Command
+from loguru import logger
+
 from osism import utils
 
 
@@ -39,9 +41,11 @@ def take_action(self, parsed_args):
         elif queue == "osism-kubernetes":
             tasks = "kubernetes"
             queue = "kubernetes"
-        # kolla-ansible, ceph-ansible, osism-ansible
-        else:
+        elif queue in ["kolla-ansible", "ceph-ansible", "osism-ansible"]:
             tasks = queue[:-8]
+        else:
+            logger.error(f"Unknown worker type: {queue}")
+            return 1
 
         # NOTE: use python interface in the future, works for the moment
         if tasks == "osism":

tests/unit/commands/_helpers.py

@@ -0,0 +1,37 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared helpers for the ``osism.commands`` unit tests.
+
+The command tests all instantiate a cliff command, parse a small argument
+vector and - for the scheduling and service commands - assert that the task
+lock is consulted before any work is dispatched. These helpers centralize
+those recurring steps so the individual tests stay focused on behavior.
+"""
+
+from unittest.mock import MagicMock
+
+
+def make_command(command_class):
+    """Instantiate a cliff command with a mocked app and command namespace."""
+    return command_class(MagicMock(), MagicMock())
+
+
+def parse_args(command_class, args):
+    """Build ``command_class`` and parse ``args`` with its own parser.
+
+    Returns the ``(command, parsed_args)`` pair; parser-only tests discard the
+    command, while ``take_action`` tests drive it afterwards.
+    """
+    cmd = make_command(command_class)
+    return cmd, cmd.get_parser("test").parse_args(args)
+
+
+def assert_not_called_before_lock_check(guarded_mock):
+    """Return a ``check_task_lock_and_exit`` side effect that enforces ordering.
+
+    Scheduling and service commands must consult the task lock *before* they
+    schedule a Celery task or spawn a subprocess. Install the returned callable
+    as the mocked ``check_task_lock_and_exit`` side effect; it fails the test if
+    ``guarded_mock`` has already run by the time the lock is checked.
+    """
+    return lambda *args, **kwargs: guarded_mock.assert_not_called()

tests/unit/commands/test_configuration.py

@@ -0,0 +1,61 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism configuration sync`` command.
+
+``Sync`` schedules the ``configuration`` playbook on the manager environment
+and returns the task's exit code via ``handle_task``.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import configuration
+
+from ._helpers import assert_not_called_before_lock_check, parse_args
+
+
+def test_parser_forwards_mixed_positional_and_option_arguments():
+    _, parsed_args = parse_args(configuration.Sync, ["playbook", "-e", "foo=1"])
+    assert parsed_args.arguments == ["playbook", "-e", "foo=1"]
+
+
+def test_parser_forwards_leading_option_arguments():
+    # The parser delegates to ``parse_known_args`` (see ``_PassthroughParser``),
+    # so option-like tokens (``-e``, ``--limit``, ...) are forwarded verbatim
+    # even without a leading positional.
+    _, parsed_args = parse_args(configuration.Sync, ["-e", "foo=1"])
+    assert parsed_args.arguments == ["-e", "foo=1"]
+
+
+@pytest.mark.parametrize("rc", [0, 2])
+def test_take_action_schedules_configuration_sync_and_returns_rc(rc):
+    cmd, parsed_args = parse_args(configuration.Sync, ["playbook", "-e", "foo=1"])
+
+    with patch(
+        "osism.commands.configuration.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
+    ):
+        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",
+        "configuration",
+        ["playbook", "-e", "foo=1"],
+        auto_release_time=60,
+    )
+    assert result == rc
+
+
+def test_take_action_waits_with_log_format():
+    cmd, parsed_args = parse_args(configuration.Sync, [])
+
+    with patch("osism.commands.configuration.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, True, "log", 60)

tests/unit/commands/test_container.py

@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the ``osism container`` command.
+
+``Run`` executes ``docker`` commands on a remote host via SSH, either from
+the command line (one-shot) or from an interactive prompt loop.
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from osism.commands import container
+
+from ._helpers import parse_args
+
+
+def _run(args, *, prompt_side_effect=None, known_hosts=True):
+    cmd, parsed_args = parse_args(container.Run, args)
+
+    with patch("osism.commands.container.subprocess.call") as mock_call, patch(
+        "osism.commands.container.ensure_known_hosts_file", return_value=known_hosts
+    ), patch(
+        "osism.commands.container.prompt", side_effect=prompt_side_effect
+    ) as mock_prompt, patch(
+        "osism.commands.container.settings.OPERATOR_USER", "testuser"
+    ):
+        cmd.take_action(parsed_args)
+
+    return mock_call, mock_prompt
+
+
+def _ssh_command(mock_call):
+    """Flatten the ``subprocess.call`` invocation to a single command string.
+
+    The tests assert only the observable contract (ssh target and docker
+    command), so they hold whether the command is passed as a shell string or
+    as an argv vector.
+    """
+    command = mock_call.call_args[0][0]
+    return command if isinstance(command, str) else " ".join(command)
+
+
+def test_parser_splits_host_and_remainder_command():
+    _, parsed_args = parse_args(container.Run, ["node1", "ps", "-a"])
+    assert parsed_args.host == ["node1"]
+    assert parsed_args.command == ["ps", "-a"]
+
+
+def test_one_shot_command_runs_docker_via_ssh():
+    mock_call, mock_prompt = _run(["node1", "ps", "-a"])
+
+    mock_prompt.assert_not_called()
+    mock_call.assert_called_once()
+    ssh_command = _ssh_command(mock_call)
+    assert "testuser@node1" in ssh_command
+    assert ssh_command.endswith("docker ps -a")
+
+
+def test_known_hosts_failure_warns_but_still_connects(loguru_logs):
+    mock_call, _ = _run(["node1", "ps"], known_hosts=False)
+
+    assert any(
+        record["level"] == "WARNING" and "/share/known_hosts" in record["message"]
+        for record in loguru_logs
+    )
+    mock_call.assert_called_once()
+
+
+def test_interactive_prompt_runs_commands_until_exit():
+    mock_call, mock_prompt = _run(["node1"], prompt_side_effect=["ps", "exit"])
+
+    assert mock_prompt.call_count == 2
+    mock_call.assert_called_once()
+    ssh_command = _ssh_command(mock_call)
+    assert "testuser@node1" in ssh_command
+    assert ssh_command.endswith("docker ps")
+
+
+@pytest.mark.parametrize("keyword", ["Exit", "exit", "EXIT"])
+def test_interactive_exit_keywords_break_immediately(keyword):
+    mock_call, mock_prompt = _run(["node1"], prompt_side_effect=[keyword])
+
+    mock_prompt.assert_called_once()
+    mock_call.assert_not_called()