wrolpi/controller/test/test_scripts.py

"""
Unit tests for controller.lib.scripts module.
"""

import subprocess
import time
from unittest import mock

from controller.lib.scripts import (
    _get_service_timing,
    AVAILABLE_SCRIPTS,
    get_current_branch,
    get_script_output,
    get_script_status,
    list_available_scripts,
    start_script,
)


class TestGetCurrentBranch:
    """Tests for get_current_branch function."""

    def test_returns_branch_name(self):
        """Should return the branch name from `git branch --show-current`.

        `git rev-parse --abbrev-ref HEAD` must not be used: it reports "heads/release" when a
        stale `release` tag exists alongside the `release` branch.
        """
        mock_result = mock.MagicMock()
        mock_result.returncode = 0
        mock_result.stdout = "release\n"

        with mock.patch("subprocess.run", return_value=mock_result) as mock_run:
            assert get_current_branch() == "release"
        cmd = mock_run.call_args[0][0]
        assert "--show-current" in cmd
        assert "--abbrev-ref" not in cmd

    def test_detached_head_defaults_to_release(self):
        """A detached HEAD (empty output) should default to 'release'."""
        mock_result = mock.MagicMock()
        mock_result.returncode = 0
        mock_result.stdout = "\n"

        with mock.patch("subprocess.run", return_value=mock_result):
            assert get_current_branch() == "release"

    def test_git_failure_returns_none(self):
        """A failing git command should return None."""
        mock_result = mock.MagicMock()
        mock_result.returncode = 128
        mock_result.stdout = ""

        with mock.patch("subprocess.run", return_value=mock_result):
            assert get_current_branch() is None


class TestAvailableScripts:
    """Tests for AVAILABLE_SCRIPTS constant."""

    def test_repair_script_exists(self):
        """Should have a repair script defined."""
        assert "repair" in AVAILABLE_SCRIPTS

    def test_repair_script_has_required_fields(self):
        """Repair script should have all required fields."""
        repair = AVAILABLE_SCRIPTS["repair"]
        assert "name" in repair
        assert "display_name" in repair
        assert "description" in repair
        assert "service_name" in repair
        assert "warnings" in repair

    def test_repair_script_has_warnings(self):
        """Repair script should have warnings about its effects."""
        repair = AVAILABLE_SCRIPTS["repair"]
        assert len(repair["warnings"]) > 0

    def test_help_script_exists(self):
        """Should have a help (diagnostics) script defined."""
        assert "help" in AVAILABLE_SCRIPTS
        help_script = AVAILABLE_SCRIPTS["help"]
        assert help_script["service_name"] == "wrolpi-help-script.service"
        # help.sh is read-only, so it needs no warnings.
        assert help_script["warnings"] == []


class TestListAvailableScripts:
    """Tests for list_available_scripts function."""

    def test_returns_list(self):
        """Should return a list."""
        result = list_available_scripts()
        assert isinstance(result, list)

    def test_returns_scripts_with_required_fields(self):
        """Each script should have required fields."""
        result = list_available_scripts()
        for script in result:
            assert "name" in script
            assert "display_name" in script
            assert "description" in script
            assert "warnings" in script
            assert "available" in script

    def test_scripts_unavailable_in_docker_mode(self):
        """Scripts should be marked unavailable in Docker mode."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=True):
            result = list_available_scripts()
            for script in result:
                assert script["available"] is False

    def test_scripts_available_when_not_docker(self):
        """Scripts should be available when not in Docker mode."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            result = list_available_scripts()
            for script in result:
                assert script["available"] is True


