Портал технической службы: наряды и чек-листы, задачи, отказы и неисправности, анализ надёжности, карты, склады, приложение для Android

tests/test_z_backup.py

@@ -0,0 +1,72 @@
+"""Резервное копирование: создание, состав, восстановление (в т. ч. циклические ссылки), права, защита имён."""
+import io
+import json
+import os
+import tempfile
+import zipfile
+
+os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "bk.db").replace("\\", "/"))
+os.environ.setdefault("SCHEDULER_INTERVAL", "3600")
+os.environ.setdefault("SECRET_KEY", "test-secret-key-for-pytest-0123456789abcdef")
+os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
+
+from fastapi.testclient import TestClient  # noqa: E402
+
+from app.main import app  # noqa: E402
+
+V1 = "/api/v1"
+PNG = b"\x89PNG\r\n\x1a\n" + b"\0" * 64
+
+
+def test_backup_and_restore():
+    with TestClient(app) as c:
+        H = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]}
+        c.post(f"{V1}/users", headers=H, json={"username": "bk_tech", "password": "Pass1234567"})
+        T = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "bk_tech", "password": "Pass1234567"}).json()["access_token"]}
+        assert c.get(f"{V1}/backups", headers=T).status_code == 403
+        # данные с циклом «отдел ↔ руководитель» и вложенностью
+        boss = c.post(f"{V1}/users", headers=H, json={"username": "bk_boss", "password": "Pass1234567"}).json()
+        top = c.post(f"{V1}/departments", headers=H, json={"name": "БК-верх", "head_user_id": boss["id"]}).json()
+        sub = c.post(f"{V1}/departments", headers=H, json={"name": "БК-низ", "parent_id": top["id"]}).json()
+        c.patch(f"{V1}/users/{boss['id']}", headers=H, json={"department_id": sub["id"]})
+        eq = c.post(f"{V1}/equipment", headers=H, json={"name": "БК-насос"}).json()
+        c.put(f"{V1}/map/image", headers=H, content=PNG)
+        c.put(f"{V1}/settings", headers=H, json={"backup_keep": 5, "backup_auto_hours": 12})
+
+        r = c.post(f"{V1}/backups", headers=H, json={"include_media": True, "note": "тест"})
+        assert r.status_code == 201, r.text
+        name = r.json()["name"]
+        assert r.json()["media"] and r.json()["rows"] > 0
+        z = zipfile.ZipFile(io.BytesIO(c.get(f"{V1}/backups/{name}", headers=H).content))
+        names = z.namelist()
+        assert "manifest.json" in names and "db/users.json" in names and "config/environment.json" in names
+        assert any(n.startswith("media/") and "/map/" in n for n in names)
+        env = json.loads(z.read("config/environment.json"))
+        assert "SECRET_KEY" not in env["variables"] and "SECRET_KEY" in env["secrets_not_saved"]
+        assert env["server_settings"]["backup_keep"] == "5"
+        assert "db/refresh_tokens.json" not in names
+        lst = c.get(f"{V1}/backups", headers=H).json()
+        assert lst["auto_hours"] == 12 and any(x["name"] == name for x in lst["items"])
+
+        # меняем данные и восстанавливаем
+        c.delete(f"{V1}/equipment/{eq['id']}", headers=H)
+        c.post(f"{V1}/equipment", headers=H, json={"name": "БК-лишнее"})
+        assert c.post(f"{V1}/backups/{name}/restore", headers=H, json={}).status_code == 422
+        res = c.post(f"{V1}/backups/{name}/restore", headers=H, json={"confirm": True})
+        assert res.status_code == 200, res.text
+        assert res.json()["safety_copy"].endswith("before-restore.zip")
+        H = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]}
+        names_eq = [e["name"] for e in c.get(f"{V1}/equipment", headers=H).json()]
+        assert "БК-насос" in names_eq and "БК-лишнее" not in names_eq
+        deps = {d["name"]: d for d in c.get(f"{V1}/departments", headers=H).json()}
+        assert deps["БК-низ"]["parent_id"] == deps["БК-верх"]["id"] and deps["БК-верх"]["head_user_id"] == boss["id"]
+        assert c.get(f"{V1}/map/image", headers=H).content == PNG
+        new = c.post(f"{V1}/equipment", headers=H, json={"name": "БК-после"})  # счётчики id не сломаны
+        assert new.status_code == 201
+
+        # загрузка чужого файла и защита имён
+        assert c.post(f"{V1}/backups/upload", headers=H, content=b"not a zip").status_code == 422
+        up = c.post(f"{V1}/backups/upload", headers=H, content=c.get(f"{V1}/backups/{name}", headers=H).content)
+        assert up.status_code == 201 and up.json()["name"].endswith("_upload.zip")
+        assert c.get(f"{V1}/backups/..%2F..%2Fapp.db", headers=H).status_code == 404
+        assert c.delete(f"{V1}/backups/{up.json()['name']}", headers=H).status_code == 204