ayurishchevandClaude Sonnet 5
1e5fb17755
Имя входит в ключи бэкапов в S3 (backups/<имя>/…), поэтому после создания не меняется (docs/changes/015): - API: PATCH с другим именем → 400 «Имя устройства нельзя изменить», то же имя допустимо; - UI: в окне изменения поле имени только для чтения, форма изменения присланное имя игнорирует; - вёрстка: минимальная ширина таблиц устройств и файлов — в узком окне колонка с именем не схлопывается, включается горизонтальная прокрутка. Тесты: 12 из 12. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
298 lines
16 KiB
Python
298 lines
16 KiB
Python
import json
|
|
from datetime import date, datetime, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import s3, security
|
|
from app.main import create_app
|
|
from app.models import Backup, Device
|
|
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, groups, ops
|
|
|
|
|
|
def ros_client(handler) -> RosClient:
|
|
return RosClient("r1", 443, "admin", "pw", transport=httpx.MockTransport(handler))
|
|
|
|
|
|
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 = {}
|
|
|
|
async def fake_upload(path, key): # содержимое читаем до удаления временной папки
|
|
uploaded[key] = open(path, "rb").read()
|
|
|
|
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 # .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": "Bearer test-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": "pw"})
|
|
# 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 'class="dev-link" href="/devices/1/edit" hx-get="/ui/dialog/device/1"' in 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)
|
|
|
|
monkeypatch.setattr(s3, "delete_object", fake_delete)
|
|
ok = ["backups/a/1.backup", "backups/a/1.rsc"]
|
|
with TestClient(create_app()) as c:
|
|
c.post("/login", data={"username": "admin", "password": "pw"})
|
|
# 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": "Bearer test-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
|
|
|
|
|
|
@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": "pw"})
|
|
h = {"Authorization": "Bearer test-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")
|