class TestGetScriptStatus:
    """Tests for get_script_status function."""

    def test_returns_dict(self):
        """Should return a dict."""
        result = get_script_status()
        assert isinstance(result, dict)

    def test_has_running_field(self):
        """Should always have running field."""
        result = get_script_status()
        assert "running" in result

    def test_returns_not_running_in_docker_mode(self):
        """Should return not running in Docker mode."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=True):
            result = get_script_status()
            assert result["running"] is False

    def test_handles_systemctl_not_found(self):
        """Should handle systemctl not being available."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
                result = get_script_status()
                assert result["running"] is False

    def test_handles_timeout(self):
        """Should handle subprocess timeout."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("systemctl", 5)):
                result = get_script_status()
                assert result["running"] is False

    def test_returns_running_when_service_activating(self):
        """Should return running when service is activating."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            mock_result = mock.Mock()
            mock_result.stdout = "activating\n"
            mock_result.returncode = 0
            with mock.patch("subprocess.run", return_value=mock_result):
                with mock.patch("controller.lib.scripts._get_service_timing", return_value=(None, None)):
                    result = get_script_status()
                    assert result["running"] is True
                    # The first script checked reports as running because every systemctl call is mocked.
                    assert result["script_name"] == next(iter(AVAILABLE_SCRIPTS))

    def test_returns_not_running_when_service_inactive(self):
        """Should return not running when service is inactive."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            mock_result = mock.Mock()
            mock_result.stdout = "inactive\n"
            mock_result.returncode = 3
            with mock.patch("subprocess.run", return_value=mock_result):
                result = get_script_status()
                assert result["running"] is False


class TestStartScript:
    """Tests for start_script function."""

    def test_returns_error_in_docker_mode(self):
        """Should return error in Docker mode."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=True):
            result = start_script("repair")
            assert result["success"] is False
            assert "Docker" in result.get("error", "")

    def test_returns_error_for_unknown_script(self):
        """Should return error for unknown script."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            result = start_script("nonexistent")
            assert result["success"] is False
            assert "Unknown" in result.get("error", "")

    def test_returns_error_if_already_running(self):
        """Should return error if script is already running."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("controller.lib.scripts.get_script_status", return_value={
                "running": True,
                "script_name": "repair",
            }):
                result = start_script("repair")
                assert result["success"] is False
                assert "already running" in result.get("error", "").lower()

    def test_returns_error_if_another_script_running(self):
        """Should return error if another script is running."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("controller.lib.scripts.get_script_status", return_value={
                "running": True,
                "script_name": "other",
            }):
                result = start_script("repair")
                assert result["success"] is False
                assert "another script" in result.get("error", "").lower()

    def test_handles_systemctl_not_found(self):
        """Should handle systemctl not being available."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("controller.lib.scripts.get_script_status", return_value={"running": False}):
                with mock.patch("subprocess.Popen", side_effect=FileNotFoundError()):
                    result = start_script("repair")
                    assert result["success"] is False
                    assert "not found" in result.get("error", "").lower()

    def test_starts_service_successfully(self):
        """Should start service successfully."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("controller.lib.scripts.get_script_status", return_value={"running": False}):
                mock_popen = mock.Mock()
                with mock.patch("subprocess.Popen", return_value=mock_popen) as mock_p:
                    result = start_script("repair")
                    assert result["success"] is True
                    assert "started" in result.get("message", "").lower()
                    # Verify correct command was called
                    mock_p.assert_called_once()
                    call_args = mock_p.call_args[0][0]
                    assert "systemctl" in call_args
                    assert "start" in call_args
                    assert "wrolpi-repair.service" in call_args


class TestGetScriptOutput:
    """Tests for get_script_output function."""

    def test_returns_docker_message_in_docker_mode(self):
        """Should return Docker not available message in Docker mode."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=True):
            result = get_script_output("repair")
            assert "Docker" in result.get("output", "")

    def test_returns_error_for_unknown_script(self):
        """Should return error for unknown script."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            result = get_script_output("nonexistent")
            assert "Unknown" in result.get("output", "")

    def test_handles_journalctl_not_found(self):
        """Should handle journalctl not being available."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
                result = get_script_output("repair")
                assert "not found" in result.get("output", "").lower()

    def test_handles_timeout(self):
        """Should handle subprocess timeout."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("journalctl", 10)):
                result = get_script_output("repair")
                assert "Timeout" in result.get("output", "")

    def test_returns_log_output(self):
        """Should return log output from journalctl."""
        with mock.patch("controller.lib.scripts.is_docker_mode", return_value=False):
            mock_result = mock.Mock()
            mock_result.stdout = "Log line 1\nLog line 2\n"
            mock_result.returncode = 0
            with mock.patch("subprocess.run", return_value=mock_result) as mock_run:
                result = get_script_output("repair", lines=50)
                assert "Log line 1" in result.get("output", "")
                assert result["lines"] == 50
                assert result["script_name"] == "repair"
                # Verify correct command
                call_args = mock_run.call_args[0][0]
                assert "journalctl" in call_args
                assert "-u" in call_args
                assert "wrolpi-repair.service" in call_args
                assert "-n" in call_args
                assert "50" in call_args


class TestGetServiceTiming:

    def test_elapsed_is_independent_of_local_timezone(self):
        """systemd prints timestamps in local time; elapsed must not be off by the UTC offset."""
        started = time.time() - 90
        completed = subprocess.CompletedProcess(
            args=[], returncode=0, stdout=f"ExecMainStartTimestamp=@{started:.6f}\n")
        with mock.patch("controller.lib.scripts.subprocess.run", return_value=completed) as run:
            started_at, elapsed = _get_service_timing("wrolpi-upgrade.service")
        assert "--timestamp=unix" in run.call_args.args[0]
        assert 89 <= elapsed <= 91
        assert started_at.endswith("+00:00")

    def test_not_started(self):
        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="ExecMainStartTimestamp=\n")
        with mock.patch("controller.lib.scripts.subprocess.run", return_value=completed):
            assert _get_service_timing("wrolpi-upgrade.service") == (None, None)