117 lines
8.3 KiB
Python
117 lines
8.3 KiB
Python
"""Демо-режим персональный: в демо работает только тот, кто его включил (или кого включил администратор)."""
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "dm.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 import config, db as dbm, demo, mode # noqa: E402
|
|
from app.main import app # noqa: E402
|
|
|
|
|
|
def test_demo_mode_personal():
|
|
real_path = os.environ["DATABASE_URL"].replace("sqlite:///", "")
|
|
data_dir = Path(config.DATA_DIR)
|
|
with TestClient(app) as c:
|
|
H = lambda t: {"Authorization": "Bearer " + t}
|
|
tk = c.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json()
|
|
adm = H(tk["access_token"])
|
|
rt_user = c.post("/api/v1/users", headers=adm, json={"username": "realtech_dm", "password": "Pass1234567"}).json()
|
|
c.post("/api/v1/equipment", headers=adm, json={"name": "БОЕВОЙ-СТАНОК"})
|
|
rt = H(c.post("/api/v1/auth/login", json={"username": "realtech_dm", "password": "Pass1234567"}).json()["access_token"])
|
|
|
|
def real_snapshot():
|
|
con = sqlite3.connect(real_path)
|
|
try:
|
|
return {t: con.execute(f"select count(*) from {t}").fetchone()[0] for t in ("users", "equipment", "orders", "checklists", "positions")}
|
|
finally:
|
|
con.close()
|
|
real_names = sorted(e["name"] for e in c.get("/api/v1/equipment", headers=adm).json())
|
|
before = real_snapshot()
|
|
|
|
assert c.post("/api/v1/demo/enable", headers=rt, json={"confirm": True}).status_code == 403
|
|
assert c.post("/api/v1/demo/enable", headers=adm, json={"confirm": False}).status_code == 422
|
|
assert c.get("/api/v1/demo/status").json() == {"active": False}
|
|
|
|
new = c.post("/api/v1/demo/enable", headers=adm, json={"confirm": True})
|
|
assert new.status_code == 200, new.text
|
|
dadm = H(new.json()["access_token"])
|
|
assert c.get("/api/v1/demo/status").json() == {"active": True}
|
|
me = c.get("/api/v1/me", headers=dadm).json()
|
|
assert me["demo_mode"] and me["is_admin"] and me["username"] == "admin"
|
|
# прежний боевой токен админа больше не годится — обновление выдаёт демо-сессию
|
|
assert c.get("/api/v1/me", headers=adm).status_code == 401
|
|
ref = c.post("/api/v1/auth/refresh", json={"refresh_token": tk["refresh_token"]}).json()
|
|
assert c.get("/api/v1/me", headers=H(ref["access_token"])).json()["demo_mode"] is True
|
|
# ГЛАВНОЕ: остальные работают в боевой БД, как будто демо нет
|
|
assert c.get("/api/v1/me", headers=rt).json()["demo_mode"] is False
|
|
assert "БОЕВОЙ-СТАНОК" in [e["name"] for e in c.get("/api/v1/equipment", headers=rt).json()]
|
|
assert c.post("/api/v1/auth/login", json={"username": "realtech_dm", "password": "Pass1234567"}).status_code == 200
|
|
names = [e["name"] for e in c.get("/api/v1/equipment", headers=dadm).json()]
|
|
assert len(names) >= 8 and "БОЕВОЙ-СТАНОК" not in names
|
|
info = c.get("/api/v1/demo", headers=dadm).json()
|
|
assert info["active"] and info["in_demo"] and any(a["role"] == "Модератор" for a in info["accounts"])
|
|
assert {u["username"]: u["in_demo"] for u in info["users"]}["realtech_dm"] is False
|
|
# демо-учётная запись входит по паролю demo1234, обновляет сессию
|
|
tech = [a for a in info["accounts"] if a["role"] == "Техник" and a["on_shift"]][0]
|
|
dl = c.post("/api/v1/auth/login", json={"username": tech["username"], "password": info["password"]})
|
|
assert dl.status_code == 200 and c.get("/api/v1/me", headers=H(dl.json()["access_token"])).json()["demo_mode"]
|
|
dref = c.post("/api/v1/auth/refresh", json={"refresh_token": dl.json()["refresh_token"]})
|
|
assert dref.status_code == 200 and c.get("/api/v1/me", headers=H(dref.json()["access_token"])).json()["username"] == tech["username"]
|
|
imp = c.post("/api/v1/demo/impersonate", headers=dadm, json={"user_id": tech["id"]}).json()
|
|
assert c.get("/api/v1/me", headers=H(imp["access_token"])).json()["username"] == tech["username"]
|
|
assert c.post("/api/v1/demo/destroy", headers=H(imp["access_token"])).status_code == 403
|
|
now = datetime.now()
|
|
rep = c.get("/api/v1/reports/completed-orders", headers=dadm, params={"from": (now - timedelta(days=15)).isoformat(), "to": (now + timedelta(days=1)).isoformat()}).json()
|
|
assert rep["total"] > 20
|
|
c.post("/api/v1/equipment", headers=dadm, json={"name": "демо-правка"})
|
|
assert c.post("/api/v1/backups", headers=dadm, json={}).status_code == 409 # копия — только боевой БД
|
|
assert real_snapshot() == before # боевая БД не тронута демо-правками
|
|
|
|
# администратор включает демо другому пользователю и выключает
|
|
assert c.post(f"/api/v1/demo/users/{rt_user['id']}", headers=rt, json={"on": True}).status_code == 403
|
|
assert c.post(f"/api/v1/demo/users/{rt_user['id']}", headers=dadm, json={"on": True}).json()["in_demo"] is True
|
|
assert c.get("/api/v1/me", headers=rt).status_code == 401
|
|
rl = c.post("/api/v1/auth/login", json={"username": "realtech_dm", "password": "Pass1234567"}).json()
|
|
rtd = H(rl["access_token"])
|
|
assert c.get("/api/v1/me", headers=rtd).json()["demo_mode"] is True
|
|
assert "БОЕВОЙ-СТАНОК" not in [e["name"] for e in c.get("/api/v1/equipment", headers=rtd).json()]
|
|
assert c.post(f"/api/v1/demo/users/{rt_user['id']}", headers=dadm, json={"on": False}).json()["in_demo"] is False
|
|
assert c.get("/api/v1/me", headers=rtd).status_code == 401
|
|
back = c.post("/api/v1/auth/refresh", json={"refresh_token": rl["refresh_token"]}).json()
|
|
assert c.get("/api/v1/me", headers=H(back["access_token"])).json()["demo_mode"] is False
|
|
|
|
# перезапуск сервера во время демо: демо-БД подключается снова
|
|
dbm.set_demo_engine(None).dispose() # как при перезапуске: соединения закрыты
|
|
mode.State.active = False
|
|
demo.activate_on_startup()
|
|
assert mode.State.active and dbm.demo_engine is not None
|
|
assert c.get("/api/v1/me", headers=dadm).status_code == 200
|
|
|
|
# пересоздание: старые демо-токены недействительны, refresh выдаёт новые
|
|
assert c.post("/api/v1/demo/reset", headers=dadm).status_code == 200
|
|
assert c.get("/api/v1/me", headers=dadm).status_code == 401
|
|
ref = c.post("/api/v1/auth/refresh", json={"refresh_token": new.json()["refresh_token"]}).json()
|
|
dadm = H(ref["access_token"])
|
|
assert "демо-правка" not in [e["name"] for e in c.get("/api/v1/equipment", headers=dadm).json()]
|
|
assert len(list(data_dir.glob("demo_*.db*")) + list(data_dir.glob("demo_photos_*"))) == 2
|
|
|
|
# сам выходит из демо, затем удаляет демо-данные
|
|
off = c.post("/api/v1/demo/leave", headers=dadm)
|
|
assert off.status_code == 200, off.text
|
|
backh = H(off.json()["access_token"])
|
|
assert c.get("/api/v1/me", headers=backh).json()["demo_mode"] is False
|
|
assert sorted(e["name"] for e in c.get("/api/v1/equipment", headers=backh).json()) == real_names
|
|
assert c.post("/api/v1/demo/destroy", headers=backh).status_code == 200
|
|
assert c.get("/api/v1/demo/status").json() == {"active": False}
|
|
assert list(data_dir.glob("demo_*.db*")) + list(data_dir.glob("demo_photos_*")) == []
|
|
assert c.post("/api/v1/auth/login", json={"username": tech["username"], "password": "demo1234"}).status_code == 401
|
|
assert real_snapshot()["equipment"] == before["equipment"]
|