Vladislav Kopnin
e007a85f56
- не больше 10 неудачных входов с одного адреса за 5 минут, дальше 429 (панель доступна из интернета, простой перебор закрываем); успешный вход сбрасывает счётчик; - .env.example: SETTINGS_FILE пустой — локально настройки живут в памяти, в docker compose подставляет /data/settings.json по умолчанию.
198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
"""Тесты админки: вход по cookie, смена ссылки/интервала, принудительное обновление."""
|
||
|
||
import json
|
||
import time
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
import main
|
||
from admin import LOGIN_MAX_FAILURES, 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_login_throttled_after_failures(client):
|
||
for _ in range(LOGIN_MAX_FAILURES):
|
||
assert do_login(client, password="неверный").status_code == 401
|
||
# дальше — отказ, даже с верным паролем, пока не истекло окно
|
||
assert do_login(client, password="неверный").status_code == 429
|
||
assert do_login(client, password="s3cret").status_code == 429
|
||
|
||
|
||
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
|