schedule/tests/test_admin.py

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

import json

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, store):
    app = main.create_app(cache, admin=ADMIN, settings=store)
    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
    assert state["throttle"]["max_attempts"] == ADMIN.max_attempts
    # путь к панели не должен утекать в JSON: переход идёт через /goto
    assert "admin_path" not in state
    assert ADMIN.path not in str(state)


def test_goto_redirects_to_panel_only_for_authorized(client):
    """Переход в панель: без сессии 401, с сессией — редирект на секретный путь."""
    assert client.get("/api/admin/goto", follow_redirects=False).status_code == 401
    do_login(client)
    resp = client.get("/api/admin/goto", follow_redirects=False)
    assert resp.status_code == 302
    assert resp.headers["location"] == ADMIN.path


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_lockout_after_configured_attempts(client):
    """Разрешённые попытки тратятся, следующая попытка включает паузу."""
    for _ in range(ADMIN.max_attempts - 1):
        resp = do_login(client, password="неверный")
        assert resp.status_code == 401
        assert "Осталось попыток" in resp.json()["detail"]

    exhausted = do_login(client, password="неверный")
    assert exhausted.status_code == 429
    assert "заблокирован" in exhausted.json()["detail"].lower()
    # пауза действует и для верного пароля
    assert do_login(client, password="s3cret").status_code == 429


def test_successful_login_resets_failures(client):
    assert do_login(client, password="неверный").status_code == 401
    assert do_login(client).status_code == 200
    # счётчик сброшен: ещё две неудачи не блокируют
    assert do_login(client, password="неверный").status_code == 401
    assert do_login(client, password="неверный").status_code == 401
    assert do_login(client).status_code == 200


# --- история входов ---------------------------------------------------------------


def test_login_history_visible_and_saved(client, store):
    assert do_login(client, password="неверный").status_code == 401
    assert do_login(client).status_code == 200

    state = client.get("/api/admin/state").json()
    assert state["last_attempt"]["ok"] is True
    assert state["last_attempt"]["ip"]  # откуда
    assert state["last_attempt"]["at"]  # когда
    assert state["last_success"]["ip"] == state["last_attempt"]["ip"]

    saved = json.loads(store.path.read_text(encoding="utf-8"))
    assert saved["last_attempt"]["ok"] is True
    assert saved["last_success"]["at"]


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


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
    assert resp.json()["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, store):
    app = main.create_app(cache, admin=AdminConfig(), settings=store)
    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