LEDMatrix/test/test_api_v3_on_demand_restart.py

"""POST /display/on-demand/start and /stop must not restart a running display.

The start route used to treat ``start_service`` (default True, and what both
the web UI and the MQTT bridge send) as "restart": with the service running it
ran ``systemctl stop``, slept 1.5s and started it again. Every on-demand or
"Preview on display" click therefore cold-restarted the display process --
every plugin reloaded, the panel blank for seconds -- to deliver a request the
running process polls for every ON_DEMAND_POLL_INTERVAL anyway (see
test_on_demand_mailbox.py and test_display_pending_changes.py for the display
side: the mailbox is read mid-dwell, mid-screen and mid-Vegas-iteration).

The restart did not buy anything either: a freshly started display restores
only the on-demand session it saved itself (``display_on_demand_config``), so
the new request reached it through the same mailbox, one cold start later.

This file previously pinned that restart path (it guarded a broken
``import _pkg.time`` inside it). The path is gone; these tests pin its
replacement: a running service is left alone, a stopped one is started (only
when start_service is set), and the request lands in the mailbox either way.

The service helpers are patched where they run. display.py binds
_get_display_service_status by value, while _ensure_display_service_running
(in the package __init__) looks it up in its own module, so both are patched;
_run_systemctl_command is the one place a systemctl command is issued.
"""

import sys
from pathlib import Path
from unittest.mock import patch

import pytest

sys.path.insert(0, str(Path(__file__).parent.parent))

from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402

START_URL = "/api/v3/display/on-demand/start"
STOP_URL = "/api/v3/display/on-demand/stop"
MAILBOX = "display_on_demand_request"


@pytest.fixture
def service(api_v3_module):
    """A display service whose state the test sets; records systemctl calls.

    plugin_manager and config_manager are None so the route skips plugin
    resolution (not what is under test here). The cache is the blueprint's
    MagicMock cache_manager, so mailbox writes are visible as set() calls.
    """
    api_v3_module.api_v3.plugin_manager = None
    api_v3_module.api_v3.config_manager = None
    state = {"active": True}

    def status():
        return {"active": state["active"]}

    def systemctl(args):
        if args[-2:] == ["start", "ledmatrix.service"]:
            state["active"] = True
        elif args[-2:] == ["stop", "ledmatrix.service"]:
            state["active"] = False
        return {"returncode": 0, "stdout": "", "stderr": ""}

    with patch("web_interface.blueprints.api_v3._get_display_service_status",
               side_effect=status), \
         patch("web_interface.blueprints.api_v3.display._get_display_service_status",
               side_effect=status), \
         patch("web_interface.blueprints.api_v3._run_systemctl_command",
               side_effect=systemctl) as run_systemctl, \
         patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop_service:
        yield {
            "state": state,
            "systemctl": run_systemctl,
            "stop_service": stop_service,
            "cache": api_v3_module.api_v3.cache_manager,
        }


def _mailbox_writes(cache):
    return [c.args[1] for c in cache.set.call_args_list if c.args and c.args[0] == MAILBOX]


def _systemctl_verbs(run_systemctl):
    return [c.args[0][-2] for c in run_systemctl.call_args_list]


class TestStartWhileTheServiceIsRunning:
    @pytest.mark.parametrize("body", [
        {"plugin_id": "weather"},                           # "Preview on display", MQTT
        {"plugin_id": "weather", "start_service": True},    # on-demand modal, box ticked
        {"plugin_id": "weather", "start_service": "true"},
    ])
    def test_the_service_is_not_stopped_or_restarted(self, api_v3_client, service, body):
        response = api_v3_client.post(START_URL, json=body)
        assert response.status_code == 200, response.get_json()
        assert response.get_json()["status"] == "success"
        service["stop_service"].assert_not_called()
        assert _systemctl_verbs(service["systemctl"]) == [], (
            "a running display service was sent a systemctl command")

    def test_the_request_is_posted_for_the_running_display(self, api_v3_client, service):
        response = api_v3_client.post(
            START_URL, json={"plugin_id": "weather", "mode": "weather_current",
                             "duration": 60, "pinned": True})
        data = response.get_json()["data"]
        writes = _mailbox_writes(service["cache"])
        assert len(writes) == 1
        assert writes[0]["action"] == "start"
        assert writes[0]["request_id"] == data["request_id"]
        assert writes[0]["plugin_id"] == "weather"
        assert writes[0]["mode"] == "weather_current"
        assert writes[0]["duration"] == 60
        assert writes[0]["pinned"] is True

    def test_the_response_reports_the_service_was_not_started(self, api_v3_client, service):
        data = api_v3_client.post(START_URL, json={"plugin_id": "weather"}).get_json()["data"]
        assert data["service"]["active"] is True
        assert data["service"]["started"] is False

    def test_it_answers_without_the_old_restart_pause(self, api_v3_client, service):
        # The restart slept 1.5s; nothing here should sleep at all.
        with patch("time.sleep") as sleep:
            api_v3_client.post(START_URL, json={"plugin_id": "weather"})
        sleep.assert_not_called()


class TestStartWhileTheServiceIsStopped:
    def test_start_service_starts_it_once_and_never_stops_it(self, api_v3_client, service):
        service["state"]["active"] = False
        response = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
        assert response.status_code == 200, response.get_json()
        assert _systemctl_verbs(service["systemctl"]) == ["start"]
        service["stop_service"].assert_not_called()
        # Written before the start, so the new process finds it on its first poll.
        assert len(_mailbox_writes(service["cache"])) == 1

    def test_without_start_service_it_is_left_stopped(self, api_v3_client, service):
        service["state"]["active"] = False
        response = api_v3_client.post(
            START_URL, json={"plugin_id": "weather", "start_service": "false"})
        assert response.status_code == 400
        assert _systemctl_verbs(service["systemctl"]) == []

    def test_a_start_that_fails_is_reported(self, api_v3_client, service):
        service["state"]["active"] = False
        service["systemctl"].side_effect = lambda args: {
            "returncode": 1, "stdout": "", "stderr": "denied"}
        response = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
        assert response.status_code == 500
        assert response.get_json()["status"] == "error"


class TestStop:
    def test_stop_posts_a_stop_request_and_leaves_the_service_running(
            self, api_v3_client, service):
        response = api_v3_client.post(STOP_URL, json={})
        assert response.status_code == 200, response.get_json()
        writes = _mailbox_writes(service["cache"])
        assert [w["action"] for w in writes] == ["stop"]
        service["stop_service"].assert_not_called()
        assert _systemctl_verbs(service["systemctl"]) == []

    def test_a_string_false_stop_service_does_not_stop_it(self, api_v3_client, service):
        # bool("false") is True: the flag was read raw and stopped the service.
        api_v3_client.post(STOP_URL, json={"stop_service": "false"})
        service["stop_service"].assert_not_called()

    def test_stop_service_true_still_stops_it(self, api_v3_client, service):
        api_v3_client.post(STOP_URL, json={"stop_service": True})
        service["stop_service"].assert_called_once()