73 lines
4.7 KiB
Python
73 lines
4.7 KiB
Python
"""Резервное копирование: создание, состав, восстановление (в т. ч. циклические ссылки), права, защита имён."""
|
|
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
|