From e1f197ccff7bd620693ed09e70bc816d99b6fe78 Mon Sep 17 00:00:00 2001 From: ayurishchev Date: Mon, 28 Sep 2026 21:30:23 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A2=D0=B5=D1=81=D1=82=D1=8B=20=D0=BF=D0=BE?= =?UTF-8?q?=20=D0=BC=D0=BE=D0=B4=D1=83=D0=BB=D1=8F=D0=BC=20=D0=BF=D1=80?= =?UTF-8?q?=D0=B5=D0=B4=D0=BC=D0=B5=D1=82=D0=BD=D1=8B=D1=85=20=D0=BE=D0=B1?= =?UTF-8?q?=D0=BB=D0=B0=D1=81=D1=82=D0=B5=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Пункт 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 --- README.md | 10 +- app/security.py | 7 + docs/changes/026-split-tests/plan.md | 48 ++ docs/changes/026-split-tests/summary.md | 20 + tests/conftest.py | 3 +- tests/helpers.py | 15 + tests/test_app.py | 990 ------------------------ tests/test_architecture.py | 68 ++ tests/test_backups.py | 175 +++++ tests/test_devices.py | 167 ++++ tests/test_events.py | 180 +++++ tests/test_ids_migrations.py | 110 +++ tests/test_operations.py | 185 +++++ tests/test_security.py | 150 ++++ 14 files changed, 1133 insertions(+), 995 deletions(-) create mode 100644 docs/changes/026-split-tests/plan.md create mode 100644 docs/changes/026-split-tests/summary.md create mode 100644 tests/helpers.py delete mode 100644 tests/test_app.py create mode 100644 tests/test_architecture.py create mode 100644 tests/test_backups.py create mode 100644 tests/test_devices.py create mode 100644 tests/test_events.py create mode 100644 tests/test_ids_migrations.py create mode 100644 tests/test_operations.py create mode 100644 tests/test_security.py diff --git a/README.md b/README.md index c8150fe..3f10851 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,8 @@ app/ros/ client.py (REST-клиент) operations.py (статус, бэ app/services/ devices groups backups ops jobs poller events settings rotation app/api/ v1.py (JSON API) app/ui/ routes.py templates/ static/ (стили, app.js, шрифты) -tests/ автотесты (pytest) +tests/ автотесты (pytest) — по модулям: test_security test_devices test_operations test_backups + test_events test_ids_migrations test_architecture; helpers.py — общие хелперы, conftest.py — фикстуры docs/changes/ планы и итоги доработок docs/reviews/ отчёты ревью ``` @@ -168,10 +169,12 @@ curl -s -H "Authorization: Bearer $API_TOKEN" http://localhost:8000/api/v1/devic - Светлая и тёмная темы — по настройке системы. ## Тесты -36 тестов, фоновый опрос выключен; стенд не нужен (временная SQLite, RouterOS и S3 — заглушки). Тест-линтер не допускает синхронных обращений к БД в `async`-коде. +36 тестов по модулям предметных областей (`tests/test_*.py`), фоновый опрос выключен; стенд не нужен (временная SQLite, +RouterOS и S3 — заглушки). Тест-линтер (`test_architecture.py`) не допускает синхронных обращений к БД в `async`-коде. ```bash python3 -m venv venv && venv/bin/pip install -r requirements.txt -venv/bin/python -m pytest -q +venv/bin/python -m pytest -q # все тесты +venv/bin/python -m pytest -q tests/test_backups.py # один модуль ``` Локальный запуск без Docker: `set -a; . ./.env; set +a; venv/bin/uvicorn app.main:app --reload`. @@ -205,6 +208,7 @@ venv/bin/python -m pytest -q | 023 | Состояния «Upgrade ROS», откат ROS до версии канала | [план](docs/changes/023-ros-downgrade/plan.md) · [итог](docs/changes/023-ros-downgrade/summary.md) | | 024 | Оптимизация README | [план](docs/changes/024-readme-restructure/plan.md) · [итог](docs/changes/024-readme-restructure/summary.md) | | 025 | Доверенные прокси (реальный IP клиента), параметризация порта | [план](docs/changes/025-trusted-proxies-port/plan.md) · [итог](docs/changes/025-trusted-proxies-port/summary.md) | +| 026 | Тесты по модулям предметных областей | [план](docs/changes/026-split-tests/plan.md) · [итог](docs/changes/026-split-tests/summary.md) | ## Отчёты ревью - [Ревью кодовой базы 2026-09-27](docs/reviews/2026-09-27-codebase-review.md) (→ 018–021) diff --git a/app/security.py b/app/security.py index ed25577..205381d 100644 --- a/app/security.py +++ b/app/security.py @@ -109,6 +109,13 @@ def reset_failures(key: str) -> None: _locked_until.pop(key, None) +def reset_all_failures() -> None: + """Сбрасывает счётчики неудачных попыток по всем ключам (для тестов и будущего администрирования).""" + with _guard: + _fails.clear() + _locked_until.clear() + + async def require_api_token(cred: HTTPAuthorizationCredentials | None = Depends(_bearer)) -> None: expected = get_settings().api_token if cred is None or not hmac.compare_digest(cred.credentials.encode(), expected.encode()): diff --git a/docs/changes/026-split-tests/plan.md b/docs/changes/026-split-tests/plan.md new file mode 100644 index 0000000..7a9a7f7 --- /dev/null +++ b/docs/changes/026-split-tests/plan.md @@ -0,0 +1,48 @@ +# План: 026 — тесты по модулям (п. 12 ревью) + +## Context +Ревью `docs/reviews/2026-09-28-1735-codebase-review.md`, п. 12: все тесты — в одном `tests/test_app.py` (936 строк, 36 тестов), +файл растёт с каждым изменением (570 → 753 → 936). Трудно найти тест по области, конфликтуют правки параллельных изменений. +Нужно разнести тесты по модулям предметных областей **без изменения логики тестов**. + +Решений пользователя не требуется: чисто структурное изменение; код приложения и стенд не затрагиваются (тесты не входят в образ). + +## Изменения + +### Модули (`tests/`) +| Файл | Тесты | +|---|---| +| `test_security.py` | `test_password_encryption_roundtrip`, `test_api_auth_and_no_password_leak`, `test_insecure_settings_rejects_weak_secrets_without_leaking_them`, `test_insecure_settings_rejects_invalid_trusted_proxies`, `test_client_ip_trusted_proxies`, `test_app_refuses_to_start_with_insecure_config`, `test_login_lockout_by_ip`, `test_login_lockout_by_ip_behind_trusted_proxy`, `test_move_redirect_rejects_open_redirect_next`, `test_process_lock_blocks_second_process` | +| `test_devices.py` | `test_status_parsing`, `test_chr_status_and_version_compare`, `test_groups_and_device_filters`, `test_create_device_via_ui_binds_group`, `test_device_name_is_immutable`, `test_poll_marks_offline_and_recovers`, `test_ros_state_values` | +| `test_operations.py` | `test_firmware_reboots_on_log_message`, `test_batch_channel_runs_as_jobs`, `test_run_ros_downgrade_order_and_failures`, `test_downgrade_api_endpoints`, `test_ui_downgrade_dialog_and_mismatch` | +| `test_backups.py` | `test_backup_flow`, `test_backup_filters`, `test_bulk_delete_backups`, `test_backup_files_get_ids_and_deletion_is_recorded`, `test_bucket_list_is_cached_between_reads` | +| `test_events.py` | `test_events_link_entities`, `test_events_api_and_id_validation`, `test_journal_rotation_by_age_and_count`, `test_journal_clear_requires_password_through_modal`, `test_journal_page_filters_cursor_dialogs_and_settings` (+ хелпер `_fill_events`) | +| `test_ids_migrations.py` | `test_ids_are_unique_typed_and_ordered`, `test_migration_replaces_numeric_ids`, `test_migration_adds_legacy_columns_before_id_migration` (+ `LEGACY_DDL`, `LEGACY_DDL_NO_EXTRA_COLUMNS`) | +| `test_architecture.py` | `test_no_sync_db_calls_in_async_functions` (путь к `app/` — через `Path(__file__)`, работает и из нового файла) | + +- Общие хелперы `ros_client`, `_request` → `tests/helpers.py` (импорт `from tests.helpers import …`); константы секретов остаются в `conftest.py`. +- В каждом модуле — только нужные ему импорты (без «всё из всего»). +- Тела тестов переносятся **без изменений**. Допустимы только правки импортов и, при необходимости, изоляции (ниже). +- `tests/test_app.py` удаляется. + +### Изоляция тестов (`tests/conftest.py`) +- Счётчики неудачных попыток (`security._fails`, `security._locked_until`) — состояние процесса; сейчас фикстура сбрасывает только ключи + `admin` и `login:testclient`, а тесты используют и другие IP (`10.0.0.x`, `1.1.1.1`). После разбиения порядок выполнения меняется — + сбрасывать **все** счётчики в автоиспользуемой фикстуре: добавить в `app/security.py` функцию `reset_all_failures()` (для тестов и + будущего администрирования) и вызывать её вместо двух точечных сбросов. + +## Тесты +Логика тестов не меняется. Критерий — тот же набор из 36 тестов, все зелёные, в том числе при запуске каждого модуля отдельно. + +## Документация +README: «Архитектура» (дерево — `tests/` по модулям), «Тесты» (как запустить модуль), строка 026 в истории. `summary.md` — оркестратор. + +## Исполнение +Исполнитель (Sonnet): перенос, `helpers.py`, `conftest.py`, `reset_all_failures`, README. Тесты не запускает, не коммитит. Стенд не пересобирается +(код приложения меняется только добавлением `reset_all_failures`, которую приложение не вызывает). + +## Проверка (оркестратор) +- Набор имён тестов до и после совпадает (36; список «до» сохранён оркестратором). +- `pytest -q` — все зелёные; **каждый модуль отдельно** — зелёный (ловит скрытые зависимости от порядка и импортов); + обратный порядок модулей (`pytest tests/test_security.py … в обратном порядке`) — зелёный. +- `tests/test_app.py` отсутствует; `grep` тел тестов: построчное сравнение тел функций до/после (AST-дамп тел тестов совпадает). diff --git a/docs/changes/026-split-tests/summary.md b/docs/changes/026-split-tests/summary.md new file mode 100644 index 0000000..b8899f5 --- /dev/null +++ b/docs/changes/026-split-tests/summary.md @@ -0,0 +1,20 @@ +# Итоги: 026 — тесты по модулям (п. 12 ревью) + +Источник — ревью `docs/reviews/2026-09-28-1735-codebase-review.md`, п. 12. + +## Сделано +- `tests/test_app.py` (936 строк, 36 тестов) разнесён по модулям без изменения тел тестов: + `test_security.py` (10), `test_devices.py` (7), `test_operations.py` (5), `test_backups.py` (5), `test_events.py` (5), + `test_ids_migrations.py` (3), `test_architecture.py` (1). Общие `ros_client`, `_request` — `tests/helpers.py`; каждый модуль импортирует только нужное (pyflakes чисто). +- Изоляция: `security.reset_all_failures()` (очистка всех счётчиков неудачных попыток под блокировкой); фикстура `env` вызывает её вместо двух + точечных сбросов — тесты с разными IP не влияют друг на друга при любом порядке. Приложение функцию не вызывает. +- README: дерево `tests/` в «Архитектуре», запуск одного модуля в «Тестах», строка 026. + +## Проверено +- AST всех 36 тестовых функций (с декораторами) совпадает с `tests/test_app.py` из HEAD; имена тестов до и после совпадают. +- `pytest`: 36 из 36 полным прогоном; каждый модуль отдельно — зелёный; модули в обратном порядке — 36 из 36. +- Тест-линтер после переноса находит `app/`: мутация (прямой `get_device` в `refresh_device`) → падение с точным местом; файл восстановлен. +- Стенд не пересобирался (тесты не входят в образ). + +## Оговорки +- Ссылки на `tests/test_app.py` в прежних отчётах ревью (`docs/reviews/*`) оставлены как есть — это исторические документы. diff --git a/tests/conftest.py b/tests/conftest.py index 9cdd387..d9a2c78 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -23,8 +23,7 @@ def env(tmp_path, monkeypatch): monkeypatch.setenv("ADMIN_USER", "admin") # не зависеть от реального .env monkeypatch.setenv("ADMIN_PASSWORD", ADMIN_PASSWORD) get_settings.cache_clear() - security.reset_failures("admin") # блокировки очистки журнала — в памяти процесса - security.reset_failures("login:testclient") # блокировки входа — в памяти процесса (IP TestClient по умолчанию) + security.reset_all_failures() # блокировки входа/очистки журнала — в памяти процесса, по всем ключам backups.invalidate() # кэш списка бакета — модульное состояние, не должен переживать тест db.init_db() yield diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..b9e3d97 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,15 @@ +"""Общие хелперы тестов: клиент RouterOS с мок-транспортом, минимальный ASGI-запрос.""" +import httpx +from fastapi import Request + +from app.ros.client import RosClient + + +def ros_client(handler) -> RosClient: + return RosClient("r1", 443, "admin", "pw", transport=httpx.MockTransport(handler)) + + +def _request(client_host: str | None, xff: str | None = None) -> Request: + """Минимальный ASGI-запрос для проверки security.client_ip без реального соединения.""" + headers = [(b"x-forwarded-for", xff.encode())] if xff else [] + return Request({"type": "http", "client": (client_host, 1) if client_host else None, "headers": headers}) diff --git a/tests/test_app.py b/tests/test_app.py deleted file mode 100644 index c15cc1d..0000000 --- a/tests/test_app.py +++ /dev/null @@ -1,990 +0,0 @@ -import ast -import asyncio -import html -import json -import re -import sqlite3 -from datetime import date, datetime, timedelta, timezone -from pathlib import Path - -import httpx -import pytest -from cryptography.fernet import Fernet -from fastapi import Request -from fastapi.testclient import TestClient - -from app import db, ids, process_lock, s3, security -from app.config import Settings, get_settings, insecure_settings -from app.main import create_app -from app.models import Backup, Device, Event, now -from app.db import session_scope -from app.ros import operations as ros -from app.ros.client import RosClient -from app.services import backups, devices, events, groups, jobs, ops, settings -from tests.conftest import ADMIN_PASSWORD, API_TOKEN - - -def ros_client(handler) -> RosClient: - return RosClient("r1", 443, "admin", "pw", transport=httpx.MockTransport(handler)) - - -def _request(client_host: str | None, xff: str | None = None) -> Request: - """Минимальный ASGI-запрос для проверки security.client_ip без реального соединения.""" - headers = [(b"x-forwarded-for", xff.encode())] if xff else [] - return Request({"type": "http", "client": (client_host, 1) if client_host else None, "headers": headers}) - - -def test_password_encryption_roundtrip(): - token = security.encrypt("s3cret") - assert token != "s3cret" and security.decrypt(token) == "s3cret" - - -@pytest.mark.asyncio -async def test_status_parsing(): - data = { - "/rest/system/resource": {"board-name": "hAP ax3", "version": "7.15 (stable)", "uptime": "1d2h"}, - "/rest/system/routerboard": {"routerboard": "true", "model": "C53UiG", "current-firmware": "7.14", "upgrade-firmware": "7.15"}, - "/rest/system/package/update": {"channel": "stable", "installed-version": "7.15", "latest-version": "7.16", "status": "New version is available"}, - "/rest/system/package/update/check-for-updates": [], # POST: проверка обновлений - } - async with ros_client(lambda r: httpx.Response(200, json=data[r.url.path])) as c: - st = await ros.get_status(c) - assert (st["model"], st["channel"], st["fw_upgrade"], st["ros_latest"]) == ("C53UiG", "stable", "7.15", "7.16") - assert st["ros_installed"] == "7.15" and st["ros_check_error"] is None - - -@pytest.mark.asyncio -async def test_backup_flow(monkeypatch): - calls, export_body = [], {} - - def handler(r: httpx.Request) -> httpx.Response: - calls.append(f"{r.method} {r.url.path}") - if r.url.path == "/rest/export": - export_body.update(json.loads(r.content)) - if r.url.path == "/rest/file" and r.method == "GET": - return httpx.Response(200, json=[{".id": "*1", "size": "5"}]) - if r.url.path == "/rest/execute": - return httpx.Response(200, json={"ret": "aGVsbG8="}) # base64("hello") - return httpx.Response(200, json=[]) - - uploaded = {} - - metas = {} - - async def fake_upload(path, key, metadata=None): # содержимое читаем до удаления временной папки - uploaded[key] = open(path, "rb").read() - metas[key] = metadata - - d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") - monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler)) - monkeypatch.setattr(s3, "upload_file", fake_upload) - - await ops.run_backup(d.id) - - posts = [c for c in calls if c.startswith("POST")] - assert posts == ["POST /rest/system/backup/save", "POST /rest/export", - "POST /rest/execute", "POST /rest/execute"] # по одному блоку на файл - assert "show-sensitive" in export_body - with session_scope() as s: # файлы в бакете связаны с метаданными: ID копии и устройства в S3-metadata - b = s.query(Backup).one() - assert ids.is_id(b.id, "bkp") and b.device_id == d.id and b.device_name == "r1" - assert all(m == {"backup-id": b.id, "device-id": d.id} for m in metas.values()) and len(metas) == 2 # .rsc с секретами: из него можно восстановить всё - assert calls.count("DELETE /rest/file/*1") == 2 # файлы удалены с устройства - assert sorted(uploaded.values()) == [b"hello", b"hello"] - assert all(k.startswith("backups/r1/") for k in uploaded) - with session_scope() as s: - assert s.query(Backup).one().status == "done" - assert devices.get_device(d.id).last_backup_requested_at is not None - - -def test_api_auth_and_no_password_leak(): - with TestClient(create_app()) as client: - assert client.get("/api/v1/devices").status_code == 401 - h = {"Authorization": f"Bearer {API_TOKEN}"} - r = client.post("/api/v1/devices", headers=h, json={ - "name": "r1", "host": "10.0.0.1", "username": "admin", "password": "pw"}) - assert r.status_code == 201 and "password" not in r.text - assert client.get("/api/v1/devices", headers=h).json()[0]["name"] == "r1" - - -def test_groups_and_device_filters(): - office, store = groups.create_group("Офис"), groups.create_group("Склад") - a = devices.create_device("a", "10.0.0.1", 80, "u", "p", group_id=office.id) - b = devices.create_device("b", "10.0.0.2", 80, "u", "p", group_id=store.id) - devices.create_device("c", "10.0.0.3", 80, "u", "p") - with session_scope() as s: - s.get(Device, a.id).online = True - s.get(Device, a.id).status_json = json.dumps({"ros_installed": "7.1", "ros_latest": "7.2", "channel": "stable"}) - s.get(Device, b.id).online = False - - def names(**kw): - return sorted(d.name for d in devices.filter_devices(devices.list_devices(), **kw)) - - assert names(group=str(office.id)) == ["a"] and names(group="none") == ["c"] - assert names(status="offline") == ["b", "c"] and names(updates="ros") == ["a"] - assert names(q="10.0.0.2") == ["b"] and names(channel="stable") == ["a"] - - groups.delete_group(office.id) # устройства остаются, но без группы - assert devices.get_device(a.id).group_id is None - - -@pytest.mark.asyncio -async def test_backup_filters(monkeypatch): - office = groups.create_group("Офис") - devices.create_device("a", "10.0.0.1", 80, "u", "p", group_id=office.id) - devices.create_device("c", "10.0.0.3", 80, "u", "p") - - def item(key, day): - return {"key": key, "size": 1, "last_modified": datetime(2026, 9, day, 12, tzinfo=timezone.utc)} - - items = [item("backups/a/1.backup", 10), item("backups/a/1.rsc", 10), - item("backups/c/2.rsc", 20), item("backups/gone/3.rsc", 20)] # gone — устройство удалено - - async def fake_list(device=None): - return [i for i in items if device is None or i["key"].startswith(f"backups/{device}/")] - - monkeypatch.setattr(s3, "list_backups", fake_list) - keys = lambda r: [i["key"] for i in r] - assert keys(await backups.list_backups(group=str(office.id))) == ["backups/a/1.backup", "backups/a/1.rsc"] - assert keys(await backups.list_backups(group="none", kind="rsc")) == ["backups/c/2.rsc", "backups/gone/3.rsc"] - assert keys(await backups.list_backups(date_from=date(2026, 9, 15))) == ["backups/c/2.rsc", "backups/gone/3.rsc"] - - -@pytest.mark.asyncio -async def test_firmware_reboots_on_log_message(): - """Перезагрузка — по записи в журнале (current-firmware до неё не меняется).""" - def run(log_at_poll, old_entry_time): - calls, polls = [], {"n": 0} - - def handler(r: httpx.Request) -> httpx.Response: - calls.append(f"{r.method} {r.url.path}") - data = { - "/rest/system/routerboard": {"routerboard": "true", "current-firmware": "7.22", "upgrade-firmware": "7.24"}, - "/rest/system/clock": {"date": "2026-09-19", "time": "12:00:00"}, - "/rest/system/resource": {"uptime": "1h"}, # загрузка была в 11:00 - } - if r.url.path == "/rest/log": - polls["n"] += 1 - log = [{".id": "*1", "time": old_entry_time, "message": ros.FW_DONE_MSG}] - if polls["n"] >= log_at_poll: - log.append({".id": "*9", "time": "2026-09-19 12:00:00", "message": ros.FW_DONE_MSG}) - return httpx.Response(200, json=log) - return httpx.Response(200, json=data.get(r.url.path, [])) - - async def go(): - async with ros_client(handler) as c: - return await ros.upgrade_firmware(c, wait_seconds=10, poll=0) - return calls, go() - - # 1) новая запись появилась на 3-м чтении журнала -> команда upgrade, затем сразу reboot - calls, coro = run(3, "2026-09-19 08:00:00") - await coro - assert calls.index("POST /rest/system/routerboard/upgrade") < calls.index("POST /rest/system/reboot") == len(calls) - 1 - - # 2) запись новее последней загрузки уже есть (прошивка записана, перезагрузки не было) -> сразу reboot - calls, coro = run(99, "2026-09-19 11:30:00") - await coro - assert "POST /rest/system/routerboard/upgrade" not in calls and calls[-1] == "POST /rest/system/reboot" - - -def test_create_device_via_ui_binds_group(monkeypatch): - """Добавление устройства из окна: привязка к существующей группе и создание группы «на месте».""" - async def no_poll(_device_id): # без обращения к реальному устройству - return None - - monkeypatch.setattr(ops, "refresh_status", no_poll) - office = groups.create_group("Офис") - form = dict(host="10.0.0.1", port="80", username="u", password="p") - htmx = {"HX-Request": "true"} - with TestClient(create_app()) as c: - c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) - # 1) существующая группа - r = c.post("/devices/new", data={**form, "name": "a", "group_id": str(office.id), "note": " Серверная, 2 этаж "}, - follow_redirects=False) - assert r.status_code == 303 - # имя устройства в таблице открывает окно изменения (без JS — запасную страницу) - page = c.get("/").text - assert re.search(r'class="dev-link" href="/devices/(dev_[0-9a-f-]+)/edit" hx-get="/ui/dialog/device/\1"', page) - # 2) «Новая группа…» из окна: группа создаётся и привязывается, окно просит обновить страницу - r = c.post("/devices/new", data={**form, "name": "b", "group_id": "__new__", "new_group": "Склад"}, headers=htmx) - assert r.headers["HX-Refresh"] == "true" - # 3) «Новая группа…» без названия: ошибка в окне, устройство не создаётся - r = c.post("/devices/new", data={**form, "name": "c", "group_id": "__new__", "new_group": ""}, headers=htmx) - assert "Введите название новой группы" in r.text and 'id="modal-title"' in r.text - - # окна групп открываются и показывают ошибку (дубликат) внутри окна - assert "Новая группа" in c.get("/ui/dialog/group").text - r = c.post("/groups/new", data={"name": "Склад"}, headers=htmx) - assert "уже существует" in r.text and 'id="modal-title"' in r.text - - by_name = {d.name: d for d in devices.list_devices()} - store = next(g for g in groups.list_groups() if g["name"] == "Склад") - assert by_name["a"].group_id == office.id and by_name["b"].group_id == store["id"] and "c" not in by_name - assert by_name["a"].note == "Серверная, 2 этаж" and by_name["b"].note is None # пробелы обрезаются, пусто -> None - - -@pytest.mark.asyncio -async def test_poll_marks_offline_and_recovers(monkeypatch): - """Фоновый опрос: недоступность видна сразу; после возвращения — полный опрос (версии обновились).""" - d = devices.create_device("r1", "10.0.0.1", 80, "u", "p") - up = {"on": False} - data = { - "/rest/system/resource": {"version": "7.24.4 (stable)", "uptime": "1m", "board-name": "hAP"}, - "/rest/system/routerboard": {"routerboard": "false"}, - "/rest/system/package/update": {"channel": "stable", "installed-version": "7.24.4", "latest-version": "7.24.4"}, - } - - def handler(r: httpx.Request) -> httpx.Response: - if not up["on"]: - raise httpx.ConnectError("connection refused") - 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) - assert devices.get_device(d.id).online is False - - up["on"] = True - await ops.poll_device(d.id) # был offline -> полный опрос - dev = devices.get_device(d.id) - assert dev.online is True and dev.status["ros_installed"] == "7.24.4" - - data["/rest/system/resource"]["uptime"] = "2m" - await ops.poll_device(d.id) # онлайн -> лёгкий опрос обновляет uptime, остальное сохраняется - dev = devices.get_device(d.id) - assert dev.status["uptime"] == "2m" and dev.status["ros_latest"] == "7.24.4" - - -def test_bulk_delete_backups(monkeypatch): - """Групповое удаление: чужой ключ отклоняет всю операцию; допустимые — удаляются (UI и API).""" - removed = [] - - async def fake_delete(key): - removed.append(key) - - async def empty_list(device=None): # после удаления сервис сверяет метаданные с бакетом - return [] - - monkeypatch.setattr(s3, "delete_object", fake_delete) - monkeypatch.setattr(s3, "list_backups", empty_list) - ok = ["backups/a/1.backup", "backups/a/1.rsc"] - with TestClient(create_app()) as c: - c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) - # UI: ключ вне префикса бэкапов -> ошибка, ничего не удалено - r = c.post("/backups/delete-many", data={"key": ok + ["other/secret.txt"]}, follow_redirects=False) - assert r.status_code == 400 and removed == [] - # UI: допустимые ключи -> возврат на страницу с итогом - r = c.post("/backups/delete-many", data={"key": ok, "next": "/backups?group=1"}, follow_redirects=False) - assert r.status_code == 303 and r.headers["location"] == "/backups?group=1&deleted=2&failed=0" - assert sorted(removed) == ok - # API - removed.clear() - h = {"Authorization": f"Bearer {API_TOKEN}"} - assert c.post("/api/v1/backups/delete", headers=h, json={"keys": ok}).json() == {"deleted": 2, "failed": 0} - assert c.post("/api/v1/backups/delete", headers=h, json={"keys": ["x/../y"]}).status_code == 400 - # UI: одиночное удаление — тот же сервис delete_many (метаданные и событие backup.deleted не теряются) - removed.clear() - r = c.post("/backups/delete", data={"key": ok[0]}, follow_redirects=False) - assert r.status_code == 303 and r.headers["location"] == "/backups?deleted=1&failed=0" - assert removed == [ok[0]] - - -@pytest.mark.asyncio -async def test_chr_status_and_version_compare(): - """CHR: раздела system/routerboard нет (HTTP 400) — это не сбой; «последняя» версия старше установленной — не обновление.""" - def handler(r: httpx.Request) -> httpx.Response: - if r.url.path == "/rest/system/routerboard": - return httpx.Response(400, json={"detail": "no such command or directory (routerboard)", "error": 400}) - data = { - "/rest/system/resource": {"board-name": "CHR Yandex epyc-9654", "version": "7.24 (stable)", "uptime": "1d"}, - "/rest/system/package/update": {"channel": "long-term", "installed-version": "7.24", "latest-version": "7.23.7"}, - } - return httpx.Response(200, json=data.get(r.url.path, [])) - - async with ros_client(handler) as c: - st = await ros.get_status(c) - assert await ros.upgrade_firmware(c) == "Устройство без RouterBOARD firmware (например, CHR) — пропущено" - assert st["model"].startswith("CHR") and st["fw_current"] is None - assert not ros.version_newer(st["ros_latest"], st["ros_installed"]) # 7.23.7 старше 7.24 - assert ros.version_newer("7.24.1", "7.24") and ros.version_newer("7.25", "7.25rc1") and not ros.version_newer("7.25rc1", "7.25") - - -def test_device_name_is_immutable(): - """Имя задаётся только при создании: API отклоняет смену, форма изменения имя игнорирует.""" - with TestClient(create_app()) as c: - c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) - h = {"Authorization": f"Bearer {API_TOKEN}"} - d = c.post("/api/v1/devices", headers=h, json={"name": "r1", "host": "10.0.0.1", "username": "u", "password": "p"}).json() - r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r2"}) - assert r.status_code == 400 and "нельзя изменить" in r.text - # то же имя — допустимо (клиенты часто присылают объект целиком); остальные поля меняются - r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r1", "host": "10.0.0.9"}) - assert r.status_code == 200 and r.json()["name"] == "r1" and r.json()["host"] == "10.0.0.9" - # форма изменения: присланное имя игнорируется, остальное сохраняется - r = c.post(f"/devices/{d['id']}/edit", data={"name": "hacked", "host": "10.0.0.7", "port": "80", "username": "u"}, - follow_redirects=False) - assert r.status_code == 303 - page = c.get(f"/ui/dialog/device/{d['id']}").text - assert "readonly" in page and 'name="name"' not in page - dev = devices.get_device(d["id"]) - assert (dev.name, dev.host) == ("r1", "10.0.0.7") - - -def test_ids_are_unique_typed_and_ordered(): - """Все ID глобально уникальны: не повторяются, типы не пересекаются, внутри типа растут со временем.""" - seen, last = set(), {} - for i in range(50_000): - prefix = list(ids.PREFIXES)[i % len(ids.PREFIXES)] - v = ids.new_id(prefix) - assert v not in seen and v > last.get(prefix, "") - assert ids.is_id(v, prefix) and not any(ids.is_id(v, o) for o in ids.PREFIXES if o != prefix) - seen.add(v) - last[prefix] = v - with pytest.raises(LookupError): # ID другого типа в пути/запросе отклоняется - ids.check(ids.new_id("job"), "dev") - with pytest.raises(LookupError): - devices.get_device("1") # числовые ID больше не существуют - - -LEGACY_DDL = """ -CREATE TABLE devices (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, host VARCHAR(255) NOT NULL, port INTEGER NOT NULL, - username VARCHAR(64) NOT NULL, password_enc TEXT NOT NULL, verify_tls BOOLEAN NOT NULL, created_at DATETIME NOT NULL, - online BOOLEAN, status_json TEXT NOT NULL, status_at DATETIME, last_error TEXT, last_backup_requested_at DATETIME, - use_tls BOOLEAN NOT NULL DEFAULT 1, group_id INTEGER, note TEXT, PRIMARY KEY (id), UNIQUE (name)); -CREATE TABLE jobs (id INTEGER NOT NULL, device_id INTEGER, device_name VARCHAR(64) NOT NULL, type VARCHAR(32) NOT NULL, - status VARCHAR(16) NOT NULL, message TEXT NOT NULL, created_at DATETIME NOT NULL, finished_at DATETIME, PRIMARY KEY (id)); -CREATE TABLE backups (id INTEGER NOT NULL, device_id INTEGER NOT NULL, requested_at DATETIME NOT NULL, status VARCHAR(16) NOT NULL, - key_binary VARCHAR(512), key_rsc VARCHAR(512), error TEXT, PRIMARY KEY (id), - FOREIGN KEY(device_id) REFERENCES devices (id) ON DELETE CASCADE); -CREATE TABLE device_groups (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, PRIMARY KEY (id), UNIQUE (name)); -""" - - -def test_migration_replaces_numeric_ids(tmp_path): - """БД старой схемы: числовые ID заменяются на dev_/grp_/job_/bkp_, ссылки и порядок сохраняются, повтор — без эффекта.""" - path = tmp_path / "legacy.db" - con = sqlite3.connect(path) - con.executescript(LEGACY_DDL) - con.executemany("INSERT INTO device_groups VALUES (?, ?)", [(1, "Alpha"), (2, "Beta")]) - for i, (name, grp) in enumerate((("r1", 1), ("r2", None)), start=1): - con.execute("INSERT INTO devices (id, name, host, port, username, password_enc, verify_tls, created_at, status_json, use_tls, group_id)" - " VALUES (?,?,?,?,?,?,?,?,?,?,?)", (i, name, "10.0.0.%d" % i, 80, "u", "enc", 0, "2026-09-1%d 10:00:00.000000" % i, "{}", 0, grp)) - # задачи: две по r1 (id 1), одна по уже удалённому устройству (id 99) - for i, dev in enumerate((1, 1, 99), start=1): - con.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?,?,?)", (i, dev, "n", "backup", "done", "ok", "2026-09-19 10:0%d:00.000000" % i, None)) - con.execute("INSERT INTO backups VALUES (1, 1, '2026-09-19 10:00:00.000000', 'done', 'backups/r1/a.backup', 'backups/r1/a.rsc', NULL)") - con.execute("INSERT INTO backups VALUES (2, 99, '2026-09-19 10:05:00.000000', 'done', 'backups/gone/b.backup', NULL, NULL)") - con.commit() - con.close() - - db.init_db(f"sqlite:///{path}") - con = sqlite3.connect(path) - assert con.execute("PRAGMA user_version").fetchone()[0] == 2 - counts = {t: con.execute(f"SELECT count(*) FROM {t}").fetchone()[0] for t in ("device_groups", "devices", "jobs", "backups")} - assert counts == {"device_groups": 2, "devices": 2, "jobs": 3, "backups": 2} - alpha = con.execute("SELECT id FROM device_groups WHERE name='Alpha'").fetchone()[0] - r1, r2 = (con.execute("SELECT id, group_id FROM devices WHERE name=?", (n,)).fetchone() for n in ("r1", "r2")) - assert ids.is_id(r1[0], "dev") and r1[1] == alpha and r2[1] is None # связь устройство → группа сохранена - job_dev = [r[0] for r in con.execute("SELECT device_id FROM jobs ORDER BY id")] - assert job_dev[0] == job_dev[1] == r1[0] and job_dev[2] not in (r1[0], r2[0]) # удалённое устройство — свой стабильный ID - assert [r[0] for r in con.execute("SELECT id FROM jobs ORDER BY created_at")] == [r[0] for r in con.execute("SELECT id FROM jobs ORDER BY id")] - bk = con.execute("SELECT device_id, device_name FROM backups ORDER BY requested_at").fetchall() - assert bk[0] == (r1[0], "r1") and bk[1][0] == job_dev[2] and bk[1][1] == "gone" # имя — из ключа S3 - every = [r[0] for t in ("device_groups", "devices", "jobs", "backups", "events") for r in con.execute(f"SELECT id FROM {t}")] - assert len(every) == len(set(every)) and all(ids.is_id(i) for i in every) # ни одного пересечения между таблицами - assert con.execute("SELECT type FROM events").fetchall() == [("system.migrated",)] - assert len(list(tmp_path.glob("legacy.db.bak-*"))) == 1 # копия БД перед миграцией - con.close() - db.init_db(f"sqlite:///{path}") # повторный запуск ничего не меняет - con = sqlite3.connect(path) - assert con.execute("SELECT count(*) FROM events").fetchone()[0] == 1 and len(list(tmp_path.glob("legacy.db.bak-*"))) == 1 - con.close() - - -LEGACY_DDL_NO_EXTRA_COLUMNS = """ -CREATE TABLE devices (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, host VARCHAR(255) NOT NULL, port INTEGER NOT NULL, - username VARCHAR(64) NOT NULL, password_enc TEXT NOT NULL, verify_tls BOOLEAN NOT NULL, created_at DATETIME NOT NULL, - online BOOLEAN, status_json TEXT NOT NULL, status_at DATETIME, last_error TEXT, last_backup_requested_at DATETIME, - PRIMARY KEY (id), UNIQUE (name)); -CREATE TABLE jobs (id INTEGER NOT NULL, device_id INTEGER, device_name VARCHAR(64) NOT NULL, type VARCHAR(32) NOT NULL, - status VARCHAR(16) NOT NULL, message TEXT NOT NULL, created_at DATETIME NOT NULL, finished_at DATETIME, PRIMARY KEY (id)); -CREATE TABLE backups (id INTEGER NOT NULL, device_id INTEGER NOT NULL, requested_at DATETIME NOT NULL, status VARCHAR(16) NOT NULL, - key_binary VARCHAR(512), key_rsc VARCHAR(512), error TEXT, PRIMARY KEY (id), - FOREIGN KEY(device_id) REFERENCES devices (id) ON DELETE CASCADE); -CREATE TABLE device_groups (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, PRIMARY KEY (id), UNIQUE (name)); -""" - - -def test_migration_adds_legacy_columns_before_id_migration(tmp_path): - """Схема до появления групп/TLS-настроек/примечаний (без use_tls/group_id/note): колонки добавляются - раньше миграции ID — _to_v1 читает их из старой таблицы.""" - path = tmp_path / "legacy2.db" - con = sqlite3.connect(path) - con.executescript(LEGACY_DDL_NO_EXTRA_COLUMNS) - con.execute("INSERT INTO devices (id, name, host, port, username, password_enc, verify_tls, created_at, status_json)" - " VALUES (1, 'r1', '10.0.0.1', 80, 'u', 'enc', 0, '2026-09-10 10:00:00.000000', '{}')") - con.commit() - con.close() - - db.init_db(f"sqlite:///{path}") - con = sqlite3.connect(path) - assert con.execute("PRAGMA user_version").fetchone()[0] == 2 - dev_id, use_tls, group_id, note = con.execute("SELECT id, use_tls, group_id, note FROM devices").fetchone() - assert ids.is_id(dev_id, "dev") and (use_tls, group_id, note) == (1, None, None) # значения по умолчанию - con.close() - - -@pytest.mark.asyncio -async def test_batch_channel_runs_as_jobs(monkeypatch): - """Групповая смена канала — фоновыми задачами: 202 + job_ids, задачи завершаются done, канал — в data job.created.""" - calls = [] - - async def fake_set_channel(device_id, channel): - calls.append((device_id, channel)) - - monkeypatch.setattr(ops, "set_channel", fake_set_channel) - d1 = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") - d2 = devices.create_device("r2", "10.0.0.2", 443, "admin", "pw") - - with TestClient(create_app()) as c: - h = {"Authorization": f"Bearer {API_TOKEN}"} - r = c.put("/api/v1/batch/channel", headers=h, json={"device_ids": [d1.id, d2.id], "channel": "testing"}) - assert r.status_code == 202 - job_ids = r.json()["job_ids"] - assert len(job_ids) == 2 - await asyncio.sleep(0.3) - - assert sorted(calls) == sorted([(d1.id, "testing"), (d2.id, "testing")]) - for jid in job_ids: - assert jobs.get_job(jid).status == "done" - created = [e for e in events.list_events(type_="job.created") if e.entity_id in job_ids] - assert len(created) == 2 and all(json.loads(e.data) == {"type": "set_channel", "channel": "testing"} for e in created) - # неверный канал отклоняется до создания задач - assert c.put("/api/v1/batch/channel", headers=h, json={"device_ids": [d1.id], "channel": "bogus"}).status_code == 422 - - -def test_ros_state_values(): - """Состояния колонки «Upgrade ROS»: unknown / check_error / update / downgrade / current.""" - assert devices.ros_state({}) == "unknown" - assert devices.ros_state({"ros_installed": "7.24.4"}) == "unknown" # нет latest — проверка ещё не выполнялась - assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.16", "ros_check_error": "timeout"}) == "check_error" - assert devices.ros_state({"ros_installed": "7.15", "ros_latest": "7.16"}) == "update" - assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.23.7"}) == "downgrade" # long-term старше установленной - assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.24.4"}) == "current" - - -@pytest.mark.asyncio -async def test_run_ros_downgrade_order_and_failures(monkeypatch): - """run_ros_downgrade: версия не совпала с подтверждённой — бэкап и install не вызываются; версия совпала — - бэкап выполняется до install (порядок); бэкап упал — install не вызывается.""" - calls = [] - - def handler(r: httpx.Request) -> httpx.Response: - if r.url.path == "/rest/system/package/update/install": - calls.append("install") - return httpx.Response(200, json=[]) - data = {"/rest/system/package/update": {"installed-version": "7.24.4", "latest-version": "7.23.7"}} - return httpx.Response(200, json=data.get(r.url.path, [])) - - d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") - monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler)) - - async def fake_backup_ok(device_id): - calls.append("backup") - return "Бэкап загружен в S3: k.backup, k.rsc" - - async def fake_backup_fail(device_id): - calls.append("backup") - raise RuntimeError("бэкап не удался") - - # 1) версия канала (7.23.7) не совпала с подтверждённой пользователем -> задача failed, ничего не вызывается - monkeypatch.setattr(ops, "run_backup", fake_backup_ok) - [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.6"}) - await asyncio.sleep(0.2) - j = jobs.get_job(jid) - assert j.status == "failed" and "не совпадает" in j.message and calls == [] - - # 2) версия совпала -> бэкап выполняется до install, install вызван - [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.7"}) - await asyncio.sleep(0.2) - j = jobs.get_job(jid) - assert j.status == "done" and calls == ["backup", "install"] - - # 3) бэкап падает -> install не вызывается - calls.clear() - monkeypatch.setattr(ops, "run_backup", fake_backup_fail) - [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.7"}) - await asyncio.sleep(0.2) - j = jobs.get_job(jid) - assert j.status == "failed" and calls == ["backup"] - - -@pytest.mark.asyncio -async def test_downgrade_api_endpoints(monkeypatch): - """target_version обязателен для одного устройства и группы; /batch/ros_downgrade — отдельный типизированный - эндпоинт (не через общий /batch/{action}, у которого в Literal ros_downgrade нет).""" - async def fake_downgrade(device_id, target_version): - return f"откат до {target_version}" - - monkeypatch.setitem(jobs.JOB_TYPES, "ros_downgrade", fake_downgrade) - d1 = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") - d2 = devices.create_device("r2", "10.0.0.2", 443, "admin", "pw") - - with TestClient(create_app()) as c: - h = {"Authorization": f"Bearer {API_TOKEN}"} - assert c.post(f"/api/v1/devices/{d1.id}/update/downgrade", headers=h, json={}).status_code == 422 - r = c.post(f"/api/v1/devices/{d1.id}/update/downgrade", headers=h, json={"target_version": "7.23.7"}) - assert r.status_code == 202 - job_ids = r.json()["job_ids"] - - # тело без target_version — если бы запрос ушёл в общий /batch/{action} (там target_version не нужен), - # он был бы принят; 422 подтверждает, что обрабатывает именно типизированный /batch/ros_downgrade - assert c.post("/api/v1/batch/ros_downgrade", headers=h, json={"device_ids": [d1.id]}).status_code == 422 - r = c.post("/api/v1/batch/ros_downgrade", headers=h, - json={"device_ids": [d1.id, d2.id], "target_version": "7.23.7"}) - assert r.status_code == 202 - job_ids += r.json()["job_ids"] - assert len(job_ids) == 3 - - await asyncio.sleep(0.3) - for jid in job_ids: - j = jobs.get_job(jid) - assert j.type == "ros_downgrade" and j.status == "done" and j.message == "откат до 7.23.7" - - -def test_ui_downgrade_dialog_and_mismatch(monkeypatch): - """Окно отката: неактивная кнопка, пока версия не введена (data-enables); несовпадающая версия — ошибка - в окне, задача не создаётся.""" - d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") - - def _set_status(device, status): - with session_scope() as s: - s.get(Device, device.id).status_json = json.dumps(status) - - _set_status(d, {"channel": "long-term", "ros_installed": "7.24.4", "ros_latest": "7.23.7"}) - - with TestClient(create_app()) as c: - c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) - htmx = {"HX-Request": "true"} - page = c.get(f"/ui/dialog/downgrade?device_ids={d.id}", headers=htmx).text - assert "r1" in page and 'data-enables="#downgrade-confirm"' in page and 'id="downgrade-confirm" disabled' in page - - r = c.post("/ui/downgrade", data={"device_ids": [d.id], "target_version": "7.16"}, headers=htmx) - assert "не совпадает" in r.text and 'id="modal-title"' in r.text - assert jobs.list_jobs() == [] - - r = c.post("/ui/downgrade", data={"device_ids": [d.id], "target_version": "7.23.7"}, headers=htmx) - assert r.headers.get("HX-Refresh") == "true" - assert len(jobs.list_jobs()) == 1 and jobs.list_jobs()[0].type == "ros_downgrade" - - -@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 - - -@pytest.mark.asyncio -async def test_backup_files_get_ids_and_deletion_is_recorded(monkeypatch): - """Файлы бакета без метаданных получают bkp_ID (парой .backup/.rsc); пропавшие из бакета — помечаются удалёнными.""" - items = [{"key": f"backups/r1/{s}", "size": 1, "last_modified": datetime(2026, 9, 10, 12, tzinfo=timezone.utc)} - for s in ("20260910-1.backup", "20260910-1.rsc")] - listing = {"items": items} - - async def fake_list(device=None): - return listing["items"] - - monkeypatch.setattr(s3, "list_backups", fake_list) - found, _ = await backups.search() - assert len({i["backup_id"] for i in found}) == 1 and ids.is_id(found[0]["backup_id"], "bkp") # одна копия = пара файлов - again, _ = await backups.search() - assert again[0]["backup_id"] == found[0]["backup_id"] # повторный вызов (из кэша) ID не меняет и дубликатов не создаёт - - listing["items"] = [] - await backups.search(refresh=True) # изменения бакета мимо кэша — принудительное обновление - with session_scope() as s: - b = s.query(Backup).one() - assert b.deleted_at is not None and b.id == found[0]["backup_id"] - assert [e.type for e in events.list_events(entity_id=b.id)] == ["backup.deleted", "backup.imported"] - - -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 - - -@pytest.mark.asyncio -async def test_bucket_list_is_cached_between_reads(monkeypatch): - """Подряд идущие чтения бакета не обращаются к S3 повторно; refresh=True и invalidate() — обращаются.""" - calls = [] - - async def fake_list(device=None): - calls.append(1) - return [] - - monkeypatch.setattr(s3, "list_backups", fake_list) - await backups.search() - await backups.search() - assert len(calls) == 1 # второй вызов — из кэша - - await backups.search(refresh=True) - assert len(calls) == 2 # принудительное обновление, минуя кэш - - backups.invalidate() - await backups.search() - assert len(calls) == 3 # invalidate() сбрасывает кэш - - # invalidate() пришедший, пока чтение бакета уже шло (гонка: run_backup/delete_many завершились - # во время открытой страницы «Бэкапы») не должен теряться — следующий search() обязан перечитать бакет - async def fake_list_race(device=None): - calls.append(1) - if len(calls) == 4: # ровно один раз, при первом чтении в этом сценарии - backups.invalidate() - return [] - - monkeypatch.setattr(s3, "list_backups", fake_list_race) - backups.invalidate() # кэш после предыдущего шага свежий — начать сценарий с чтения бакета - await backups.search() - assert len(calls) == 4 - await backups.search() - assert len(calls) == 5 # кэш не помечен свежим из-за invalidate() во время предыдущего чтения - - -def test_process_lock_blocks_second_process(tmp_path): - """Второй процесс на той же файловой БД не стартует; после освобождения блокировки — снова можно.""" - db_url = f"sqlite:///{tmp_path}/lock.db" - process_lock.acquire(db_url) - try: - with pytest.raises(RuntimeError, match="уже используется"): - process_lock.acquire(db_url) - finally: - process_lock.release() - process_lock.acquire(db_url) # после освобождения — успешно - process_lock.release() - - -def test_insecure_settings_rejects_weak_secrets_without_leaking_them(): - """Небезопасные секреты выявляются по каждому правилу; значения секретов в описание проблем не попадают.""" - secret = "s3cr3t-value-must-not-leak-anywhere" - bad = Settings(api_token="change-me", session_secret=secret[:10], admin_password=secret[:8], secret_key=secret) - problems = insecure_settings(bad) - assert len(problems) == 4 # все четыре секрета нарушают правила - text = " ".join(problems) - assert secret not in text and secret[:10] not in text and secret[:8] not in text - - ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, - secret_key=Fernet.generate_key().decode()) - assert insecure_settings(ok) == [] - - -def test_insecure_settings_rejects_invalid_trusted_proxies(): - """TRUSTED_PROXIES с записью не в формате CIDR — тоже отказ старта; значение (не секрет) видно в тексте проблемы.""" - ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, - secret_key=Fernet.generate_key().decode(), trusted_proxies="10.0.0.0/8, not-a-network") - problems = insecure_settings(ok) - assert len(problems) == 1 and "TRUSTED_PROXIES" in problems[0] and "not-a-network" in problems[0] - - -def test_client_ip_trusted_proxies(monkeypatch): - """X-Forwarded-For учитывается, только если TRUSTED_PROXIES не пуст и peer — из доверенной сети (иначе подделка - заголовка бессмысленна); цепочка разбирается справа налево до первого недоверенного адреса; невалидная запись - обрывает разбор на последнем валидном; работает и с IPv6.""" - monkeypatch.setenv("TRUSTED_PROXIES", "") - get_settings.cache_clear() - assert security.client_ip(_request("1.2.3.4", "9.9.9.9")) == "1.2.3.4" # пусто — XFF полностью игнорируется - - monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") - get_settings.cache_clear() - assert security.client_ip(_request("9.9.9.9", "9.9.9.9")) == "9.9.9.9" # peer не доверенный — XFF игнорируется - assert security.client_ip(_request("10.0.0.1", "1.2.3.4, 10.0.0.2")) == "1.2.3.4" # доверенный прокси пропущен - assert security.client_ip(_request("10.0.0.1", "10.0.0.3, 10.0.0.2")) == "10.0.0.1" # все доверенные — peer - assert security.client_ip(_request("10.0.0.1", "garbage, 10.0.0.2")) == "10.0.0.2" # мусор — последний валидный - assert security.client_ip(_request("10.0.0.1", "garbage")) == "10.0.0.1" # мусор без валидных перед ним — peer - assert security.client_ip(_request("10.0.0.1")) == "10.0.0.1" # заголовка нет - assert security.client_ip(_request("10.0.0.1", "2001:db8::1, 10.0.0.2")) == "2001:db8::1" # IPv6 - get_settings.cache_clear() - - -def test_app_refuses_to_start_with_insecure_config(monkeypatch): - """Приложение не стартует с небезопасной конфигурацией (например, API_TOKEN=change-me).""" - monkeypatch.setenv("API_TOKEN", "change-me") - get_settings.cache_clear() - with pytest.raises(RuntimeError, match="Небезопасная конфигурация"): - with TestClient(create_app()): - pass - get_settings.cache_clear() - - -def test_login_lockout_by_ip(): - """5 неверных попыток входа с одного IP блокируют его на 6-ю (пароль уже не проверяется, код 429); - повторные попытки во время блокировки не засоряют журнал новыми auth.locked; блокировка не распространяется - на другой IP; успешный вход сбрасывает счётчик неудач.""" - with TestClient(create_app()) as c: - for _ in range(5): - r = c.post("/login", data={"username": "admin", "password": "wrong"}) - assert r.status_code == 401 - r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) # верный пароль уже не спасает - assert r.status_code == 429 and "Слишком много попыток" in r.text - assert len(events.list_events(type_="auth.locked")) == 1 - for _ in range(3): # попытки во время блокировки — 429, но новых auth.locked не пишут (не засоряют журнал) - assert c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}).status_code == 429 - assert len(events.list_events(type_="auth.locked")) == 1 - - with TestClient(create_app(), client=("10.0.0.2", 1)) as c2: - r = c2.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) - assert r.status_code == 303 # другой IP свободен - - with TestClient(create_app(), client=("10.0.0.3", 1)) as c3: - for _ in range(3): - assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 - r = c3.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) - assert r.status_code == 303 - for _ in range(4): # если бы счётчик не сбросился успехом, пятая по счёту неудача (2-я в этом цикле) заблокировала бы - assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 - - -def test_login_lockout_by_ip_behind_trusted_proxy(monkeypatch): - """За доверенным прокси блокировка и событие auth.* считаются по адресу из X-Forwarded-For, а не по адресу - прокси: 5 неверных попыток с одним XFF блокируют этот адрес, попытка с другим XFF (тот же прокси-peer) проходит.""" - monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") - get_settings.cache_clear() - with TestClient(create_app(), client=("10.0.0.5", 1)) as c: - for _ in range(5): - r = c.post("/login", data={"username": "admin", "password": "wrong"}, headers={"X-Forwarded-For": "1.1.1.1"}) - assert r.status_code == 401 - r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "1.1.1.1"}) - assert r.status_code == 429 # заблокирован именно реальный адрес клиента - assert json.loads(events.list_events(type_="auth.locked")[0].data)["ip"] == "1.1.1.1" - - r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "2.2.2.2"}, - follow_redirects=False) - assert r.status_code == 303 # другой адрес за тем же прокси не заблокирован - get_settings.cache_clear() - - -def test_move_redirect_rejects_open_redirect_next(): - """`next` в /ui/move принимает только локальный путь — иначе редирект на «/» (открытый редирект).""" - with TestClient(create_app()) as c: - c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) - for bad in ("/\\evil.com", "//evil.com", "https://evil.com", "/\r\nX"): # последний — декодированный /%0d%0aX - r = c.post("/ui/move", data={"next": bad}, follow_redirects=False) - assert r.headers["location"] == "/" - r = c.post("/ui/move", data={"next": "/?f_group=none"}, follow_redirects=False) - assert r.headers["location"] == "/?f_group=none" # обычный путь с фильтром сохраняется - - -def test_no_sync_db_calls_in_async_functions(): - """Регресс п.9 ревью: в async-функциях app/ не должно быть прямых вызовов синхронных функций, - обращающихся к БД (session_scope) — напрямую или через один уровень вызовов, — это блокирует event loop. - Правильный способ — asyncio.to_thread(fn, ...). Ссылка на функцию, переданная в asyncio.to_thread(fn, ...), - вызовом не считается; тела вложенных def/async def в область видимости внешней функции не входят. - Единственное исключение — app.main.lifespan (выполняется до приёма запросов).""" - app_dir = Path(__file__).resolve().parent.parent / "app" - allowed = {"lifespan"} - - class _OwnScope(ast.NodeVisitor): - """Вызовы прямо в теле функции, не заходя в тела вложенных def/async def.""" - - def __init__(self): - self.calls: list[ast.Call] = [] - - def visit_FunctionDef(self, node): # не спускаемся во вложенную функцию - pass - - def visit_AsyncFunctionDef(self, node): - pass - - def visit_Call(self, node): - self.calls.append(node) - self.generic_visit(node) - - def own_calls(func) -> list[ast.Call]: - v = _OwnScope() - for stmt in func.body: - v.visit(stmt) - return v.calls - - def call_name(call: ast.Call) -> str | None: - f = call.func - if isinstance(f, ast.Name): - return f.id - if isinstance(f, ast.Attribute): - return f.attr - return None - - sync_funcs, async_funcs = [], [] - for path in sorted(app_dir.rglob("*.py")): - tree = ast.parse(path.read_text(), filename=str(path)) - for node in ast.walk(tree): - if isinstance(node, ast.AsyncFunctionDef): - async_funcs.append((path, node)) - elif isinstance(node, ast.FunctionDef): - sync_funcs.append((path, node)) - - # функции, обращающиеся к БД напрямую, и функции, вызывающие их (один уровень) - level0 = {f.name for _, f in sync_funcs if any(call_name(c) == "session_scope" for c in own_calls(f))} - level1 = {f.name for _, f in sync_funcs - if f.name not in level0 and any(call_name(c) in level0 for c in own_calls(f))} - dangerous = level0 | level1 - - violations = [] - for path, func in async_funcs: - if func.name in allowed: - continue - for c in own_calls(func): - name = call_name(c) - if name in dangerous: - violations.append(f"{path.relative_to(app_dir.parent)}::{func.name} вызывает {name}() напрямую, " - "в обход asyncio.to_thread — обращение к БД блокирует event loop") - assert not violations, "\n".join(violations) diff --git a/tests/test_architecture.py b/tests/test_architecture.py new file mode 100644 index 0000000..17f9185 --- /dev/null +++ b/tests/test_architecture.py @@ -0,0 +1,68 @@ +import ast +from pathlib import Path + + +def test_no_sync_db_calls_in_async_functions(): + """Регресс п.9 ревью: в async-функциях app/ не должно быть прямых вызовов синхронных функций, + обращающихся к БД (session_scope) — напрямую или через один уровень вызовов, — это блокирует event loop. + Правильный способ — asyncio.to_thread(fn, ...). Ссылка на функцию, переданная в asyncio.to_thread(fn, ...), + вызовом не считается; тела вложенных def/async def в область видимости внешней функции не входят. + Единственное исключение — app.main.lifespan (выполняется до приёма запросов).""" + app_dir = Path(__file__).resolve().parent.parent / "app" + allowed = {"lifespan"} + + class _OwnScope(ast.NodeVisitor): + """Вызовы прямо в теле функции, не заходя в тела вложенных def/async def.""" + + def __init__(self): + self.calls: list[ast.Call] = [] + + def visit_FunctionDef(self, node): # не спускаемся во вложенную функцию + pass + + def visit_AsyncFunctionDef(self, node): + pass + + def visit_Call(self, node): + self.calls.append(node) + self.generic_visit(node) + + def own_calls(func) -> list[ast.Call]: + v = _OwnScope() + for stmt in func.body: + v.visit(stmt) + return v.calls + + def call_name(call: ast.Call) -> str | None: + f = call.func + if isinstance(f, ast.Name): + return f.id + if isinstance(f, ast.Attribute): + return f.attr + return None + + sync_funcs, async_funcs = [], [] + for path in sorted(app_dir.rglob("*.py")): + tree = ast.parse(path.read_text(), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFunctionDef): + async_funcs.append((path, node)) + elif isinstance(node, ast.FunctionDef): + sync_funcs.append((path, node)) + + # функции, обращающиеся к БД напрямую, и функции, вызывающие их (один уровень) + level0 = {f.name for _, f in sync_funcs if any(call_name(c) == "session_scope" for c in own_calls(f))} + level1 = {f.name for _, f in sync_funcs + if f.name not in level0 and any(call_name(c) in level0 for c in own_calls(f))} + dangerous = level0 | level1 + + violations = [] + for path, func in async_funcs: + if func.name in allowed: + continue + for c in own_calls(func): + name = call_name(c) + if name in dangerous: + violations.append(f"{path.relative_to(app_dir.parent)}::{func.name} вызывает {name}() напрямую, " + "в обход asyncio.to_thread — обращение к БД блокирует event loop") + assert not violations, "\n".join(violations) diff --git a/tests/test_backups.py b/tests/test_backups.py new file mode 100644 index 0000000..6f73299 --- /dev/null +++ b/tests/test_backups.py @@ -0,0 +1,175 @@ +import json +from datetime import date, datetime, timezone + +import httpx +import pytest +from fastapi.testclient import TestClient + +from app import ids, s3 +from app.db import session_scope +from app.main import create_app +from app.models import Backup +from app.services import backups, devices, events, groups, ops +from tests.conftest import ADMIN_PASSWORD, API_TOKEN +from tests.helpers import ros_client + + +@pytest.mark.asyncio +async def test_backup_flow(monkeypatch): + calls, export_body = [], {} + + def handler(r: httpx.Request) -> httpx.Response: + calls.append(f"{r.method} {r.url.path}") + if r.url.path == "/rest/export": + export_body.update(json.loads(r.content)) + if r.url.path == "/rest/file" and r.method == "GET": + return httpx.Response(200, json=[{".id": "*1", "size": "5"}]) + if r.url.path == "/rest/execute": + return httpx.Response(200, json={"ret": "aGVsbG8="}) # base64("hello") + return httpx.Response(200, json=[]) + + uploaded = {} + + metas = {} + + async def fake_upload(path, key, metadata=None): # содержимое читаем до удаления временной папки + uploaded[key] = open(path, "rb").read() + metas[key] = metadata + + d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") + monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler)) + monkeypatch.setattr(s3, "upload_file", fake_upload) + + await ops.run_backup(d.id) + + posts = [c for c in calls if c.startswith("POST")] + assert posts == ["POST /rest/system/backup/save", "POST /rest/export", + "POST /rest/execute", "POST /rest/execute"] # по одному блоку на файл + assert "show-sensitive" in export_body + with session_scope() as s: # файлы в бакете связаны с метаданными: ID копии и устройства в S3-metadata + b = s.query(Backup).one() + assert ids.is_id(b.id, "bkp") and b.device_id == d.id and b.device_name == "r1" + assert all(m == {"backup-id": b.id, "device-id": d.id} for m in metas.values()) and len(metas) == 2 # .rsc с секретами: из него можно восстановить всё + assert calls.count("DELETE /rest/file/*1") == 2 # файлы удалены с устройства + assert sorted(uploaded.values()) == [b"hello", b"hello"] + assert all(k.startswith("backups/r1/") for k in uploaded) + with session_scope() as s: + assert s.query(Backup).one().status == "done" + assert devices.get_device(d.id).last_backup_requested_at is not None + + +@pytest.mark.asyncio +async def test_backup_filters(monkeypatch): + office = groups.create_group("Офис") + devices.create_device("a", "10.0.0.1", 80, "u", "p", group_id=office.id) + devices.create_device("c", "10.0.0.3", 80, "u", "p") + + def item(key, day): + return {"key": key, "size": 1, "last_modified": datetime(2026, 9, day, 12, tzinfo=timezone.utc)} + + items = [item("backups/a/1.backup", 10), item("backups/a/1.rsc", 10), + item("backups/c/2.rsc", 20), item("backups/gone/3.rsc", 20)] # gone — устройство удалено + + async def fake_list(device=None): + return [i for i in items if device is None or i["key"].startswith(f"backups/{device}/")] + + monkeypatch.setattr(s3, "list_backups", fake_list) + keys = lambda r: [i["key"] for i in r] + assert keys(await backups.list_backups(group=str(office.id))) == ["backups/a/1.backup", "backups/a/1.rsc"] + assert keys(await backups.list_backups(group="none", kind="rsc")) == ["backups/c/2.rsc", "backups/gone/3.rsc"] + assert keys(await backups.list_backups(date_from=date(2026, 9, 15))) == ["backups/c/2.rsc", "backups/gone/3.rsc"] + + +def test_bulk_delete_backups(monkeypatch): + """Групповое удаление: чужой ключ отклоняет всю операцию; допустимые — удаляются (UI и API).""" + removed = [] + + async def fake_delete(key): + removed.append(key) + + async def empty_list(device=None): # после удаления сервис сверяет метаданные с бакетом + return [] + + monkeypatch.setattr(s3, "delete_object", fake_delete) + monkeypatch.setattr(s3, "list_backups", empty_list) + ok = ["backups/a/1.backup", "backups/a/1.rsc"] + with TestClient(create_app()) as c: + c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) + # UI: ключ вне префикса бэкапов -> ошибка, ничего не удалено + r = c.post("/backups/delete-many", data={"key": ok + ["other/secret.txt"]}, follow_redirects=False) + assert r.status_code == 400 and removed == [] + # UI: допустимые ключи -> возврат на страницу с итогом + r = c.post("/backups/delete-many", data={"key": ok, "next": "/backups?group=1"}, follow_redirects=False) + assert r.status_code == 303 and r.headers["location"] == "/backups?group=1&deleted=2&failed=0" + assert sorted(removed) == ok + # API + removed.clear() + h = {"Authorization": f"Bearer {API_TOKEN}"} + assert c.post("/api/v1/backups/delete", headers=h, json={"keys": ok}).json() == {"deleted": 2, "failed": 0} + assert c.post("/api/v1/backups/delete", headers=h, json={"keys": ["x/../y"]}).status_code == 400 + # UI: одиночное удаление — тот же сервис delete_many (метаданные и событие backup.deleted не теряются) + removed.clear() + r = c.post("/backups/delete", data={"key": ok[0]}, follow_redirects=False) + assert r.status_code == 303 and r.headers["location"] == "/backups?deleted=1&failed=0" + assert removed == [ok[0]] + + +@pytest.mark.asyncio +async def test_backup_files_get_ids_and_deletion_is_recorded(monkeypatch): + """Файлы бакета без метаданных получают bkp_ID (парой .backup/.rsc); пропавшие из бакета — помечаются удалёнными.""" + items = [{"key": f"backups/r1/{s}", "size": 1, "last_modified": datetime(2026, 9, 10, 12, tzinfo=timezone.utc)} + for s in ("20260910-1.backup", "20260910-1.rsc")] + listing = {"items": items} + + async def fake_list(device=None): + return listing["items"] + + monkeypatch.setattr(s3, "list_backups", fake_list) + found, _ = await backups.search() + assert len({i["backup_id"] for i in found}) == 1 and ids.is_id(found[0]["backup_id"], "bkp") # одна копия = пара файлов + again, _ = await backups.search() + assert again[0]["backup_id"] == found[0]["backup_id"] # повторный вызов (из кэша) ID не меняет и дубликатов не создаёт + + listing["items"] = [] + await backups.search(refresh=True) # изменения бакета мимо кэша — принудительное обновление + with session_scope() as s: + b = s.query(Backup).one() + assert b.deleted_at is not None and b.id == found[0]["backup_id"] + assert [e.type for e in events.list_events(entity_id=b.id)] == ["backup.deleted", "backup.imported"] + + +@pytest.mark.asyncio +async def test_bucket_list_is_cached_between_reads(monkeypatch): + """Подряд идущие чтения бакета не обращаются к S3 повторно; refresh=True и invalidate() — обращаются.""" + calls = [] + + async def fake_list(device=None): + calls.append(1) + return [] + + monkeypatch.setattr(s3, "list_backups", fake_list) + await backups.search() + await backups.search() + assert len(calls) == 1 # второй вызов — из кэша + + await backups.search(refresh=True) + assert len(calls) == 2 # принудительное обновление, минуя кэш + + backups.invalidate() + await backups.search() + assert len(calls) == 3 # invalidate() сбрасывает кэш + + # invalidate() пришедший, пока чтение бакета уже шло (гонка: run_backup/delete_many завершились + # во время открытой страницы «Бэкапы») не должен теряться — следующий search() обязан перечитать бакет + async def fake_list_race(device=None): + calls.append(1) + if len(calls) == 4: # ровно один раз, при первом чтении в этом сценарии + backups.invalidate() + return [] + + monkeypatch.setattr(s3, "list_backups", fake_list_race) + backups.invalidate() # кэш после предыдущего шага свежий — начать сценарий с чтения бакета + await backups.search() + assert len(calls) == 4 + await backups.search() + assert len(calls) == 5 # кэш не помечен свежим из-за invalidate() во время предыдущего чтения diff --git a/tests/test_devices.py b/tests/test_devices.py new file mode 100644 index 0000000..191107e --- /dev/null +++ b/tests/test_devices.py @@ -0,0 +1,167 @@ +import json +import re + +import httpx +import pytest +from fastapi.testclient import TestClient + +from app.db import session_scope +from app.main import create_app +from app.models import Device +from app.ros import operations as ros +from app.services import devices, groups, ops +from tests.conftest import ADMIN_PASSWORD, API_TOKEN +from tests.helpers import ros_client + + +@pytest.mark.asyncio +async def test_status_parsing(): + data = { + "/rest/system/resource": {"board-name": "hAP ax3", "version": "7.15 (stable)", "uptime": "1d2h"}, + "/rest/system/routerboard": {"routerboard": "true", "model": "C53UiG", "current-firmware": "7.14", "upgrade-firmware": "7.15"}, + "/rest/system/package/update": {"channel": "stable", "installed-version": "7.15", "latest-version": "7.16", "status": "New version is available"}, + "/rest/system/package/update/check-for-updates": [], # POST: проверка обновлений + } + async with ros_client(lambda r: httpx.Response(200, json=data[r.url.path])) as c: + st = await ros.get_status(c) + assert (st["model"], st["channel"], st["fw_upgrade"], st["ros_latest"]) == ("C53UiG", "stable", "7.15", "7.16") + assert st["ros_installed"] == "7.15" and st["ros_check_error"] is None + + +@pytest.mark.asyncio +async def test_chr_status_and_version_compare(): + """CHR: раздела system/routerboard нет (HTTP 400) — это не сбой; «последняя» версия старше установленной — не обновление.""" + def handler(r: httpx.Request) -> httpx.Response: + if r.url.path == "/rest/system/routerboard": + return httpx.Response(400, json={"detail": "no such command or directory (routerboard)", "error": 400}) + data = { + "/rest/system/resource": {"board-name": "CHR Yandex epyc-9654", "version": "7.24 (stable)", "uptime": "1d"}, + "/rest/system/package/update": {"channel": "long-term", "installed-version": "7.24", "latest-version": "7.23.7"}, + } + return httpx.Response(200, json=data.get(r.url.path, [])) + + async with ros_client(handler) as c: + st = await ros.get_status(c) + assert await ros.upgrade_firmware(c) == "Устройство без RouterBOARD firmware (например, CHR) — пропущено" + assert st["model"].startswith("CHR") and st["fw_current"] is None + assert not ros.version_newer(st["ros_latest"], st["ros_installed"]) # 7.23.7 старше 7.24 + assert ros.version_newer("7.24.1", "7.24") and ros.version_newer("7.25", "7.25rc1") and not ros.version_newer("7.25rc1", "7.25") + + +def test_groups_and_device_filters(): + office, store = groups.create_group("Офис"), groups.create_group("Склад") + a = devices.create_device("a", "10.0.0.1", 80, "u", "p", group_id=office.id) + b = devices.create_device("b", "10.0.0.2", 80, "u", "p", group_id=store.id) + devices.create_device("c", "10.0.0.3", 80, "u", "p") + with session_scope() as s: + s.get(Device, a.id).online = True + s.get(Device, a.id).status_json = json.dumps({"ros_installed": "7.1", "ros_latest": "7.2", "channel": "stable"}) + s.get(Device, b.id).online = False + + def names(**kw): + return sorted(d.name for d in devices.filter_devices(devices.list_devices(), **kw)) + + assert names(group=str(office.id)) == ["a"] and names(group="none") == ["c"] + assert names(status="offline") == ["b", "c"] and names(updates="ros") == ["a"] + assert names(q="10.0.0.2") == ["b"] and names(channel="stable") == ["a"] + + groups.delete_group(office.id) # устройства остаются, но без группы + assert devices.get_device(a.id).group_id is None + + +def test_create_device_via_ui_binds_group(monkeypatch): + """Добавление устройства из окна: привязка к существующей группе и создание группы «на месте».""" + async def no_poll(_device_id): # без обращения к реальному устройству + return None + + monkeypatch.setattr(ops, "refresh_status", no_poll) + office = groups.create_group("Офис") + form = dict(host="10.0.0.1", port="80", username="u", password="p") + htmx = {"HX-Request": "true"} + with TestClient(create_app()) as c: + c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) + # 1) существующая группа + r = c.post("/devices/new", data={**form, "name": "a", "group_id": str(office.id), "note": " Серверная, 2 этаж "}, + follow_redirects=False) + assert r.status_code == 303 + # имя устройства в таблице открывает окно изменения (без JS — запасную страницу) + page = c.get("/").text + assert re.search(r'class="dev-link" href="/devices/(dev_[0-9a-f-]+)/edit" hx-get="/ui/dialog/device/\1"', page) + # 2) «Новая группа…» из окна: группа создаётся и привязывается, окно просит обновить страницу + r = c.post("/devices/new", data={**form, "name": "b", "group_id": "__new__", "new_group": "Склад"}, headers=htmx) + assert r.headers["HX-Refresh"] == "true" + # 3) «Новая группа…» без названия: ошибка в окне, устройство не создаётся + r = c.post("/devices/new", data={**form, "name": "c", "group_id": "__new__", "new_group": ""}, headers=htmx) + assert "Введите название новой группы" in r.text and 'id="modal-title"' in r.text + + # окна групп открываются и показывают ошибку (дубликат) внутри окна + assert "Новая группа" in c.get("/ui/dialog/group").text + r = c.post("/groups/new", data={"name": "Склад"}, headers=htmx) + assert "уже существует" in r.text and 'id="modal-title"' in r.text + + by_name = {d.name: d for d in devices.list_devices()} + store = next(g for g in groups.list_groups() if g["name"] == "Склад") + assert by_name["a"].group_id == office.id and by_name["b"].group_id == store["id"] and "c" not in by_name + assert by_name["a"].note == "Серверная, 2 этаж" and by_name["b"].note is None # пробелы обрезаются, пусто -> None + + +def test_device_name_is_immutable(): + """Имя задаётся только при создании: API отклоняет смену, форма изменения имя игнорирует.""" + with TestClient(create_app()) as c: + c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) + h = {"Authorization": f"Bearer {API_TOKEN}"} + d = c.post("/api/v1/devices", headers=h, json={"name": "r1", "host": "10.0.0.1", "username": "u", "password": "p"}).json() + r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r2"}) + assert r.status_code == 400 and "нельзя изменить" in r.text + # то же имя — допустимо (клиенты часто присылают объект целиком); остальные поля меняются + r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r1", "host": "10.0.0.9"}) + assert r.status_code == 200 and r.json()["name"] == "r1" and r.json()["host"] == "10.0.0.9" + # форма изменения: присланное имя игнорируется, остальное сохраняется + r = c.post(f"/devices/{d['id']}/edit", data={"name": "hacked", "host": "10.0.0.7", "port": "80", "username": "u"}, + follow_redirects=False) + assert r.status_code == 303 + page = c.get(f"/ui/dialog/device/{d['id']}").text + assert "readonly" in page and 'name="name"' not in page + dev = devices.get_device(d["id"]) + assert (dev.name, dev.host) == ("r1", "10.0.0.7") + + +@pytest.mark.asyncio +async def test_poll_marks_offline_and_recovers(monkeypatch): + """Фоновый опрос: недоступность видна сразу; после возвращения — полный опрос (версии обновились).""" + d = devices.create_device("r1", "10.0.0.1", 80, "u", "p") + up = {"on": False} + data = { + "/rest/system/resource": {"version": "7.24.4 (stable)", "uptime": "1m", "board-name": "hAP"}, + "/rest/system/routerboard": {"routerboard": "false"}, + "/rest/system/package/update": {"channel": "stable", "installed-version": "7.24.4", "latest-version": "7.24.4"}, + } + + def handler(r: httpx.Request) -> httpx.Response: + if not up["on"]: + raise httpx.ConnectError("connection refused") + 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) + assert devices.get_device(d.id).online is False + + up["on"] = True + await ops.poll_device(d.id) # был offline -> полный опрос + dev = devices.get_device(d.id) + assert dev.online is True and dev.status["ros_installed"] == "7.24.4" + + data["/rest/system/resource"]["uptime"] = "2m" + await ops.poll_device(d.id) # онлайн -> лёгкий опрос обновляет uptime, остальное сохраняется + dev = devices.get_device(d.id) + assert dev.status["uptime"] == "2m" and dev.status["ros_latest"] == "7.24.4" + + +def test_ros_state_values(): + """Состояния колонки «Upgrade ROS»: unknown / check_error / update / downgrade / current.""" + assert devices.ros_state({}) == "unknown" + assert devices.ros_state({"ros_installed": "7.24.4"}) == "unknown" # нет latest — проверка ещё не выполнялась + assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.16", "ros_check_error": "timeout"}) == "check_error" + assert devices.ros_state({"ros_installed": "7.15", "ros_latest": "7.16"}) == "update" + assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.23.7"}) == "downgrade" # long-term старше установленной + assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.24.4"}) == "current" diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..5275f71 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,180 @@ +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 diff --git a/tests/test_ids_migrations.py b/tests/test_ids_migrations.py new file mode 100644 index 0000000..c77940c --- /dev/null +++ b/tests/test_ids_migrations.py @@ -0,0 +1,110 @@ +import sqlite3 + +import pytest + +from app import db, ids +from app.services import devices + + +def test_ids_are_unique_typed_and_ordered(): + """Все ID глобально уникальны: не повторяются, типы не пересекаются, внутри типа растут со временем.""" + seen, last = set(), {} + for i in range(50_000): + prefix = list(ids.PREFIXES)[i % len(ids.PREFIXES)] + v = ids.new_id(prefix) + assert v not in seen and v > last.get(prefix, "") + assert ids.is_id(v, prefix) and not any(ids.is_id(v, o) for o in ids.PREFIXES if o != prefix) + seen.add(v) + last[prefix] = v + with pytest.raises(LookupError): # ID другого типа в пути/запросе отклоняется + ids.check(ids.new_id("job"), "dev") + with pytest.raises(LookupError): + devices.get_device("1") # числовые ID больше не существуют + + +LEGACY_DDL = """ +CREATE TABLE devices (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, host VARCHAR(255) NOT NULL, port INTEGER NOT NULL, + username VARCHAR(64) NOT NULL, password_enc TEXT NOT NULL, verify_tls BOOLEAN NOT NULL, created_at DATETIME NOT NULL, + online BOOLEAN, status_json TEXT NOT NULL, status_at DATETIME, last_error TEXT, last_backup_requested_at DATETIME, + use_tls BOOLEAN NOT NULL DEFAULT 1, group_id INTEGER, note TEXT, PRIMARY KEY (id), UNIQUE (name)); +CREATE TABLE jobs (id INTEGER NOT NULL, device_id INTEGER, device_name VARCHAR(64) NOT NULL, type VARCHAR(32) NOT NULL, + status VARCHAR(16) NOT NULL, message TEXT NOT NULL, created_at DATETIME NOT NULL, finished_at DATETIME, PRIMARY KEY (id)); +CREATE TABLE backups (id INTEGER NOT NULL, device_id INTEGER NOT NULL, requested_at DATETIME NOT NULL, status VARCHAR(16) NOT NULL, + key_binary VARCHAR(512), key_rsc VARCHAR(512), error TEXT, PRIMARY KEY (id), + FOREIGN KEY(device_id) REFERENCES devices (id) ON DELETE CASCADE); +CREATE TABLE device_groups (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, PRIMARY KEY (id), UNIQUE (name)); +""" + + +def test_migration_replaces_numeric_ids(tmp_path): + """БД старой схемы: числовые ID заменяются на dev_/grp_/job_/bkp_, ссылки и порядок сохраняются, повтор — без эффекта.""" + path = tmp_path / "legacy.db" + con = sqlite3.connect(path) + con.executescript(LEGACY_DDL) + con.executemany("INSERT INTO device_groups VALUES (?, ?)", [(1, "Alpha"), (2, "Beta")]) + for i, (name, grp) in enumerate((("r1", 1), ("r2", None)), start=1): + con.execute("INSERT INTO devices (id, name, host, port, username, password_enc, verify_tls, created_at, status_json, use_tls, group_id)" + " VALUES (?,?,?,?,?,?,?,?,?,?,?)", (i, name, "10.0.0.%d" % i, 80, "u", "enc", 0, "2026-09-1%d 10:00:00.000000" % i, "{}", 0, grp)) + # задачи: две по r1 (id 1), одна по уже удалённому устройству (id 99) + for i, dev in enumerate((1, 1, 99), start=1): + con.execute("INSERT INTO jobs VALUES (?,?,?,?,?,?,?,?)", (i, dev, "n", "backup", "done", "ok", "2026-09-19 10:0%d:00.000000" % i, None)) + con.execute("INSERT INTO backups VALUES (1, 1, '2026-09-19 10:00:00.000000', 'done', 'backups/r1/a.backup', 'backups/r1/a.rsc', NULL)") + con.execute("INSERT INTO backups VALUES (2, 99, '2026-09-19 10:05:00.000000', 'done', 'backups/gone/b.backup', NULL, NULL)") + con.commit() + con.close() + + db.init_db(f"sqlite:///{path}") + con = sqlite3.connect(path) + assert con.execute("PRAGMA user_version").fetchone()[0] == 2 + counts = {t: con.execute(f"SELECT count(*) FROM {t}").fetchone()[0] for t in ("device_groups", "devices", "jobs", "backups")} + assert counts == {"device_groups": 2, "devices": 2, "jobs": 3, "backups": 2} + alpha = con.execute("SELECT id FROM device_groups WHERE name='Alpha'").fetchone()[0] + r1, r2 = (con.execute("SELECT id, group_id FROM devices WHERE name=?", (n,)).fetchone() for n in ("r1", "r2")) + assert ids.is_id(r1[0], "dev") and r1[1] == alpha and r2[1] is None # связь устройство → группа сохранена + job_dev = [r[0] for r in con.execute("SELECT device_id FROM jobs ORDER BY id")] + assert job_dev[0] == job_dev[1] == r1[0] and job_dev[2] not in (r1[0], r2[0]) # удалённое устройство — свой стабильный ID + assert [r[0] for r in con.execute("SELECT id FROM jobs ORDER BY created_at")] == [r[0] for r in con.execute("SELECT id FROM jobs ORDER BY id")] + bk = con.execute("SELECT device_id, device_name FROM backups ORDER BY requested_at").fetchall() + assert bk[0] == (r1[0], "r1") and bk[1][0] == job_dev[2] and bk[1][1] == "gone" # имя — из ключа S3 + every = [r[0] for t in ("device_groups", "devices", "jobs", "backups", "events") for r in con.execute(f"SELECT id FROM {t}")] + assert len(every) == len(set(every)) and all(ids.is_id(i) for i in every) # ни одного пересечения между таблицами + assert con.execute("SELECT type FROM events").fetchall() == [("system.migrated",)] + assert len(list(tmp_path.glob("legacy.db.bak-*"))) == 1 # копия БД перед миграцией + con.close() + db.init_db(f"sqlite:///{path}") # повторный запуск ничего не меняет + con = sqlite3.connect(path) + assert con.execute("SELECT count(*) FROM events").fetchone()[0] == 1 and len(list(tmp_path.glob("legacy.db.bak-*"))) == 1 + con.close() + + +LEGACY_DDL_NO_EXTRA_COLUMNS = """ +CREATE TABLE devices (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, host VARCHAR(255) NOT NULL, port INTEGER NOT NULL, + username VARCHAR(64) NOT NULL, password_enc TEXT NOT NULL, verify_tls BOOLEAN NOT NULL, created_at DATETIME NOT NULL, + online BOOLEAN, status_json TEXT NOT NULL, status_at DATETIME, last_error TEXT, last_backup_requested_at DATETIME, + PRIMARY KEY (id), UNIQUE (name)); +CREATE TABLE jobs (id INTEGER NOT NULL, device_id INTEGER, device_name VARCHAR(64) NOT NULL, type VARCHAR(32) NOT NULL, + status VARCHAR(16) NOT NULL, message TEXT NOT NULL, created_at DATETIME NOT NULL, finished_at DATETIME, PRIMARY KEY (id)); +CREATE TABLE backups (id INTEGER NOT NULL, device_id INTEGER NOT NULL, requested_at DATETIME NOT NULL, status VARCHAR(16) NOT NULL, + key_binary VARCHAR(512), key_rsc VARCHAR(512), error TEXT, PRIMARY KEY (id), + FOREIGN KEY(device_id) REFERENCES devices (id) ON DELETE CASCADE); +CREATE TABLE device_groups (id INTEGER NOT NULL, name VARCHAR(64) NOT NULL, PRIMARY KEY (id), UNIQUE (name)); +""" + + +def test_migration_adds_legacy_columns_before_id_migration(tmp_path): + """Схема до появления групп/TLS-настроек/примечаний (без use_tls/group_id/note): колонки добавляются + раньше миграции ID — _to_v1 читает их из старой таблицы.""" + path = tmp_path / "legacy2.db" + con = sqlite3.connect(path) + con.executescript(LEGACY_DDL_NO_EXTRA_COLUMNS) + con.execute("INSERT INTO devices (id, name, host, port, username, password_enc, verify_tls, created_at, status_json)" + " VALUES (1, 'r1', '10.0.0.1', 80, 'u', 'enc', 0, '2026-09-10 10:00:00.000000', '{}')") + con.commit() + con.close() + + db.init_db(f"sqlite:///{path}") + con = sqlite3.connect(path) + assert con.execute("PRAGMA user_version").fetchone()[0] == 2 + dev_id, use_tls, group_id, note = con.execute("SELECT id, use_tls, group_id, note FROM devices").fetchone() + assert ids.is_id(dev_id, "dev") and (use_tls, group_id, note) == (1, None, None) # значения по умолчанию + con.close() diff --git a/tests/test_operations.py b/tests/test_operations.py new file mode 100644 index 0000000..d9b48eb --- /dev/null +++ b/tests/test_operations.py @@ -0,0 +1,185 @@ +import asyncio +import json + +import httpx +import pytest +from fastapi.testclient import TestClient + +from app.db import session_scope +from app.main import create_app +from app.models import Device +from app.ros import operations as ros +from app.services import devices, events, jobs, ops +from tests.conftest import ADMIN_PASSWORD, API_TOKEN +from tests.helpers import ros_client + + +@pytest.mark.asyncio +async def test_firmware_reboots_on_log_message(): + """Перезагрузка — по записи в журнале (current-firmware до неё не меняется).""" + def run(log_at_poll, old_entry_time): + calls, polls = [], {"n": 0} + + def handler(r: httpx.Request) -> httpx.Response: + calls.append(f"{r.method} {r.url.path}") + data = { + "/rest/system/routerboard": {"routerboard": "true", "current-firmware": "7.22", "upgrade-firmware": "7.24"}, + "/rest/system/clock": {"date": "2026-09-19", "time": "12:00:00"}, + "/rest/system/resource": {"uptime": "1h"}, # загрузка была в 11:00 + } + if r.url.path == "/rest/log": + polls["n"] += 1 + log = [{".id": "*1", "time": old_entry_time, "message": ros.FW_DONE_MSG}] + if polls["n"] >= log_at_poll: + log.append({".id": "*9", "time": "2026-09-19 12:00:00", "message": ros.FW_DONE_MSG}) + return httpx.Response(200, json=log) + return httpx.Response(200, json=data.get(r.url.path, [])) + + async def go(): + async with ros_client(handler) as c: + return await ros.upgrade_firmware(c, wait_seconds=10, poll=0) + return calls, go() + + # 1) новая запись появилась на 3-м чтении журнала -> команда upgrade, затем сразу reboot + calls, coro = run(3, "2026-09-19 08:00:00") + await coro + assert calls.index("POST /rest/system/routerboard/upgrade") < calls.index("POST /rest/system/reboot") == len(calls) - 1 + + # 2) запись новее последней загрузки уже есть (прошивка записана, перезагрузки не было) -> сразу reboot + calls, coro = run(99, "2026-09-19 11:30:00") + await coro + assert "POST /rest/system/routerboard/upgrade" not in calls and calls[-1] == "POST /rest/system/reboot" + + +@pytest.mark.asyncio +async def test_batch_channel_runs_as_jobs(monkeypatch): + """Групповая смена канала — фоновыми задачами: 202 + job_ids, задачи завершаются done, канал — в data job.created.""" + calls = [] + + async def fake_set_channel(device_id, channel): + calls.append((device_id, channel)) + + monkeypatch.setattr(ops, "set_channel", fake_set_channel) + d1 = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") + d2 = devices.create_device("r2", "10.0.0.2", 443, "admin", "pw") + + with TestClient(create_app()) as c: + h = {"Authorization": f"Bearer {API_TOKEN}"} + r = c.put("/api/v1/batch/channel", headers=h, json={"device_ids": [d1.id, d2.id], "channel": "testing"}) + assert r.status_code == 202 + job_ids = r.json()["job_ids"] + assert len(job_ids) == 2 + await asyncio.sleep(0.3) + + assert sorted(calls) == sorted([(d1.id, "testing"), (d2.id, "testing")]) + for jid in job_ids: + assert jobs.get_job(jid).status == "done" + created = [e for e in events.list_events(type_="job.created") if e.entity_id in job_ids] + assert len(created) == 2 and all(json.loads(e.data) == {"type": "set_channel", "channel": "testing"} for e in created) + # неверный канал отклоняется до создания задач + assert c.put("/api/v1/batch/channel", headers=h, json={"device_ids": [d1.id], "channel": "bogus"}).status_code == 422 + + +@pytest.mark.asyncio +async def test_run_ros_downgrade_order_and_failures(monkeypatch): + """run_ros_downgrade: версия не совпала с подтверждённой — бэкап и install не вызываются; версия совпала — + бэкап выполняется до install (порядок); бэкап упал — install не вызывается.""" + calls = [] + + def handler(r: httpx.Request) -> httpx.Response: + if r.url.path == "/rest/system/package/update/install": + calls.append("install") + return httpx.Response(200, json=[]) + data = {"/rest/system/package/update": {"installed-version": "7.24.4", "latest-version": "7.23.7"}} + return httpx.Response(200, json=data.get(r.url.path, [])) + + d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") + monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler)) + + async def fake_backup_ok(device_id): + calls.append("backup") + return "Бэкап загружен в S3: k.backup, k.rsc" + + async def fake_backup_fail(device_id): + calls.append("backup") + raise RuntimeError("бэкап не удался") + + # 1) версия канала (7.23.7) не совпала с подтверждённой пользователем -> задача failed, ничего не вызывается + monkeypatch.setattr(ops, "run_backup", fake_backup_ok) + [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.6"}) + await asyncio.sleep(0.2) + j = jobs.get_job(jid) + assert j.status == "failed" and "не совпадает" in j.message and calls == [] + + # 2) версия совпала -> бэкап выполняется до install, install вызван + [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.7"}) + await asyncio.sleep(0.2) + j = jobs.get_job(jid) + assert j.status == "done" and calls == ["backup", "install"] + + # 3) бэкап падает -> install не вызывается + calls.clear() + monkeypatch.setattr(ops, "run_backup", fake_backup_fail) + [jid] = await jobs.start_jobs("ros_downgrade", [d.id], {"target_version": "7.23.7"}) + await asyncio.sleep(0.2) + j = jobs.get_job(jid) + assert j.status == "failed" and calls == ["backup"] + + +@pytest.mark.asyncio +async def test_downgrade_api_endpoints(monkeypatch): + """target_version обязателен для одного устройства и группы; /batch/ros_downgrade — отдельный типизированный + эндпоинт (не через общий /batch/{action}, у которого в Literal ros_downgrade нет).""" + async def fake_downgrade(device_id, target_version): + return f"откат до {target_version}" + + monkeypatch.setitem(jobs.JOB_TYPES, "ros_downgrade", fake_downgrade) + d1 = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") + d2 = devices.create_device("r2", "10.0.0.2", 443, "admin", "pw") + + with TestClient(create_app()) as c: + h = {"Authorization": f"Bearer {API_TOKEN}"} + assert c.post(f"/api/v1/devices/{d1.id}/update/downgrade", headers=h, json={}).status_code == 422 + r = c.post(f"/api/v1/devices/{d1.id}/update/downgrade", headers=h, json={"target_version": "7.23.7"}) + assert r.status_code == 202 + job_ids = r.json()["job_ids"] + + # тело без target_version — если бы запрос ушёл в общий /batch/{action} (там target_version не нужен), + # он был бы принят; 422 подтверждает, что обрабатывает именно типизированный /batch/ros_downgrade + assert c.post("/api/v1/batch/ros_downgrade", headers=h, json={"device_ids": [d1.id]}).status_code == 422 + r = c.post("/api/v1/batch/ros_downgrade", headers=h, + json={"device_ids": [d1.id, d2.id], "target_version": "7.23.7"}) + assert r.status_code == 202 + job_ids += r.json()["job_ids"] + assert len(job_ids) == 3 + + await asyncio.sleep(0.3) + for jid in job_ids: + j = jobs.get_job(jid) + assert j.type == "ros_downgrade" and j.status == "done" and j.message == "откат до 7.23.7" + + +def test_ui_downgrade_dialog_and_mismatch(monkeypatch): + """Окно отката: неактивная кнопка, пока версия не введена (data-enables); несовпадающая версия — ошибка + в окне, задача не создаётся.""" + d = devices.create_device("r1", "10.0.0.1", 443, "admin", "pw") + + def _set_status(device, status): + with session_scope() as s: + s.get(Device, device.id).status_json = json.dumps(status) + + _set_status(d, {"channel": "long-term", "ros_installed": "7.24.4", "ros_latest": "7.23.7"}) + + with TestClient(create_app()) as c: + c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) + htmx = {"HX-Request": "true"} + page = c.get(f"/ui/dialog/downgrade?device_ids={d.id}", headers=htmx).text + assert "r1" in page and 'data-enables="#downgrade-confirm"' in page and 'id="downgrade-confirm" disabled' in page + + r = c.post("/ui/downgrade", data={"device_ids": [d.id], "target_version": "7.16"}, headers=htmx) + assert "не совпадает" in r.text and 'id="modal-title"' in r.text + assert jobs.list_jobs() == [] + + r = c.post("/ui/downgrade", data={"device_ids": [d.id], "target_version": "7.23.7"}, headers=htmx) + assert r.headers.get("HX-Refresh") == "true" + assert len(jobs.list_jobs()) == 1 and jobs.list_jobs()[0].type == "ros_downgrade" diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..fd6412e --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,150 @@ +import json + +import pytest +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient + +from app import process_lock, security +from app.config import Settings, get_settings, insecure_settings +from app.main import create_app +from app.services import events +from tests.conftest import ADMIN_PASSWORD, API_TOKEN +from tests.helpers import _request + + +def test_password_encryption_roundtrip(): + token = security.encrypt("s3cret") + assert token != "s3cret" and security.decrypt(token) == "s3cret" + + +def test_api_auth_and_no_password_leak(): + with TestClient(create_app()) as client: + assert client.get("/api/v1/devices").status_code == 401 + h = {"Authorization": f"Bearer {API_TOKEN}"} + r = client.post("/api/v1/devices", headers=h, json={ + "name": "r1", "host": "10.0.0.1", "username": "admin", "password": "pw"}) + assert r.status_code == 201 and "password" not in r.text + assert client.get("/api/v1/devices", headers=h).json()[0]["name"] == "r1" + + +def test_insecure_settings_rejects_weak_secrets_without_leaking_them(): + """Небезопасные секреты выявляются по каждому правилу; значения секретов в описание проблем не попадают.""" + secret = "s3cr3t-value-must-not-leak-anywhere" + bad = Settings(api_token="change-me", session_secret=secret[:10], admin_password=secret[:8], secret_key=secret) + problems = insecure_settings(bad) + assert len(problems) == 4 # все четыре секрета нарушают правила + text = " ".join(problems) + assert secret not in text and secret[:10] not in text and secret[:8] not in text + + ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, + secret_key=Fernet.generate_key().decode()) + assert insecure_settings(ok) == [] + + +def test_insecure_settings_rejects_invalid_trusted_proxies(): + """TRUSTED_PROXIES с записью не в формате CIDR — тоже отказ старта; значение (не секрет) видно в тексте проблемы.""" + ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, + secret_key=Fernet.generate_key().decode(), trusted_proxies="10.0.0.0/8, not-a-network") + problems = insecure_settings(ok) + assert len(problems) == 1 and "TRUSTED_PROXIES" in problems[0] and "not-a-network" in problems[0] + + +def test_client_ip_trusted_proxies(monkeypatch): + """X-Forwarded-For учитывается, только если TRUSTED_PROXIES не пуст и peer — из доверенной сети (иначе подделка + заголовка бессмысленна); цепочка разбирается справа налево до первого недоверенного адреса; невалидная запись + обрывает разбор на последнем валидном; работает и с IPv6.""" + monkeypatch.setenv("TRUSTED_PROXIES", "") + get_settings.cache_clear() + assert security.client_ip(_request("1.2.3.4", "9.9.9.9")) == "1.2.3.4" # пусто — XFF полностью игнорируется + + monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") + get_settings.cache_clear() + assert security.client_ip(_request("9.9.9.9", "9.9.9.9")) == "9.9.9.9" # peer не доверенный — XFF игнорируется + assert security.client_ip(_request("10.0.0.1", "1.2.3.4, 10.0.0.2")) == "1.2.3.4" # доверенный прокси пропущен + assert security.client_ip(_request("10.0.0.1", "10.0.0.3, 10.0.0.2")) == "10.0.0.1" # все доверенные — peer + assert security.client_ip(_request("10.0.0.1", "garbage, 10.0.0.2")) == "10.0.0.2" # мусор — последний валидный + assert security.client_ip(_request("10.0.0.1", "garbage")) == "10.0.0.1" # мусор без валидных перед ним — peer + assert security.client_ip(_request("10.0.0.1")) == "10.0.0.1" # заголовка нет + assert security.client_ip(_request("10.0.0.1", "2001:db8::1, 10.0.0.2")) == "2001:db8::1" # IPv6 + get_settings.cache_clear() + + +def test_app_refuses_to_start_with_insecure_config(monkeypatch): + """Приложение не стартует с небезопасной конфигурацией (например, API_TOKEN=change-me).""" + monkeypatch.setenv("API_TOKEN", "change-me") + get_settings.cache_clear() + with pytest.raises(RuntimeError, match="Небезопасная конфигурация"): + with TestClient(create_app()): + pass + get_settings.cache_clear() + + +def test_login_lockout_by_ip(): + """5 неверных попыток входа с одного IP блокируют его на 6-ю (пароль уже не проверяется, код 429); + повторные попытки во время блокировки не засоряют журнал новыми auth.locked; блокировка не распространяется + на другой IP; успешный вход сбрасывает счётчик неудач.""" + with TestClient(create_app()) as c: + for _ in range(5): + r = c.post("/login", data={"username": "admin", "password": "wrong"}) + assert r.status_code == 401 + r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) # верный пароль уже не спасает + assert r.status_code == 429 and "Слишком много попыток" in r.text + assert len(events.list_events(type_="auth.locked")) == 1 + for _ in range(3): # попытки во время блокировки — 429, но новых auth.locked не пишут (не засоряют журнал) + assert c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}).status_code == 429 + assert len(events.list_events(type_="auth.locked")) == 1 + + with TestClient(create_app(), client=("10.0.0.2", 1)) as c2: + r = c2.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) + assert r.status_code == 303 # другой IP свободен + + with TestClient(create_app(), client=("10.0.0.3", 1)) as c3: + for _ in range(3): + assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 + r = c3.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) + assert r.status_code == 303 + for _ in range(4): # если бы счётчик не сбросился успехом, пятая по счёту неудача (2-я в этом цикле) заблокировала бы + assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 + + +def test_login_lockout_by_ip_behind_trusted_proxy(monkeypatch): + """За доверенным прокси блокировка и событие auth.* считаются по адресу из X-Forwarded-For, а не по адресу + прокси: 5 неверных попыток с одним XFF блокируют этот адрес, попытка с другим XFF (тот же прокси-peer) проходит.""" + monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") + get_settings.cache_clear() + with TestClient(create_app(), client=("10.0.0.5", 1)) as c: + for _ in range(5): + r = c.post("/login", data={"username": "admin", "password": "wrong"}, headers={"X-Forwarded-For": "1.1.1.1"}) + assert r.status_code == 401 + r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "1.1.1.1"}) + assert r.status_code == 429 # заблокирован именно реальный адрес клиента + assert json.loads(events.list_events(type_="auth.locked")[0].data)["ip"] == "1.1.1.1" + + r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "2.2.2.2"}, + follow_redirects=False) + assert r.status_code == 303 # другой адрес за тем же прокси не заблокирован + get_settings.cache_clear() + + +def test_move_redirect_rejects_open_redirect_next(): + """`next` в /ui/move принимает только локальный путь — иначе редирект на «/» (открытый редирект).""" + with TestClient(create_app()) as c: + c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) + for bad in ("/\\evil.com", "//evil.com", "https://evil.com", "/\r\nX"): # последний — декодированный /%0d%0aX + r = c.post("/ui/move", data={"next": bad}, follow_redirects=False) + assert r.headers["location"] == "/" + r = c.post("/ui/move", data={"next": "/?f_group=none"}, follow_redirects=False) + assert r.headers["location"] == "/?f_group=none" # обычный путь с фильтром сохраняется + + +def test_process_lock_blocks_second_process(tmp_path): + """Второй процесс на той же файловой БД не стартует; после освобождения блокировки — снова можно.""" + db_url = f"sqlite:///{tmp_path}/lock.db" + process_lock.acquire(db_url) + try: + with pytest.raises(RuntimeError, match="уже используется"): + process_lock.acquire(db_url) + finally: + process_lock.release() + process_lock.acquire(db_url) # после освобождения — успешно + process_lock.release()