darkorion
e67c175f80
- docker-compose.yml: профили docs (Nextcloud), office (ONLYOFFICE Document Server), ai (Ollama + скачивание модели) - автонастройка: адреса Nextcloud, ONLYOFFICE, портала и модели подставляются из .env или по именам сервисов во внутренней сети Docker; значения из интерфейса важнее - локальная модель в своей сети (имя сервиса, частный IP) подключается без ключа API - .env.example, docs/DEPLOY.md (раздел 7а), README
116 lines
7.8 KiB
Python
116 lines
7.8 KiB
Python
"""График у сотрудника (сменщики на одной должности), автоочистка старых фото с каталогом 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"])
|