schedule/tests/test_admin.py

"""Тесты админки: вход по cookie, смена ссылки/интервала, принудительное обновление."""

import json
import time

import pytest
from fastapi.testclient import TestClient

import main
from admin import AdminConfig, make_token, verify_token
from fixtures import fixture_xlsx, make_cache
from schedule import ScheduleError
from schedule.settings import SettingsStore

ADMIN = AdminConfig(path="/secret-panel", login="admin", password="s3cret")


@pytest.fixture()
def store(tmp_path):
    return SettingsStore(link="test-link", interval_minutes=15, path=tmp_path / "settings.json")


@pytest.fixture()
def cache(monkeypatch, store):
    c, calls, fail = make_cache(monkeypatch, fixture_xlsx(), settings=store)
    c.calls = calls
    c.fail = fail
    return c


@pytest.fixture()
def client(cache):
    app = main.create_app(cache, admin=ADMIN)
    with TestClient(app) as c:
        yield c


def do_login(client, login="admin", password="s3cret"):
    return client.post("/api/admin/login", json={"login": login, "password": password})


# --- авторизация ---------------------------------------------------------------


@pytest.mark.parametrize(
    "method, path, payload",
    [
        ("get", "/api/admin/state", None),
        ("post", "/api/admin/refresh", {}),
        ("post", "/api/admin/link", {"link": "https://disk.yandex.ru/i/x"}),
        ("post", "/api/admin/interval", {"interval_minutes": 30}),
    ],
)
def test_requires_login(client, method, path, payload):
    resp = getattr(client, method)(path, json=payload) if method == "post" else client.get(path)
    assert resp.status_code == 401


@pytest.mark.parametrize(
    "login, password",
    [("admin", "неверный"), ("неверный", "s3cret"), ("", "")],
)
def test_login_rejects_wrong_credentials(client, login, password):
    resp = client.post("/api/admin/login", json={"login": login, "password": password})
    assert resp.status_code == 401
    assert "schedule_admin" not in client.cookies


def test_login_and_state(client, cache):
    assert do_login(client).status_code == 200
    resp = client.get("/api/admin/state")
    assert resp.status_code == 200
    state = resp.json()
    assert state["link"] == "test-link"
    assert state["interval_minutes"] == 15
    assert state["groups"] == 2
    assert state["error"] is None


def test_logout(client):
    do_login(client)
    assert client.get("/api/admin/state").status_code == 200
    assert client.post("/api/admin/logout").status_code == 200
    assert client.get("/api/admin/state").status_code == 401


def test_token_helpers():
    token = make_token("admin", "s3cret", ttl=60)
    assert verify_token(token, "s3cret")
    assert not verify_token(token, "другой-пароль")
    assert not verify_token(token + "x", "s3cret")
    assert not verify_token("мусор", "s3cret")
    assert not verify_token(make_token("admin", "s3cret", ttl=-5), "s3cret")


# --- интервал ------------------------------------------------------------------


def test_set_interval(client, cache, store):
    do_login(client)
    resp = client.post("/api/admin/interval", json={"interval_minutes": 30})
    assert resp.status_code == 200
    assert resp.json()["interval_minutes"] == 30
    assert cache.interval_minutes == 30
    assert cache._ttl == 1800
    assert client.get("/api/admin/state").json()["interval_minutes"] == 30
    # сохранилось на диск
    saved = json.loads((store._path).read_text(encoding="utf-8"))
    assert saved["interval_minutes"] == 30
    # новый процесс видит сохранённое значение
    assert SettingsStore(path=store._path).interval_minutes == 30


@pytest.mark.parametrize("value, status", [(0, 400), (1441, 400), (-5, 400), ("abc", 422)])
def test_set_interval_validation(client, value, status):
    do_login(client)
    resp = client.post("/api/admin/interval", json={"interval_minutes": value})
    assert resp.status_code == status


# --- смена ссылки ---------------------------------------------------------------


def test_set_link(client, cache, store):
    do_login(client)
    before = cache.calls["download"]
    resp = client.post("/api/admin/link", json={"link": "  https://disk.yandex.ru/i/new  "})
    assert resp.status_code == 200
    state = resp.json()
    assert state["link"] == "https://disk.yandex.ru/i/new"
    assert cache.calls["download"] == before + 1  # перечитали таблицу
    assert json.loads(store._path.read_text(encoding="utf-8"))["link"] == "https://disk.yandex.ru/i/new"


def test_set_link_rejects_broken(client, cache, store):
    do_login(client)
    old_link = cache.state()["link"]
    cache.fail["download"] = True
    resp = client.post("/api/admin/link", json={"link": "https://disk.yandex.ru/i/broken"})
    assert resp.status_code == 400
    assert "не подошла" in resp.json()["detail"]
    # ссылка и данные не изменились
    assert cache.state()["link"] == old_link
    assert cache.state()["has_data"] is True
    # в файл настроек нерабочая ссылка не попала
    if store._path.exists():
        assert json.loads(store._path.read_text(encoding="utf-8"))["link"] != "https://disk.yandex.ru/i/broken"


def test_set_empty_link(client, cache):
    do_login(client)
    resp = client.post("/api/admin/link", json={"link": "   "})
    assert resp.status_code == 400
    assert cache.state()["has_data"] is True


# --- принудительное обновление ---------------------------------------------------


def test_force_refresh_ignores_ttl_and_modified(client, cache):
    do_login(client)
    client.get("/api/admin/state")  # кэш свежий, дальше без принуждения не качали бы
    downloads = cache.calls["download"]
    cache._ttl = 3600
    resp = client.post("/api/admin/refresh")
    assert resp.status_code == 200
    assert cache.calls["download"] == downloads + 1
    assert resp.json()["has_data"] is True


def test_force_refresh_reports_error(client, cache):
    do_login(client)
    cache.fail["download"] = True
    resp = client.post("/api/admin/refresh")
    assert resp.status_code == 502
    assert cache.state()["error"] is not None
    # старые данные не потеряли
    assert cache.state()["has_data"] is True


# --- выключенная админка -----------------------------------------------------------


def test_admin_disabled_without_credentials(cache):
    app = main.create_app(cache, admin=AdminConfig())
    with TestClient(app) as c:
        assert c.get("/api/admin/state").status_code == 404
        assert c.post("/api/admin/login", json={"login": "a", "password": "b"}).status_code == 404
        assert c.get("/healthz").status_code == 200