grm/tests/unit/test_run_molecule_parallel.py

"""Unit tests for scripts/run_molecule_parallel.py."""

from __future__ import annotations

import os
import subprocess  # nosec B404
from pathlib import Path
from unittest.mock import MagicMock, patch

from scripts.run_molecule_parallel import cli, run_pair


class TestRunPair:
    def test_default_scenario(self, tmp_path: Path) -> None:
        with (
            patch("scripts.run_molecule_parallel.subprocess.Popen") as mock_popen,
            patch.dict(os.environ, {"MOLECULE_PLATFORM_COMMAND": "old-command"}, clear=False),
        ):
            mock_popen.return_value = MagicMock()
            result = run_pair("default|ubuntu-2204|img:latest|", tmp_path)
            assert result is mock_popen.return_value
            call_args = mock_popen.call_args
            assert call_args.kwargs["cwd"] == str(tmp_path)
            env = call_args.kwargs["env"]
            assert env["MOLECULE_PLATFORM_NAME"] == "ubuntu-2204"
            assert env["MOLECULE_PLATFORM_IMAGE"] == "img:latest"
            assert "MOLECULE_PLATFORM_COMMAND" not in env
            assert call_args.args[0] == ["molecule", "test"]

    def test_named_scenario_with_command(self, tmp_path: Path) -> None:
        with patch("scripts.run_molecule_parallel.subprocess.Popen") as mock_popen:
            mock_popen.return_value = MagicMock()
            result = run_pair("lifecycle|archlinux|img:arch|/usr/lib/systemd/systemd", tmp_path)
            assert result is mock_popen.return_value
            call_args = mock_popen.call_args
            env = call_args.kwargs["env"]
            assert env["MOLECULE_PLATFORM_NAME"] == "archlinux"
            assert env["MOLECULE_PLATFORM_IMAGE"] == "img:arch"
            assert env["MOLECULE_PLATFORM_COMMAND"] == "/usr/lib/systemd/systemd"
            assert call_args.args[0] == ["molecule", "test", "-s", "lifecycle"]


class TestCli:
    def test_all_pass(self, tmp_path: Path) -> None:
        from click.testing import CliRunner

        with patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair:
            mock_run_pair.return_value = MagicMock()
            mock_run_pair.return_value.poll.side_effect = [None, 0]
            mock_run_pair.return_value.wait.return_value = None

            runner = CliRunner()
            result = runner.invoke(cli, ["default|ubuntu-2204|img:latest|"])
            assert result.exit_code == 0
            assert "All molecule tests passed" in result.output

    def test_failure_kills_remaining(self, tmp_path: Path) -> None:
        from click.testing import CliRunner

        with (
            patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
            patch("os.killpg") as mock_killpg,
            patch("os.getpgid") as mock_getpgid,
        ):
            mock_getpgid.return_value = 123

            good_proc = MagicMock()
            good_proc.poll.return_value = None
            good_proc.wait.return_value = None
            bad_proc = MagicMock()
            bad_proc.poll.side_effect = [None, 1, 1, 1]
            bad_proc.wait.return_value = None
            mock_run_pair.side_effect = [bad_proc, good_proc]

            runner = CliRunner()
            result = runner.invoke(
                cli,
                [
                    "default|ubuntu-2204|img:latest|",
                    "lifecycle|debian-12|img:deb|",
                ],
            )
            assert result.exit_code == 1
            assert "FAILURE" in result.output
            mock_killpg.assert_called()

    def test_failure_sends_sigkill_when_sigterm_times_out(self, tmp_path: Path) -> None:
        from click.testing import CliRunner

        with (
            patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
            patch("os.killpg") as mock_killpg,
            patch("os.getpgid") as mock_getpgid,
        ):
            mock_getpgid.return_value = 123
            # First SIGTERM succeeds, second SIGKILL raises ProcessLookupError
            mock_killpg.side_effect = [None, ProcessLookupError("no such process")]

            good_proc = MagicMock()
            good_proc.poll.return_value = None
            good_proc.wait.side_effect = [None, subprocess.TimeoutExpired("cmd", 5)]
            bad_proc = MagicMock()
            bad_proc.poll.side_effect = [None, 1, 1, 1]
            bad_proc.wait.return_value = None
            mock_run_pair.side_effect = [bad_proc, good_proc]

            runner = CliRunner()
            result = runner.invoke(
                cli,
                [
                    "default|ubuntu-2204|img:latest|",
                    "lifecycle|debian-12|img:deb|",
                ],
            )
            assert result.exit_code == 1
            assert "FAILURE" in result.output
            # First SIGTERM, then SIGKILL
            assert mock_killpg.call_count >= 2

    def test_failure_handles_process_lookup_error(self, tmp_path: Path) -> None:
        from click.testing import CliRunner

        with (
            patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
            patch("os.killpg") as mock_killpg,
            patch("os.getpgid") as mock_getpgid,
        ):
            mock_getpgid.return_value = 123
            mock_killpg.side_effect = ProcessLookupError("no such process")

            good_proc = MagicMock()
            good_proc.poll.return_value = None
            good_proc.wait.return_value = None
            bad_proc = MagicMock()
            bad_proc.poll.side_effect = [None, 1, 1, 1]
            bad_proc.wait.return_value = None
            mock_run_pair.side_effect = [bad_proc, good_proc]

            runner = CliRunner()
            result = runner.invoke(
                cli,
                [
                    "default|ubuntu-2204|img:latest|",
                    "lifecycle|debian-12|img:deb|",
                ],
            )
            assert result.exit_code == 1
            assert "FAILURE" in result.output


def test_main_module_block() -> None:
    import scripts.run_molecule_parallel as rm

    with open(rm.__file__) as f:
        source = f.read()
    source = source.replace('if __name__ == "__main__":\n    cli()\n', "")
    namespace = dict(rm.__dict__)
    exec(compile(source, rm.__file__, "exec"), namespace)
    assert callable(namespace["cli"])