wrolpi/controller/test/test_systemd.py

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

from unittest import mock

import pytest

from controller.lib.systemd import (
    get_managed_services,
    get_service_config,
    get_service_status,
    discover_wrolpi_services,
    get_all_services_status,
    get_discovered_service_status,
    get_service_kind,
    classify_service_group,
    _get_systemd_name,
    start_service,
    stop_service,
    restart_service,
    enable_service,
    disable_service,
    get_service_logs,
    _run_systemctl,
)


class TestGetManagedServices:
    """Tests for get_managed_services function."""

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

    def test_returns_services_from_config(self):
        """Should return services from config."""
        result = get_managed_services()
        # Default config has managed_services
        assert len(result) > 0

    def test_each_service_has_name(self):
        """Each service should have a name."""
        result = get_managed_services()
        for service in result:
            assert "name" in service


class TestGetServiceConfig:
    """Tests for get_service_config function."""

    def test_returns_config_for_known_service(self):
        """Should return config for a known service."""
        # wrolpi-api is in default config
        result = get_service_config("wrolpi-api")
        assert result is not None
        assert result["name"] == "wrolpi-api"

    def test_returns_none_for_unknown_service(self):
        """Should return None for unknown service."""
        result = get_service_config("nonexistent-service")
        assert result is None


class TestRunSystemctl:
    """Tests for _run_systemctl function."""

    def test_returns_dict(self):
        """Should return a dictionary."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = _run_systemctl("is-active", "test.service")
            assert isinstance(result, dict)

    def test_success_on_zero_returncode(self):
        """Should return success=True when returncode is 0."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="done", stderr="")
            result = _run_systemctl("start", "test.service")
            assert result["success"] is True

    def test_failure_on_nonzero_returncode(self):
        """Should return success=False when returncode is non-zero."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=1, stdout="", stderr="failed")
            result = _run_systemctl("start", "test.service")
            assert result["success"] is False
            assert result["error"] == "failed"

    def test_handles_timeout(self):
        """Should handle subprocess timeout."""
        import subprocess
        with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("systemctl", 30)):
            result = _run_systemctl("start", "test.service")
            assert result["success"] is False
            assert "timed out" in result["error"].lower()

    def test_handles_file_not_found(self):
        """Should handle systemctl not found."""
        with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
            result = _run_systemctl("start", "test.service")
            assert result["success"] is False
            assert "not found" in result["error"].lower()


class TestGetServiceStatus:
    """Tests for get_service_status function."""

    def test_returns_error_for_unknown_service(self):
        """Should return error for unknown service."""
        result = get_service_status("nonexistent-service")
        assert "error" in result

    def test_returns_status_dict(self):
        """Should return a status dictionary."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = get_service_status("wrolpi-api")
            assert "name" in result
            assert "status" in result

    @pytest.mark.parametrize("systemd_state,expected_status", [
        ("active", "running"),
        ("activating", "running"),  # oneshot services during execution
        ("inactive", "stopped"),
        ("failed", "failed"),
    ])
    def test_maps_systemd_state_to_status(self, systemd_state, expected_status):
        """Should map systemd states to simple status values."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout=systemd_state, stderr="")
            result = get_service_status("wrolpi-api")
            assert result["status"] == expected_status

    @pytest.mark.parametrize("service_name,expected_https", [
        ("wrolpi-help", True),
        ("wrolpi-kiwix", True),
        ("wrolpi-api", False),
    ])
    def test_returns_use_https_field(self, service_name, expected_https):
        """Should return use_https field from service config."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = get_service_status(service_name)
            assert result.get("use_https", False) == expected_https


