"""График у сотрудника (сменщики на одной должности), автоочистка старых фото с каталогом keep/, мониторинг сервера.""" import os import tempfile from datetime import timedelta os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "tw.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, logic, photo_cleanup, settings_store as ss # noqa: E402 from app.db import SessionLocal # noqa: E402 from app.main import app # noqa: E402 from app.models import Photo, Position, TaskPhoto, User # noqa: E402 V1 = "/api/v1" PW = "Pass1234567" PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 64 def test_schedule_moves_from_position_to_user(): with TestClient(app) as c: adm = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]} off = c.post(f"{V1}/schedules", headers=adm, json={"name": "TW-выходной", "kind": "weekly", "weekdays": []}).json() assert c.post(f"{V1}/users", headers=adm, json={"username": "tw_bad", "password": PW, "schedule_id": 999999}).status_code == 422 # старые данные: график задан должности — однократный перенос ставит его сотрудникам без своего графика with SessionLocal() as db: p = Position(name="TW-Дежурный", kind="technician", schedule_id=off["id"]) db.add(p) db.flush() db.add_all([User(username="tw_old1", password_hash="x", position_id=p.id), User(username="tw_old2", password_hash="x", position_id=p.id)]) ss.set_(db, "schedule_per_user", "0") db.commit() assert logic.schedules_to_users(db) >= 2 assert logic.schedules_to_users(db) == 0 # повторно не выполняется u1 = db.query(User).filter_by(username="tw_old1").one() assert u1.schedule_id == off["id"] and not logic.user_on_shift(u1) u1.schedule_id = None # свой график убрали — всегда на смене, график должности больше не влияет db.commit() db.refresh(u1) assert logic.user_on_shift(u1) r = c.get(f"{V1}/users", headers=adm).json() assert next(u for u in r if u["username"] == "tw_old2")["schedule_id"] == off["id"] def test_photo_cleanup_keeps_marked_photos(): with TestClient(app) as c: adm = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]} cl = c.post(f"{V1}/checklists", headers=adm, json={"name": "TW-фото", "items": [ {"text": "Общий вид", "answer_type": "status", "photo": "optional"}, {"text": "Табличка (хранить)", "answer_type": "status", "photo": "optional", "photo_keep": True}]}).json() assert [i["photo_keep"] for i in cl["items"]] == [False, True] pos = c.post(f"{V1}/positions", headers=adm, json={"name": "TW-слесарь", "kind": "technician"}).json() tech = c.post(f"{V1}/users", headers=adm, json={"username": "tw_tech", "password": PW, "position_id": pos["id"]}).json() th = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "tw_tech", "password": PW}).json()["access_token"]} o = c.post(f"{V1}/orders", headers=adm, json={"title": "TW-наряд", "checklist_id": cl["id"], "assignee_user_id": tech["id"]}).json() a, b = o["items"] assert b["photo_keep"] is True for it in (a, b): r = c.post(f"{V1}/orders/{o['id']}/items/{it['id']}/photos", headers={**th, "Content-Type": "image/png"}, content=PNG) assert r.status_code in (200, 201), r.text t = c.post(f"{V1}/tasks", headers=adm, json={"title": "TW-задание", "photo_keep": True}).json() assert t["photo_keep"] is True c.post(f"{V1}/tasks/{t['id']}/photos", headers={**adm, "Content-Type": "image/png"}, content=PNG) t2 = c.post(f"{V1}/tasks", headers=adm, json={"title": "TW-задание 2"}).json() c.post(f"{V1}/tasks/{t2['id']}/photos", headers={**adm, "Content-Type": "image/png"}, content=PNG) with SessionLocal() as db: root = ss.photo_root(db) ph = {p.item_id: p for p in db.query(Photo).filter(Photo.order_id == o["id"])} assert ph[b["id"]].path.startswith("keep/") and not ph[a["id"]].path.startswith("keep/") tp = {p.task_id: p for p in db.query(TaskPhoto).filter(TaskPhoto.task_id.in_([t["id"], t2["id"]]))} assert tp[t["id"]].path.startswith("keep/") and not tp[t2["id"]].path.startswith("keep/") files = [root / ph[a["id"]].path, root / ph[b["id"]].path, root / "media" / "tasks" / tp[t["id"]].path, root / "media" / "tasks" / tp[t2["id"]].path] assert all(f.is_file() for f in files) # всё «состарим» на два года old = config.now() - timedelta(days=730) for p in list(ph.values()) + list(tp.values()): p.created_at = old db.commit() # выключено по умолчанию — ничего не удаляется assert c.get(f"{V1}/photo-cleanup", headers=adm).json()["enabled"] is False assert c.put(f"{V1}/settings", headers=adm, json={"photo_keep_days": 365}).status_code == 200 prev = c.get(f"{V1}/photo-cleanup", headers=adm).json() assert prev["enabled"] and prev["photos"] >= 2 assert all(f.is_file() for f in files) # предпросмотр ничего не удаляет r = c.post(f"{V1}/photo-cleanup", headers=adm).json() assert r["photos"] >= 2 assert not files[0].exists() and files[1].is_file() and files[2].is_file() and not files[3].exists() with SessionLocal() as db: assert not photo_cleanup.due(db) # фото «хранить» по-прежнему открывается o2 = c.get(f"{V1}/orders/{o['id']}", headers=adm).json() assert [len(i["photos"]) for i in o2["items"]] == [0, 1] assert c.get(f"{V1}/photos/{o2['items'][1]['photos'][0]['id']}", headers=adm).status_code == 200 tid = c.get(f"{V1}/tasks/{t['id']}", headers=adm).json()["photos"][0] assert c.get(f"{V1}/task-photos/{tid}", headers=adm).status_code == 200 s = c.get(f"{V1}/settings", headers=adm).json() assert s["photo_keep_days"] == "365" and s["photo_keep_stats"]["files"] >= 2 c.put(f"{V1}/settings", headers=adm, json={"photo_keep_days": 0}) def test_server_monitor(): with TestClient(app) as c: adm = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]} c.post(f"{V1}/users", headers=adm, json={"username": "tw_adm", "password": PW, "is_admin": True}) adm2 = {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": "tw_adm", "password": PW}).json()["access_token"]} assert c.get(f"{V1}/server/monitor", headers=adm2).status_code == 403 # только администратор портала m = c.get(f"{V1}/server/monitor", headers=adm).json() assert m["cpu"] is None or 0 <= m["cpu"] <= 100 assert m["memory"] is None or 0 < m["memory"]["used"] <= m["memory"]["total"] assert m["disks"] and all(0 < d["used"] <= d["total"] for d in m["disks"])