ayurishchevandClaude Opus 5.5
e1f197ccff
Пункт 12 ревью 2026-09-28 17:35 (docs/changes/026): tests/test_app.py (936 строк) разнесён на test_security, test_devices, test_operations, test_backups, test_events, test_ids_migrations, test_architecture; общие хелперы — tests/helpers.py. Тела тестов не менялись (AST совпадает). security.reset_all_failures() — фикстура сбрасывает счётчики попыток по всем ключам, тесты не зависят от порядка. Тесты: 36 из 36 полным прогоном, каждый модуль отдельно и в обратном порядке; тест-линтер после переноса проверен мутацией. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
181 lines
11 KiB
Python
181 lines
11 KiB
Python
import asyncio
|
|
import html
|
|
import json
|
|
import re
|
|
from datetime import timedelta
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import ids, security
|
|
from app.db import session_scope
|
|
from app.main import create_app
|
|
from app.models import Event, now
|
|
from app.services import devices, events, jobs, ops, settings
|
|
from tests.conftest import ADMIN_PASSWORD, API_TOKEN
|
|
from tests.helpers import ros_client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_link_entities(monkeypatch):
|
|
"""Журнал: у каждой записи свой ID, ссылка на ID сущности и актор; в журнал попадают только смены online/offline."""
|
|
events.set_actor("api")
|
|
d = devices.create_device("r1", "10.0.0.1", 80, "u", "p")
|
|
up = {"on": False}
|
|
|
|
def handler(r: httpx.Request) -> httpx.Response:
|
|
if not up["on"]:
|
|
raise httpx.ConnectError("refused")
|
|
data = {"/rest/system/resource": {"version": "7.24", "uptime": "1m", "board-name": "hAP"},
|
|
"/rest/system/routerboard": {"routerboard": "false"},
|
|
"/rest/system/package/update": {"channel": "stable", "installed-version": "7.24", "latest-version": "7.24"}}
|
|
return httpx.Response(200, json=data.get(r.url.path, []))
|
|
|
|
monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler))
|
|
await ops.poll_device(d.id)
|
|
await ops.poll_device(d.id) # повторный опрос без смены состояния — нового события нет
|
|
up["on"] = True
|
|
await ops.poll_device(d.id)
|
|
|
|
async def fake_backup(device_id):
|
|
return "готово"
|
|
|
|
monkeypatch.setitem(jobs.JOB_TYPES, "backup", fake_backup)
|
|
[job_id] = await jobs.start_jobs("backup", [d.id])
|
|
await asyncio.sleep(0.3)
|
|
|
|
evs = events.list_events(device_id=d.id)
|
|
assert [e.type for e in reversed(evs)] == ["device.created", "device.offline", "device.online",
|
|
"job.created", "job.started", "job.done"]
|
|
assert all(ids.is_id(e.id, "evt") and e.actor == "api" and e.device_id == d.id for e in evs)
|
|
assert len({e.id for e in evs}) == len(evs) and [e.id for e in evs] == sorted((e.id for e in evs), reverse=True)
|
|
job_events = events.list_events(job_id=job_id)
|
|
assert {e.entity_id for e in job_events} == {job_id} and jobs.get_job(job_id).status == "done"
|
|
|
|
|
|
def test_events_api_and_id_validation():
|
|
"""API отдаёт журнал с ID; ID чужого типа и числовые ID дают 404; вход в UI попадает в журнал."""
|
|
with TestClient(create_app()) as c:
|
|
h = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
c.post("/login", data={"username": "admin", "password": "wrong"})
|
|
c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD})
|
|
d = c.post("/api/v1/devices", headers=h, json={"name": "r1", "host": "10.0.0.1", "username": "u", "password": "p"}).json()
|
|
assert ids.is_id(d["id"], "dev")
|
|
evs = c.get("/api/v1/events", headers=h, params={"entity_id": d["id"]}).json()
|
|
assert [e["type"] for e in evs] == ["device.created"] and evs[0]["actor"] == "api" and evs[0]["data"]["name"] == "r1"
|
|
assert c.get(f"/api/v1/events/{evs[0]['id']}", headers=h).json()["entity_id"] == d["id"]
|
|
auth = c.get("/api/v1/events", headers=h, params={"type": "auth"}).json()
|
|
assert sorted(e["type"] for e in auth) == ["auth.failed", "auth.login"]
|
|
assert ADMIN_PASSWORD not in json.dumps(auth) # пароли в журнал не попадают
|
|
assert c.get("/api/v1/devices/1", headers=h).status_code == 404
|
|
assert c.get(f"/api/v1/devices/{ids.new_id('job')}", headers=h).status_code == 404
|
|
assert c.get(f"/api/v1/jobs/{d['id']}", headers=h).status_code == 404
|
|
|
|
|
|
def _fill_events(n: int, *, age_days: int = 0) -> None:
|
|
with session_scope() as s:
|
|
for i in range(n):
|
|
s.add(Event(type="device.updated", entity_type="device", message=f"событие {i}", actor="system",
|
|
ts=now() - timedelta(days=age_days)))
|
|
|
|
|
|
def test_journal_rotation_by_age_and_count():
|
|
"""Ротация: по сроку, по числу записей (после неё в журнале ровно max_rows), 0 = без ограничения, валидация."""
|
|
settings.save_journal(0, 0)
|
|
_fill_events(5, age_days=100)
|
|
_fill_events(20)
|
|
assert settings.save_journal(30, 0) == {"retention_days": 30, "max_rows": 0} # срок 30 дней: старые 5 удалены
|
|
assert events.count(type_="device.updated") == 20
|
|
rotated = events.list_events(type_="journal.rotated")
|
|
assert len(rotated) == 1 and json.loads(rotated[0].data)["by_age"] == 5
|
|
|
|
_fill_events(150)
|
|
settings.save_journal(0, 100) # лимит 100: остаётся ровно 100 записей, удалены самые старые
|
|
assert events.count() == 100
|
|
newest = events.list_events(limit=1)[0]
|
|
assert newest.type == "journal.rotated" and json.loads(newest.data)["by_count"] > 0
|
|
assert events.rotate() == {"by_age": 0, "by_count": 0} # повторный запуск ничего не удаляет (нет «пилы»)
|
|
|
|
settings.save_journal(0, 0) # 0/0 — без ограничений
|
|
_fill_events(30)
|
|
assert events.rotate() == {"by_age": 0, "by_count": 0}
|
|
for bad in ((-1, 0), (0, 50), ("x", 0)):
|
|
with pytest.raises(ValueError):
|
|
settings.save_journal(*bad)
|
|
|
|
|
|
def test_journal_clear_requires_password_through_modal():
|
|
"""Очистка: только через окно с паролем пользователя; неверный пароль не очищает и ведёт к блокировке;
|
|
успех оставляет одну запись об очистке; в API очистки нет."""
|
|
htmx = {"HX-Request": "true"}
|
|
with TestClient(create_app()) as c:
|
|
h = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD})
|
|
_fill_events(30)
|
|
before = events.count()
|
|
page = c.get("/ui/dialog/events-clear", headers=htmx).text
|
|
assert "Очистить журнал" in page and 'id="clear-confirm" disabled' in page # кнопка неактивна без пароля
|
|
|
|
assert c.post("/events/clear", data={"password": ADMIN_PASSWORD}).status_code == 400 # не через окно — отклонено
|
|
assert events.count() == before
|
|
r = c.post("/events/clear", data={"password": ""}, headers=htmx)
|
|
assert "Введите пароль" in r.text and events.count() == before
|
|
for left in (4, 3, 2, 1):
|
|
r = c.post("/events/clear", data={"password": "wrong"}, headers=htmx)
|
|
assert f"Осталось попыток: {left}" in r.text and "HX-Refresh" not in r.headers
|
|
r = c.post("/events/clear", data={"password": "wrong"}, headers=htmx) # 5-я неверная — блокировка
|
|
assert "Слишком много неверных попыток" in r.text
|
|
r = c.post("/events/clear", data={"password": ADMIN_PASSWORD}, headers=htmx) # верный пароль при блокировке не принимается
|
|
assert "Повторите через" in r.text and "HX-Refresh" not in r.headers
|
|
denied = events.list_events(type_="journal.clear_denied")
|
|
assert len(denied) == 5 and "wrong" not in json.dumps([e.data for e in denied]) # пароль в журнал не пишется
|
|
assert events.list_events(type_="journal.clear_locked")
|
|
|
|
security.reset_failures("admin") # окончание блокировки
|
|
r = c.post("/events/clear", data={"password": ADMIN_PASSWORD}, headers=htmx)
|
|
assert r.headers["HX-Refresh"] == "true"
|
|
left = events.list_events()
|
|
assert [e.type for e in left] == ["journal.cleared"] and left[0].actor == "ui:admin"
|
|
assert json.loads(left[0].data)["deleted"] > 30
|
|
|
|
assert c.delete("/api/v1/events", headers=h).status_code in (404, 405) # очистки через API нет
|
|
assert c.post("/api/v1/events/clear", headers=h, json={"password": ADMIN_PASSWORD}).status_code in (404, 405, 422)
|
|
|
|
|
|
def test_journal_page_filters_cursor_dialogs_and_settings():
|
|
"""Страница журнала: фильтры, «Показать ещё» по курсору, окно записи, сохранение настроек."""
|
|
htmx = {"HX-Request": "true"}
|
|
with TestClient(create_app()) as c:
|
|
h = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD})
|
|
d = c.post("/api/v1/devices", headers=h, json={"name": "r1", "host": "10.0.0.1", "username": "u", "password": "p"}).json()
|
|
_fill_events(105)
|
|
page = c.get("/events").text
|
|
assert page.count('class="clickable"') == 100 and "Показать ещё 100" in page and "Показано 100 из" in page
|
|
last = re.findall(r'href="/events/(evt_[0-9a-f-]+)"', page)[-1]
|
|
more = c.get(f"/ui/events?before={last}&shown=100", headers=htmx).text
|
|
assert more.count('class="clickable"') >= 5 and 'hx-swap-oob="true"' in more and "Показать ещё" not in more
|
|
|
|
# фильтры: тип (группа), актор, устройство, поиск по ID
|
|
only_dev = c.get("/events?type=device&actor=api&device=" + d["id"]).text
|
|
assert only_dev.count('class="clickable"') == 1 and "device.created" in only_dev
|
|
assert c.get("/events?q=" + d["id"][-8:]).text.count('class="clickable"') >= 1
|
|
assert "Записей не найдено" in c.get("/events?q=нет-такого-текста").text
|
|
assert c.get("/api/v1/events", headers=h, params={"actor": "api"}).json()[0]["type"] == "device.created"
|
|
|
|
ev = events.list_events(entity_id=d["id"])[0]
|
|
dialog = c.get(f"/ui/dialog/event/{ev.id}", headers=htmx).text
|
|
shown = html.unescape(dialog) # Jinja экранирует кавычки в JSON, браузер показывает обычные
|
|
assert ev.id in shown and d["id"] in shown and '"name": "r1"' in shown and 'data-copy="' + ev.id in shown
|
|
assert c.get("/events/evt_bogus").status_code == 404 and c.get(f"/events/{ev.id}").status_code == 200
|
|
|
|
# настройки: ошибка остаётся в окне, успех — обновление страницы, значения видны в API
|
|
r = c.post("/events/settings", data={"retention_days": "10", "max_rows": "5"}, headers=htmx)
|
|
assert "Максимум записей" in r.text and "HX-Refresh" not in r.headers
|
|
r = c.post("/events/settings", data={"retention_days": "45", "max_rows": "5000"}, headers=htmx)
|
|
assert r.headers["HX-Refresh"] == "true"
|
|
assert c.get("/api/v1/events/settings", headers=h).json()["retention_days"] == 45
|
|
assert c.put("/api/v1/events/settings", headers=h, json={"retention_days": 0, "max_rows": 0}).json()["max_rows"] == 0
|
|
assert "ротация: без ограничений" in c.get("/events").text
|