class TestGetAllServicesStatus:
    """Tests for get_all_services_status function."""

    def test_returns_list(self):
        """Should return a list."""
        with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
            with mock.patch("subprocess.run") as mock_run:
                mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
                result = get_all_services_status()
                assert isinstance(result, list)

    def test_returns_status_for_each_visible_service(self):
        """Should return status for each visible managed service."""
        with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
            with mock.patch("subprocess.run") as mock_run:
                mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
                result = get_all_services_status()
                managed = get_managed_services()
                # All services should be returned when they're all "running" (active)
                assert len(result) == len(managed)

    def test_includes_stopped_but_installed_services(self):
        """show_only_when_running services stay visible while stopped as long
        as the unit is installed: a finished oneshot (wrolpi-upgrade) must
        keep its logs reachable from the UI."""
        def side_effect(cmd, **kwargs):
            out = "loaded" if cmd[:2] == ["systemctl", "show"] else "inactive"
            return mock.Mock(returncode=0, stdout=out, stderr="")

        with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
            with mock.patch("subprocess.run", side_effect=side_effect):
                result = get_all_services_status()
                assert len(result) == len(get_managed_services())
                assert "wrolpi-upgrade" in [s["name"] for s in result]

    def test_hides_optional_services_only_when_not_installed(self):
        """show_only_when_running services are hidden only when the unit does
        not exist on this box (e.g. wrolpi-api-dev on production)."""
        def side_effect(cmd, **kwargs):
            out = "not-found" if cmd[:2] == ["systemctl", "show"] else "inactive"
            return mock.Mock(returncode=0, stdout=out, stderr="")

        with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
            with mock.patch("subprocess.run", side_effect=side_effect):
                result = get_all_services_status()
                managed = get_managed_services()
                always_visible = [s for s in managed if not s.get("show_only_when_running")]
                assert len(result) == len(always_visible)

    def test_includes_show_only_when_running_services_when_running(self):
        """Should include show_only_when_running services when they're running."""
        with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
            with mock.patch("subprocess.run") as mock_run:
                # Simulate all services as "active" (running)
                mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
                result = get_all_services_status()
                managed = get_managed_services()
                # All services should be visible when running
                assert len(result) == len(managed)
                # Verify wrolpi-upgrade is in the result
                service_names = [s["name"] for s in result]
                assert "wrolpi-upgrade" in service_names

    def test_includes_discovered_services_even_when_stopped(self):
        """Discovered wrolpi-* units (e.g. a finished script service) appear
        in the list regardless of running state."""
        def side_effect(cmd, **kwargs):
            out = "loaded" if cmd[:2] == ["systemctl", "show"] else "inactive"
            return mock.Mock(returncode=0, stdout=out, stderr="")

        with mock.patch("controller.lib.systemd.discover_wrolpi_services",
                        return_value=["wrolpi-custom-task"]):
            with mock.patch("subprocess.run", side_effect=side_effect):
                result = get_all_services_status()
                names = [s["name"] for s in result]
                assert "wrolpi-custom-task" in names


class TestDiscoverWrolpiServices:
    """Discovery unions loaded units (--all, catches transient units) with
    installed unit files, so exited oneshots and dev services are visible."""

    def _side_effect(self, list_units_out, list_unit_files_out):
        def side_effect(cmd, **kwargs):
            if cmd[:2] == ["systemctl", "list-units"]:
                out = list_units_out
            elif cmd[:2] == ["systemctl", "list-unit-files"]:
                out = list_unit_files_out
            else:
                out = ""
            return mock.Mock(returncode=0, stdout=out, stderr="")
        return side_effect

    def test_unions_loaded_and_installed_units(self):
        list_units = "wrolpi-transient.service loaded active running Task\n"
        list_unit_files = ("wrolpi-custom.service static -\n"
                           "wrolpi-api.service enabled enabled\n")
        with mock.patch("subprocess.run",
                        side_effect=self._side_effect(list_units, list_unit_files)):
            found = discover_wrolpi_services()
        assert "wrolpi-transient" in found
        assert "wrolpi-custom" in found
        # Managed services are excluded from discovery.
        assert "wrolpi-api" not in found

    def test_deduplicates_units_present_in_both_listings(self):
        list_units = "wrolpi-custom.service loaded inactive dead Task\n"
        list_unit_files = "wrolpi-custom.service static -\n"
        with mock.patch("subprocess.run",
                        side_effect=self._side_effect(list_units, list_unit_files)):
            found = discover_wrolpi_services()
        assert found.count("wrolpi-custom") == 1


class TestServiceActions:
    """Tests for service action functions (start, stop, restart, enable, disable)."""

    @pytest.mark.parametrize("func", [
        start_service,
        stop_service,
        restart_service,
        enable_service,
        disable_service,
    ])
    def test_returns_error_for_unknown_service(self, func):
        """Should return error for unknown service."""
        result = func("nonexistent-service")
        assert result["success"] is False

    @pytest.mark.parametrize("func,expected_action", [
        (start_service, "start"),
        (stop_service, "stop"),
        (restart_service, "restart"),
        (enable_service, "enable"),
        (disable_service, "disable"),
    ])
    def test_calls_correct_systemctl_action(self, func, expected_action):
        """Should call systemctl with correct action."""
        with mock.patch("controller.lib.systemd._run_systemctl") as mock_ctl:
            mock_ctl.return_value = {"success": True, "output": "", "error": None}
            result = func("wrolpi-api")
            mock_ctl.assert_called_once()
            assert mock_ctl.call_args[0][0] == expected_action
            assert result["success"] is True
            assert result["action"] == expected_action


