manage/backend/tests/test_scheduler.py

"""Unit tests for typed scheduled actions and scheduler persistence."""

from __future__ import annotations

import threading
import time
from types import SimpleNamespace
from typing import cast
from unittest.mock import patch

import pytest
from cryptography.fernet import Fernet
from fastapi.testclient import TestClient

from media_library_viewer_api.dependencies import get_settings_store
from media_library_viewer_api.main import app
from media_library_viewer_api.services.qbittorrent_store import (
    QBITTORRENT_CONCERN,
    QbittorrentSampleStore,
)
from media_library_viewer_api.services.scheduler import (  # type: ignore[reportMissingImports]
    Scheduler,
    SchedulerBusyError,
)
from media_library_viewer_api.services.scheduler_actions import (  # type: ignore[reportMissingImports]
    QBITTORRENT_SPEED_ACTION,
    ActionResult,
)
from media_library_viewer_api.services.scheduler_store import (  # type: ignore[reportMissingImports]
    SCHEDULER_CONCERN,
    SchedulerRunStore,
)
from media_library_viewer_api.services.secrets import reset_encryption_key_cache
from media_library_viewer_api.services.service_data import ServiceDataHarness
from media_library_viewer_api.services.settings_store import SettingsStore
from media_library_viewer_api.widgets.sources import ServiceRecord


@pytest.fixture
def scheduler_client(tmp_path, monkeypatch):
    monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", Fernet.generate_key().decode())
    reset_encryption_key_cache()
    store = SettingsStore(tmp_path / "settings.sqlite")
    store.ensure_defaults()
    app.dependency_overrides[get_settings_store] = lambda: store
    with patch("media_library_viewer_api.auth.get_settings", return_value=SimpleNamespace(auth_enabled=False)):
        yield TestClient(app), store
    app.dependency_overrides.clear()
    reset_encryption_key_cache()


def _service(service_id: str = "svc-1") -> ServiceRecord:
    return ServiceRecord(
        id=service_id,
        service_type="qbittorrent",
        name="qbit",
        config={
            "poll_interval_seconds": 15,
            "sample_retention_seconds": 1_800,
            "sample_max_rows": 1200,
        },
        secrets={},
    )


def test_scheduler_routes_expose_status_history_and_disabled_manual_run(scheduler_client):
    client, store = scheduler_client
    service = store.upsert_service(
        {
            "service_type": "qbittorrent",
            "name": "qbit",
            "config": {"base_url": "http://qbit:8080", "polling_enabled": False},
            "secrets": {},
            "enabled": True,
        }
    )

    status = client.get(f"/api/scheduler/services/{service['id']}/status")
    assert status.status_code == 200
    assert not status.json()["enabled"]

    runs = client.get(f"/api/scheduler/services/{service['id']}/runs")
    assert runs.status_code == 200
    assert runs.json()["items"] == []

    manual = client.post(f"/api/scheduler/services/{service['id']}/run")
    assert manual.status_code == 400


def test_scheduler_samples_all_values_reads_all_retained_samples(scheduler_client):
    client, store = scheduler_client
    service = store.upsert_service(
        {
            "service_type": "qbittorrent",
            "name": "qbit",
            "config": {"base_url": "http://qbit:8080"},
            "secrets": {},
            "enabled": True,
        }
    )
    retained = [{"ts": 10, "dl_speed": 20, "up_speed": 30}]
    with patch("media_library_viewer_api.routers.scheduler.QbittorrentSampleStore") as store_cls:
        store_cls.return_value.window.return_value = retained
        response = client.get(f"/api/scheduler/services/{service['id']}/samples?all_values=true")

    assert response.status_code == 200
    assert response.json() == {
        "service_id": service["id"],
        "window_seconds": None,
        "all_values": True,
        "samples": retained,
    }
    store_cls.return_value.window.assert_called_once_with(service["id"])


def test_sample_store_applies_time_and_row_limits(tmp_path):
    harness = ServiceDataHarness(tmp_path)
    harness.register(QBITTORRENT_CONCERN)
    harness.run_migrations()
    store = QbittorrentSampleStore(harness)

    now = round(time.time())
    for index in range(70):
        store.append(
            "svc-1",
            now - 20 + index,
            index,
            index,
            retention_seconds=60,
            max_rows=60,
        )

    samples = store.window("svc-1")
    assert len(samples) == 60
    assert samples[0]["ts"] == now - 10


def test_scheduler_records_success_and_manual_run_resets_backoff(tmp_path, monkeypatch):
    harness = ServiceDataHarness(tmp_path)
    harness.register(SCHEDULER_CONCERN)
    harness.run_migrations()
    scheduler = Scheduler()
    scheduler._run_store = SchedulerRunStore(harness)
    service = _service()

    class Action:
        def run(self, value):
            assert value.id == "svc-1"
            return ActionResult(data={"ok": True})

    with patch("media_library_viewer_api.services.scheduler.get_scheduled_action", return_value=Action()):
        result = scheduler._execute(service, "manual")

    assert result["status"] == "success"
    fake_store = cast(
        SettingsStore,
        SimpleNamespace(get_service=lambda service_id: {"id": service_id, "service_type": "qbittorrent", "config": {}}),
    )
    status = scheduler.status("svc-1", store=fake_store)
    assert status["consecutive_failures"] == 0
    assert status["backoff_until"] is None
    runs, total = scheduler._run_store.list_runs("svc-1", QBITTORRENT_SPEED_ACTION)
    assert total == 1
    assert runs[0]["trigger"] == "manual"


def test_scheduler_rejects_overlapping_manual_runs(tmp_path):
    harness = ServiceDataHarness(tmp_path)
    harness.register(SCHEDULER_CONCERN)
    harness.run_migrations()
    scheduler = Scheduler()
    scheduler._run_store = SchedulerRunStore(harness)
    service = _service()
    started = threading.Event()
    release = threading.Event()

    class SlowAction:
        def run(self, value):
            started.set()
            release.wait(2)
            return ActionResult(data={})

    with patch("media_library_viewer_api.services.scheduler.get_scheduled_action", return_value=SlowAction()):
        worker = threading.Thread(target=scheduler._execute, args=(service, "manual"))
        worker.start()
        assert started.wait(1)
        try:
            try:
                scheduler._execute(service, "manual")
            except SchedulerBusyError:
                pass
            else:
                raise AssertionError("expected overlapping run to be rejected")
        finally:
            release.set()
            worker.join(timeout=2)


def test_scheduler_status_marks_never_run_service_stale():
    scheduler = Scheduler()
    store = cast(
        SettingsStore,
        SimpleNamespace(
            get_service=lambda service_id: {
                "id": service_id,
                "service_type": "qbittorrent",
                "enabled": True,
                "config": {},
            }
        ),
    )
    status = scheduler.status("svc-1", store=store)
    assert bool(status["is_stale"])
    assert status["poll_interval_seconds"] == 15