ros_control/tests/test_backups.py

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() во время предыдущего чтения