class TestControllerSelfRestart:
    """Tests for controller self-restart behavior."""

    def test_controller_restart_uses_popen(self):
        """Controller restart should use Popen to avoid blocking on self-termination."""
        with mock.patch("subprocess.Popen") as mock_popen:
            result = restart_service("wrolpi-controller")
            mock_popen.assert_called_once_with(["systemctl", "restart", "wrolpi-controller"])
            assert result["success"] is True
            assert result["pending"] is True
            assert result["action"] == "restart"

    def test_controller_restart_handles_exception(self):
        """Controller restart should handle exceptions gracefully."""
        with mock.patch("subprocess.Popen", side_effect=Exception("popen failed")):
            result = restart_service("wrolpi-controller")
            assert result["success"] is False
            assert "popen failed" in result["error"]


class TestGetServiceLogs:
    """Tests for get_service_logs function."""

    def test_returns_error_for_unknown_service(self):
        """Should return error for unknown service."""
        result = get_service_logs("nonexistent-service")
        assert "error" in result

    def test_returns_logs(self):
        """Should return logs."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="log line 1\nlog line 2", stderr="")
            result = get_service_logs("wrolpi-api", lines=10)
            assert "logs" in result
            assert result["service"] == "wrolpi-api"
            assert result["lines"] == 10

    def test_handles_since_parameter(self):
        """Should include since parameter in journalctl command."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="logs", stderr="")
            result = get_service_logs("wrolpi-api", lines=50, since="1h")
            assert result["since"] == "1h"
            # Verify journalctl was called with --since
            call_args = mock_run.call_args[0][0]
            assert "--since" in call_args
            assert "1h" in call_args


class TestDiscoverRunningWrolpiServices:
    """Tests for discover_wrolpi_services function."""

    def test_discovers_unknown_services(self):
        """Should discover wrolpi-* services not in managed config."""
        systemctl_output = (
            "wrolpi-api.service                        loaded active running WROLPi API\n"
            "wrolpi-fix-media-permissions.service       loaded active running Fix permissions\n"
            "wrolpi-repair.service                      loaded active running WROLPi Repair\n"
        )
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout=systemctl_output, stderr="")
            result = discover_wrolpi_services()
            # wrolpi-api is managed, so only the other two should be discovered
            assert "wrolpi-api" not in result
            assert "wrolpi-fix-media-permissions" in result
            assert "wrolpi-repair" in result

    def test_returns_empty_on_failure(self):
        """Should return empty list when systemctl fails."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=1, stdout="", stderr="error")
            result = discover_wrolpi_services()
            assert result == []

    def test_returns_empty_on_no_systemctl(self):
        """Should return empty list when systemctl is not found."""
        with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
            result = discover_wrolpi_services()
            assert result == []

    def test_returns_empty_when_no_extra_services(self):
        """Should return empty list when all wrolpi-* services are managed."""
        systemctl_output = "wrolpi-api.service loaded active running WROLPi API\n"
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout=systemctl_output, stderr="")
            result = discover_wrolpi_services()
            assert result == []


class TestGetDiscoveredServiceStatus:
    """Tests for get_discovered_service_status function."""

    def test_returns_running_status(self):
        """Should return status for a discovered service."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = get_discovered_service_status("wrolpi-fix-media-permissions")
            assert result["name"] == "wrolpi-fix-media-permissions"
            assert result["status"] == "running"
            assert result["port"] is None
            assert result["viewable"] is False

    def test_handles_no_systemctl(self):
        """Should handle missing systemctl."""
        with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
            result = get_discovered_service_status("wrolpi-fix-media-permissions")
            assert result["status"] == "unknown"
            assert "error" in result


class TestGetSystemdName:
    """Tests for _get_systemd_name helper."""

    def test_returns_systemd_name_for_managed_service(self):
        """Should return systemd_name from config for managed services."""
        result = _get_systemd_name("wrolpi-api")
        assert result == "wrolpi-api"

    def test_returns_name_for_discovered_wrolpi_service(self):
        """Should return the name itself for discovered wrolpi-* services."""
        result = _get_systemd_name("wrolpi-fix-media-permissions")
        assert result == "wrolpi-fix-media-permissions"

    def test_returns_none_for_unknown_non_wrolpi_service(self):
        """Should return None for unknown non-wrolpi services."""
        result = _get_systemd_name("some-random-service")
        assert result is None


class TestGetAllServicesStatusWithDiscovery:
    """Tests for get_all_services_status including dynamic discovery."""

    def test_includes_discovered_running_services(self):
        """Should include dynamically discovered running wrolpi-* services."""
        systemctl_list_output = (
            "wrolpi-fix-media-permissions.service loaded active running Fix permissions\n"
        )

        def mock_run_side_effect(cmd, **kwargs):
            if cmd[0] == "systemctl" and cmd[1] == "list-units":
                return mock.Mock(returncode=0, stdout=systemctl_list_output, stderr="")
            # For is-active / is-enabled calls
            return mock.Mock(returncode=0, stdout="active", stderr="")

        with mock.patch("subprocess.run", side_effect=mock_run_side_effect):
            result = get_all_services_status()
            service_names = [s["name"] for s in result]
            assert "wrolpi-fix-media-permissions" in service_names

    def test_does_not_duplicate_managed_services(self):
        """Discovered services should not duplicate managed services."""
        systemctl_list_output = (
            "wrolpi-api.service loaded active running WROLPi API\n"
        )

        def mock_run_side_effect(cmd, **kwargs):
            if cmd[0] == "systemctl" and cmd[1] == "list-units":
                return mock.Mock(returncode=0, stdout=systemctl_list_output, stderr="")
            return mock.Mock(returncode=0, stdout="active", stderr="")

        with mock.patch("subprocess.run", side_effect=mock_run_side_effect):
            result = get_all_services_status()
            api_count = sum(1 for s in result if s["name"] == "wrolpi-api")
            assert api_count == 1


class TestActionsOnDiscoveredServices:
    """Tests that service actions work on discovered wrolpi-* services."""

    @pytest.mark.parametrize("func,expected_action", [
        (start_service, "start"),
        (stop_service, "stop"),
        (restart_service, "restart"),
        (enable_service, "enable"),
        (disable_service, "disable"),
    ])
    def test_actions_work_on_discovered_services(self, func, expected_action):
        """Should allow actions on discovered wrolpi-* services."""
        with mock.patch("controller.lib.systemd._run_systemctl") as mock_ctl:
            mock_ctl.return_value = {"success": True, "output": "", "error": None}
            result = func("wrolpi-fix-media-permissions")
            assert result["success"] is True
            assert result["action"] == expected_action

    def test_logs_work_on_discovered_services(self):
        """Should return logs for discovered wrolpi-* services."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="log output", stderr="")
            result = get_service_logs("wrolpi-fix-media-permissions", lines=10)
            assert result["service"] == "wrolpi-fix-media-permissions"
            assert "logs" in result

    def test_get_service_status_for_discovered_service(self):
        """get_service_status should work for discovered wrolpi-* services."""
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = get_service_status("wrolpi-fix-media-permissions")
            assert result["name"] == "wrolpi-fix-media-permissions"
            assert result["status"] == "running"


class TestGetServiceKind:
    """Tests for get_service_kind classification."""

    def test_explicit_kind_wins(self):
        assert get_service_kind({"kind": "task"}) == "task"
        assert get_service_kind({"kind": "persistent"}) == "persistent"

    def test_show_only_when_running_is_a_task(self):
        """Dev/upgrade units (show_only_when_running) default to 'task'."""
        assert get_service_kind({"show_only_when_running": True}) == "task"

    def test_default_is_persistent(self):
        """A plain managed service is a persistent, always-on service."""
        assert get_service_kind({}) == "persistent"
        assert get_service_kind({"name": "wrolpi-api"}) == "persistent"


class TestClassifyServiceGroup:
    """Tests for classify_service_group."""

    def test_persistent_and_enabled_is_core(self):
        assert classify_service_group("persistent", True) == "core"

    def test_persistent_but_disabled_is_optional(self):
        """A persistent service the user disabled at boot (e.g. Samba with no
        shares) drops to the Optional & Maintenance group."""
        assert classify_service_group("persistent", False) == "optional"

    def test_task_is_always_optional(self):
        """One-off/dev/maintenance tasks are optional regardless of boot state."""
        assert classify_service_group("task", True) == "optional"
        assert classify_service_group("task", False) == "optional"


class TestServiceStatusGroup:
    """Group/kind are exposed on the status dicts the UI consumes."""

    def test_enabled_persistent_service_is_core(self):
        # is-active -> running, is-enabled -> enabled, LoadState -> loaded
        with mock.patch("subprocess.run") as mock_run:
            mock_run.side_effect = [
                mock.Mock(returncode=0, stdout="active", stderr=""),
                mock.Mock(returncode=0, stdout="enabled", stderr=""),
                mock.Mock(returncode=0, stdout="loaded", stderr=""),
            ]
            result = get_service_status("wrolpi-api")
            assert result["kind"] == "persistent"
            assert result["enabled"] is True
            assert result["group"] == "core"

    def test_disabled_persistent_service_is_optional(self):
        with mock.patch("subprocess.run") as mock_run:
            mock_run.side_effect = [
                mock.Mock(returncode=0, stdout="inactive", stderr=""),
                mock.Mock(returncode=0, stdout="disabled", stderr=""),
                mock.Mock(returncode=0, stdout="loaded", stderr=""),
            ]
            result = get_service_status("smbd")
            assert result["kind"] == "persistent"
            assert result["enabled"] is False
            assert result["group"] == "optional"

    def test_discovered_service_is_optional_task(self):
        with mock.patch("subprocess.run") as mock_run:
            mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
            result = get_discovered_service_status("wrolpi-repair")
            assert result["kind"] == "task"
            assert result["group"] == "optional"