import os import pytest import tempfile from datetime import datetime, date, time # TEST_DATABASE_URL позволяет прогнать те же тесты на PostgreSQL (по умолчанию — временный файл SQLite) os.environ["DATABASE_URL"] = os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "t.db").replace("\\", "/") os.environ["SCHEDULER_INTERVAL"] = "3600" os.environ.setdefault("SECRET_KEY", "test-secret-key-for-pytest-0123456789abcdef") _DATA = tempfile.mkdtemp() os.environ["DATA_DIR"] = _DATA from fastapi.testclient import TestClient # noqa: E402 from app import logic # noqa: E402 from app.main import app # noqa: E402 from app.models import ShiftSchedule, OrderTemplate # noqa: E402 def test_shifts_and_recurrence(): cyc = ShiftSchedule(kind="cycle", work_days=2, off_days=2, cycle_start=date(2026, 1, 1), start_time=time(20, 0), end_time=time(8, 0)) assert logic.shift_active(cyc, datetime(2026, 1, 1, 22, 0)) assert logic.shift_active(cyc, datetime(2026, 1, 2, 3, 0)) # ночная смена с 1-го assert logic.shift_active(cyc, datetime(2026, 1, 2, 23, 0)) assert logic.shift_active(cyc, datetime(2026, 1, 3, 3, 0)) # смена со 2-го идёт до 8:00 3-го assert not logic.shift_active(cyc, datetime(2026, 1, 3, 22, 0)) # 3-е — выходной wk = ShiftSchedule(kind="weekly", weekdays=[0], start_time=time(8, 0), end_time=time(17, 0)) assert logic.shift_active(wk, datetime(2026, 1, 5, 9, 0)) # понедельник assert not logic.shift_active(wk, datetime(2026, 1, 6, 9, 0)) t = OrderTemplate(recurrence="monthly", month_day=31, run_time=time(8, 0), start_date=date(2026, 1, 1)) assert logic.next_run_after(t, datetime(2026, 2, 1)) == datetime(2026, 2, 28, 8, 0) t = OrderTemplate(recurrence="interval", interval_days=3, run_time=time(6, 0), start_date=date(2026, 1, 1)) assert logic.next_run_after(t, datetime(2026, 1, 2)) == datetime(2026, 1, 4, 6, 0) def test_full_flow(): with TestClient(app) as c: def login(u, p): r = c.post("/api/auth/login", json={"username": u, "password": p}) assert r.status_code == 200, r.text return {"Authorization": "Bearer " + r.json()["access_token"]} adm = login("admin", "admin") sch = c.post("/api/schedules", headers=adm, json={"name": "Всегда", "kind": "weekly", "weekdays": [0, 1, 2, 3, 4, 5, 6], "start_time": "00:00:00", "end_time": "23:59:00"}).json() night = c.post("/api/schedules", headers=adm, json={"name": "Выходной", "kind": "weekly", "weekdays": []}).json() # одну должность занимают сменщики: график у каждого свой tech = c.post("/api/positions", headers=adm, json={"name": "Слесарь", "kind": "technician"}).json() eng = c.post("/api/positions", headers=adm, json={"name": "Инженер", "kind": "engineer"}).json() for name, pos, sid in (("ivan", tech, sch["id"]), ("petr", tech, night["id"]), ("eng", eng, None)): r = c.post("/api/users", headers=adm, json={"username": name, "password": "Pass1234567", "position_id": pos["id"], "schedule_id": sid}) assert r.status_code == 201, r.text ivan, petr, engh = login("ivan", "Pass1234567"), login("petr", "Pass1234567"), login("eng", "Pass1234567") # технику нельзя создавать справочники assert c.post("/api/locations", headers=ivan, json={"name": "x"}).status_code == 403 loc = c.post("/api/locations", headers=engh, json={"name": "Цех 1"}).json() eq = c.post("/api/equipment", headers=engh, json={"name": "Насос Н-1", "location_id": loc["id"]}).json() cl = c.post("/api/checklists", headers=engh, json={"name": "ТО насоса", "items": [ {"text": "Осмотр", "answer_type": "status"}, {"text": "Давление", "answer_type": "number", "unit": "бар", "min_value": 2, "max_value": 6}, {"text": "Состояние", "answer_type": "choice", "options": ["норма", "износ"]}, {"text": "Заметки", "answer_type": "text", "required": False}]}) assert cl.status_code == 201, cl.text cl = cl.json() assert c.post("/api/checklists", headers=engh, json={"name": "bad", "items": [ {"text": "a", "answer_type": "choice"}]}).status_code == 422 # шаблон по расписанию на должность tpl = c.post("/api/order-templates", headers=engh, json={ "title": "Ежедневное ТО Н-1", "checklist_id": cl["id"], "equipment_id": eq["id"], "location_id": loc["id"], "assignee_position_id": tech["id"], "reviewer_position_id": eng["id"], "recurrence": "daily"}).json() assert tpl["next_run"] order_id = c.post(f"/api/order-templates/{tpl['id']}/run-now", headers=engh).json()["order_id"] # видимость: Иван (на смене) видит, его сменщик Пётр (выходной) — нет; уведомление — только Ивану from app import push from app.db import SessionLocal with SessionLocal() as db: ids_ = {u["username"]: u["id"] for u in c.get("/api/users", headers=adm).json()} assert push.users_for_assignment(db, None, tech["id"]) == [ids_["ivan"]] assert c.get("/api/me", headers=petr).json()["on_shift"] is False assert [o["id"] for o in c.get("/api/orders", headers=ivan).json()] == [order_id] assert c.get("/api/orders", headers=petr).json() == [] assert c.get(f"/api/orders/{order_id}", headers=petr).status_code == 404 o = c.get(f"/api/orders/{order_id}", headers=ivan).json() assert o["can_execute"] and o["status"] == "new" ids = [i["id"] for i in o["items"]] # неверные ответы assert c.put(f"/api/orders/{order_id}/items/{ids[2]}", headers=ivan, json={"value": "??"}).status_code == 422 assert c.put(f"/api/orders/{order_id}/items/{ids[1]}", headers=ivan, json={"value": "abc"}).status_code == 422 # сдать рано нельзя assert c.post(f"/api/orders/{order_id}/submit", headers=ivan).status_code == 422 c.put(f"/api/orders/{order_id}/items/{ids[0]}", headers=ivan, json={"value": "done"}) o = c.put(f"/api/orders/{order_id}/items/{ids[1]}", headers=ivan, json={"value": "7,5"}).json() assert o["status"] == "in_progress" and o["items"][1]["out_of_range"] c.put(f"/api/orders/{order_id}/items/{ids[2]}", headers=ivan, json={"value": "норма"}) o = c.post(f"/api/orders/{order_id}/submit", headers=ivan).json() assert o["status"] == "review" and o["done_count"] == 3 # техник не может проверять; инженер возвращает, затем принимает assert c.post(f"/api/orders/{order_id}/review", headers=ivan, json={"approve": True}).status_code == 403 assert c.post(f"/api/orders/{order_id}/review", headers=engh, json={"approve": False}).status_code == 422 o = c.post(f"/api/orders/{order_id}/review", headers=engh, json={"approve": False, "comment": "Давление вне нормы"}).json() assert o["status"] == "in_progress" and o["review_comment"] c.put(f"/api/orders/{order_id}/items/{ids[1]}", headers=ivan, json={"value": "4"}) c.post(f"/api/orders/{order_id}/submit", headers=ivan) o = c.post(f"/api/orders/{order_id}/review", headers=engh, json={"approve": True}).json() assert o["status"] == "completed" # ручной наряд сразу на человека, без проверяющего r = c.post("/api/orders", headers=engh, json={"title": "Разовый", "checklist_id": cl["id"], "assignee_user_id": 2}).json() assert c.get("/api/orders/stats", headers=engh).json()["by_status"]["completed"] == 1 assert r["status"] == "new" PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 100 def test_comments_photos_roles_push(monkeypatch): from app import push sent = [] monkeypatch.setattr(push, "notify", lambda ids, title, body, url="/": sent.append((sorted(i for i in ids if i), title))) with TestClient(app) as c: def login(u, p): return {"Authorization": "Bearer " + c.post("/api/auth/login", json={"username": u, "password": p}).json()["access_token"]} adm = login("admin", "admin") mod = c.post("/api/users", headers=adm, json={"username": "mod", "password": "Pass1234567", "is_moderator": True}).json() assert mod["is_moderator"] modh = login("mod", "Pass1234567") # модератор: управляет пользователями/должностями, но не админами и не сервером assert c.post("/api/positions", headers=modh, json={"name": "Электрик"}).status_code == 201 assert c.post("/api/users", headers=modh, json={"username": "u1", "password": "Pass1234567"}).status_code == 201 assert c.post("/api/users", headers=modh, json={"username": "u2", "password": "Pass1234567", "is_admin": True}).status_code == 403 assert c.patch("/api/users/1", headers=modh, json={"full_name": "x"}).status_code == 403 assert c.delete("/api/users/1", headers=modh).status_code == 403 assert c.get("/api/settings", headers=modh).status_code == 403 assert c.get("/api/settings", headers=adm).status_code == 200 tech = c.post("/api/positions", headers=modh, json={"name": "Т"}).json() uid = c.post("/api/users", headers=modh, json={"username": "t1", "password": "Pass1234567", "position_id": tech["id"]}).json()["id"] t1, other = login("t1", "Pass1234567"), login("u1", "Pass1234567") cl = c.post("/api/checklists", headers=modh, json={"name": "Ф", "items": [ {"text": "Снимок", "answer_type": "status", "photo": "required"}, {"text": "Без фото", "answer_type": "text", "photo": "none", "required": False}]}).json() o = c.post("/api/orders", headers=modh, json={"title": "Фото-наряд", "checklist_id": cl["id"], "assignee_user_id": uid}).json() assert (sorted([uid]), "Новый наряд") in sent oid, i1, i2 = o["id"], o["items"][0]["id"], o["items"][1]["id"] assert not o["can_execute"] # модератор не исполнитель # настройка каталога фото import tempfile d = tempfile.mkdtemp() assert c.put("/api/settings", headers=adm, json={"photo_dir": d, "max_photo_mb": 1}).json()["photo_dir"] == d c.put(f"/api/orders/{oid}/items/{i1}", headers=t1, json={"value": "done"}) assert c.post(f"/api/orders/{oid}/submit", headers=t1).status_code == 422 # нет обязательного фото up = lambda h, item, body, ct="image/png": c.post(f"/api/orders/{oid}/items/{item}/photos", headers={**h, "Content-Type": ct}, content=body) assert up(t1, i2, PNG).status_code == 400 # фото не предусмотрено assert up(t1, i1, b"not an image").status_code == 415 assert up(t1, i1, PNG + b"0" * 1_100_000).status_code == 413 assert up(other, i1, PNG).status_code == 404 # чужой наряд не виден r = up(t1, i1, PNG) assert r.status_code == 201, r.text pid = r.json()["items"][0]["photos"][0]["id"] assert c.get(f"/api/photos/{pid}", headers=t1).content == PNG assert c.get(f"/api/photos/{pid}", headers=other).status_code == 404 assert c.get(f"/api/photos/{pid}").status_code == 401 assert c.post(f"/api/orders/{oid}/submit", headers=t1).status_code == 200 # комментарии sent.clear() assert c.post(f"/api/orders/{oid}/comments", headers=t1, json={"text": "Готово, замечаний нет"}).status_code == 201 assert c.post(f"/api/orders/{oid}/comments", headers=other, json={"text": "x"}).status_code == 404 cm = c.post(f"/api/orders/{oid}/comments", headers=modh, json={"text": "Принято"}).json() assert [x["text"] for x in c.get(f"/api/orders/{oid}/comments", headers=t1).json()] == ["Готово, замечаний нет", "Принято"] assert c.get(f"/api/orders/{oid}", headers=t1).json()["comments_count"] == 2 assert c.delete(f"/api/comments/{cm['id']}", headers=t1).status_code == 403 assert c.delete(f"/api/comments/{cm['id']}", headers=modh).status_code == 204 assert any(t.startswith("Комментарий") for _, t in sent) # push-подписка key = c.get("/api/push/key", headers=t1).json()["key"] assert len(key) > 60 sub = {"endpoint": "https://fcm.googleapis.com/fcm/send/abc", "keys": {"p256dh": "k", "auth": "a"}} assert c.post("/api/push/subscribe", headers=t1, json=sub).status_code == 204 assert c.post("/api/push/subscribe", headers=t1, json=sub).status_code == 204 # идемпотентно assert c.post("/api/push/unsubscribe", headers=t1, json=sub).status_code == 204 def test_api_v1_tokens_sync(monkeypatch): from app import push monkeypatch.setattr(push, "notify", lambda *a, **k: None) with TestClient(app) as c: # версия и формат времени h = c.get("/api/v1/health").json() assert h["api_version"] == 1 and h["time"][-6] in "+-" # смещение часового пояса assert c.get("/api/health").status_code == 200 # старый путь работает assert "/api/v1/orders" in c.get("/openapi.json").json()["paths"] tok = c.post("/api/v1/auth/login", json={"username": "admin", "password": "admin", "device": "pytest"}).json() assert tok["expires_in"] == 3600 and tok["refresh_token"] adm = {"Authorization": "Bearer " + tok["access_token"]} # refresh с ротацией, повтор старого токена отзывает сессию t2 = c.post("/api/v1/auth/refresh", json={"refresh_token": tok["refresh_token"]}).json() assert t2["refresh_token"] != tok["refresh_token"] assert c.post("/api/v1/auth/refresh", json={"refresh_token": tok["refresh_token"]}).status_code == 401 assert c.post("/api/v1/auth/refresh", json={"refresh_token": t2["refresh_token"]}).status_code == 401 assert c.post("/api/v1/auth/refresh", json={"refresh_token": "junk"}).status_code == 401 t3 = c.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json() adm = {"Authorization": "Bearer " + t3["access_token"]} c.post("/api/v1/auth/logout", json={"refresh_token": t3["refresh_token"]}) assert c.post("/api/v1/auth/refresh", json={"refresh_token": t3["refresh_token"]}).status_code == 401 # ограничение попыток входа for _ in range(5): assert c.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"}).status_code == 401 assert c.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"}).status_code == 429 _fails_reset() # смена пароля завершает остальные сессии c.post("/api/v1/users", headers=adm, json={"username": "sync1", "password": "Pass1234567"}) lg = lambda p: c.post("/api/v1/auth/login", json={"username": "sync1", "password": p}).json() a, b = lg("Pass1234567"), lg("Pass1234567") ha = {"Authorization": "Bearer " + a["access_token"]} assert c.post("/api/v1/auth/change-password", headers=ha, json={"old_password": "bad", "new_password": "NewPass12345"}).status_code == 403 n = c.post("/api/v1/auth/change-password", headers=ha, json={"old_password": "Pass1234567", "new_password": "NewPass12345"}).json() assert c.post("/api/v1/auth/refresh", json={"refresh_token": n["refresh_token"]}).status_code == 200 assert c.post("/api/v1/auth/refresh", json={"refresh_token": b["refresh_token"]}).status_code == 401 assert c.post("/api/v1/auth/login", json={"username": "sync1", "password": "Pass1234567"}).status_code == 401 uid = c.get("/api/v1/users", headers=adm).json()[-1]["id"] # офлайн-синхронизация cl = c.post("/api/v1/checklists", headers=adm, json={"name": "S", "items": [ {"text": "a", "answer_type": "status"}, {"text": "n", "answer_type": "number", "min_value": 1, "max_value": 5}, {"text": "ph", "answer_type": "text", "required": False, "photo": "optional"}]}).json() o = c.post("/api/v1/orders", headers=adm, json={"title": "Офлайн", "checklist_id": cl["id"], "assignee_user_id": uid, "due_at": "2026-12-01T10:00:00+03:00"}).json() assert o["due_at"].startswith("2026-12-01T10:00:00") assert c.post("/api/v1/orders", headers=adm, json={"title": "UTC", "checklist_id": cl["id"], "due_at": "2026-12-01T07:00:00Z"}).json()["due_at"].startswith("2026-12-01T10:00:00+03:00") sh = {"Authorization": "Bearer " + lg("NewPass12345")["access_token"]} ids = [i["id"] for i in o["items"]] r = c.post(f"/api/v1/orders/{o['id']}/sync", headers=sh, json={"items": [ {"item_id": ids[0], "value": "done", "answered_at": "2026-10-07T12:00:00+03:00"}, {"item_id": ids[1], "value": "99", "answered_at": "2026-10-07T12:01:00+03:00"}, {"item_id": ids[1], "value": "zzz"}, {"item_id": 99999, "value": "x"}], "submit": True}).json() assert [x["status"] for x in r["results"]] == ["ok", "ok", "error", "error"] assert r["submitted"] and r["order"]["status"] == "completed" and r["order"]["items"][1]["out_of_range"] # повторная отправка — безопасна и не ломается (наряд уже сдан) assert c.post(f"/api/v1/orders/{o['id']}/sync", headers=sh, json={"items": []}).status_code == 403 # устаревший ответ не затирает более свежий o2 = c.post("/api/v1/orders", headers=adm, json={"title": "Конфликт", "checklist_id": cl["id"], "assignee_user_id": uid}).json() i2 = o2["items"][1]["id"] c.put(f"/api/v1/orders/{o2['id']}/items/{i2}", headers=sh, json={"value": "3"}) r = c.post(f"/api/v1/orders/{o2['id']}/sync", headers=sh, json={"items": [ {"item_id": i2, "value": "1", "answered_at": "2020-01-01T00:00:00Z"}]}).json() assert r["results"][0]["status"] == "stale" and r["order"]["items"][1]["value"] == "3" # updated_since + include_items full = c.get("/api/v1/orders?include_items=true", headers=sh).json() assert all(x["items"] is not None for x in full) cursor = max(x["updated_at"] for x in full) assert c.get("/api/v1/orders", headers=sh, params={"updated_since": cursor}).json() == [] c.post(f"/api/v1/orders/{o2['id']}/comments", headers=adm, json={"text": "ping"}) ch = c.get("/api/v1/orders", headers=sh, params={"updated_since": cursor}).json() assert [x["id"] for x in ch] == [o2["id"]] # идемпотентная загрузка фото up = lambda: c.post(f"/api/v1/orders/{o2['id']}/items/{o2['items'][2]['id']}/photos", headers={**sh, "Content-Type": "image/png", "Idempotency-Key": "abc-1"}, content=PNG) assert up().status_code == 201 assert len(up().json()["items"][2]["photos"]) == 1 # FCM: регистрация устройства и отправка tk = "x" * 40 assert c.post("/api/v1/push/device", headers=sh, json={"token": tk}).status_code == 204 assert c.post("/api/v1/push/device", headers=sh, json={"token": tk}).status_code == 204 import httpx from app.db import SessionLocal from app import settings_store as ss calls = [] class Creds: project_id, token, valid = "proj", "tok", True monkeypatch.setattr(push, "_fcm_auth", lambda p: Creds) monkeypatch.setattr(httpx, "post", lambda url, **kw: (calls.append((url, kw["json"])), type("R", (), {"status_code": 200, "text": ""})())[1]) with SessionLocal() as db: push._send_fcm(db, {"title": "T", "body": "B", "url": "/x"}, [uid]) assert calls == [] # FCM не настроен ss.set_(db, "fcm_credentials_path", "dummy.json") db.commit() push._send_fcm(db, {"title": "T", "body": "B", "url": "/x"}, [uid]) assert calls[0][0].endswith("/projects/proj/messages:send") and calls[0][1]["message"]["token"] == tk assert c.request("DELETE", "/api/v1/push/device", headers=sh, json={"token": tk}).status_code == 204 def _fails_reset(): from app.routers import catalog catalog._fails.clear() def test_pattern_schedule(): """3 дня / отсыпной / 3 ночи / отсыпной / 3 выходных (11 дней), и «сутки через трое».""" from app.models import ShiftSchedule from datetime import date, datetime W = lambda d, a, b, label="": {"type": "work", "days": d, "start_time": a, "end_time": b, "label": label} O = lambda d: {"type": "off", "days": d} s = ShiftSchedule(kind="pattern", cycle_start=date(2026, 1, 1), pattern=[W(3, "08:00", "20:00", "день"), O(1), W(3, "20:00", "08:00", "ночь"), O(1), O(3)]) at = lambda m, d, h, mi=0: logic.shift_active(s, datetime(2026, m, d, h, mi)) assert at(1, 1, 9) and at(1, 3, 19) and not at(1, 3, 21) and not at(1, 4, 12) # дни, потом отсыпной assert not at(1, 5, 12) and at(1, 5, 21) and at(1, 6, 3) # ночь 5-го уходит на 6-е assert at(1, 7, 7) and not at(1, 7, 9) # ночь 7-го закончилась 8:00 8-го? нет — 7-го в 7:00 идёт смена 6-го assert not at(1, 8, 9) and not at(1, 8, 21) # 8-е — отсыпной после ночей assert not at(1, 9, 12) and not at(1, 10, 12) and not at(1, 11, 12) # 3 выходных assert at(1, 12, 9) # цикл 11 дней -> 12 янв снова день sut = ShiftSchedule(kind="pattern", cycle_start=date(2026, 1, 1), pattern=[W(1, "08:00", "08:00"), O(3)]) assert logic.shift_active(sut, datetime(2026, 1, 1, 8, 0)) and logic.shift_active(sut, datetime(2026, 1, 2, 7, 59)) assert not logic.shift_active(sut, datetime(2026, 1, 2, 8, 0)) and logic.shift_active(sut, datetime(2026, 1, 5, 12)) assert logic.shift_starting(ShiftSchedule(kind="pattern", pattern=[]), date(2026, 1, 1)) is None def test_schedule_api_pattern(): with TestClient(app) as c: h = {"Authorization": "Bearer " + c.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]} body = {"name": "Сутки/трое", "kind": "pattern", "cycle_start": "2026-01-01", "pattern": [{"type": "work", "days": 1, "start_time": "08:00", "end_time": "08:00"}, {"type": "off", "days": 3}]} r = c.post("/api/v1/schedules", headers=h, json=body) assert r.status_code == 201, r.text assert r.json()["pattern"][0]["start_time"] == "08:00:00" assert c.post("/api/v1/schedules", headers=h, json={**body, "name": "bad", "pattern": [{"type": "off", "days": 2}]}).status_code == 422 pv = c.post("/api/v1/schedules/preview", headers=h, json={"schedule": body, "from_date": "2026-01-01", "days": 5}).json() assert [d["work"] for d in pv] == [True, False, False, False, True] assert pv[0]["end"].startswith("2026-01-02T08:00:00") sid = r.json()["id"] up = c.patch(f"/api/v1/schedules/{sid}", headers=h, json={"pattern": [{"type": "work", "days": 2}, {"type": "off", "days": 1}]}) assert up.status_code == 200 and up.json()["pattern"][0]["days"] == 2 assert c.get("/api/v1/schedules", headers=h).status_code == 200 def test_report_completed_orders(): from datetime import datetime, timedelta with TestClient(app) as c: H = lambda u, p: {"Authorization": "Bearer " + c.post("/api/v1/auth/login", json={"username": u, "password": p}).json()["access_token"]} adm = H("admin", "admin") pos = c.post("/api/v1/positions", headers=adm, json={"name": "Слесарь РП"}).json() eng = c.post("/api/v1/positions", headers=adm, json={"name": "Инженер РП", "kind": "engineer"}).json() c.post("/api/v1/users", headers=adm, json={"username": "rep_t", "full_name": "Пётр Сидоров", "password": "Pass1234567", "position_id": pos["id"]}) c.post("/api/v1/users", headers=adm, json={"username": "rep_e", "password": "Pass1234567", "position_id": eng["id"]}) uid = [u for u in c.get("/api/v1/users", headers=adm).json() if u["username"] == "rep_t"][0]["id"] t, e = H("rep_t", "Pass1234567"), H("rep_e", "Pass1234567") eq = c.post("/api/v1/equipment", headers=adm, json={"name": "Компрессор К-2", "inventory_no": "K-002"}).json() cl = c.post("/api/v1/checklists", headers=adm, json={"name": "ТО компрессора", "items": [{"text": "осмотр", "answer_type": "status"}]}).json() o = c.post("/api/v1/orders", headers=adm, json={"title": "ТО К-2", "checklist_id": cl["id"], "equipment_id": eq["id"], "assignee_user_id": uid}).json() c.put(f"/api/v1/orders/{o['id']}/items/{o['items'][0]['id']}", headers=t, json={"value": "done"}) assert c.post(f"/api/v1/orders/{o['id']}/submit", headers=t).json()["status"] == "completed" # смена должности после выполнения не меняет отчёт other = c.post("/api/v1/positions", headers=adm, json={"name": "Другая"}).json() c.patch(f"/api/v1/users/{uid}", headers=adm, json={"position_id": other["id"]}) now = datetime.now() q = {"from": (now - timedelta(hours=2)).isoformat(), "to": (now + timedelta(hours=2)).isoformat()} j = c.get("/api/v1/reports/completed-orders", headers=e, params=q).json() r = [x for x in j["rows"] if x["order_id"] == o["id"]][0] assert (r["equipment"], r["checklist"], r["executor"], r["position"]) == ("Компрессор К-2", "ТО компрессора", "Пётр Сидоров", "Слесарь РП") assert r["finished_at"][-6] in "+-" # фильтры assert c.get("/api/v1/reports/completed-orders", headers=e, params={**q, "equipment_id": 99999}).json()["total"] == 0 assert c.get("/api/v1/reports/completed-orders", headers=e, params={**q, "position_id": pos["id"]}).json()["total"] == 1 far = {"from": (now + timedelta(days=1)).isoformat(), "to": (now + timedelta(days=2)).isoformat()} assert c.get("/api/v1/reports/completed-orders", headers=e, params=far).json()["total"] == 0 # обязательность и корректность периода, права assert c.get("/api/v1/reports/completed-orders", headers=e).status_code == 422 assert c.get("/api/v1/reports/completed-orders", headers=e, params={"from": q["to"], "to": q["from"]}).status_code == 422 assert c.get("/api/v1/reports/completed-orders", headers=t, params=q).status_code == 403 # экспорт pdf = c.get("/api/v1/reports/completed-orders", headers=e, params={**q, "format": "pdf"}) assert pdf.status_code == 200 and pdf.content[:5] == b"%PDF-" and "attachment" in pdf.headers["content-disposition"] assert c.get("/api/v1/reports/completed-orders", headers=e, params={**far, "format": "pdf"}).content[:5] == b"%PDF-" # пустой отчёт csv = c.get("/api/v1/reports/completed-orders", headers=e, params={**q, "format": "csv"}).content.decode("utf-8-sig") assert "Компрессор К-2" in csv and "Слесарь РП" in csv assert [x["id"] for x in c.get("/api/v1/reports", headers=e).json()] == ["completed-orders", "equipment-state", "execution", "people"] @pytest.mark.skipif(not os.environ["DATABASE_URL"].startswith("sqlite"), reason="проверка неизменности файла БД — только для SQLite") def test_order_search(): with TestClient(app) as c: h = {"Authorization": "Bearer " + c.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json()["access_token"]} cl = c.post("/api/v1/checklists", headers=h, json={"name": "П", "items": []}).json() eq = c.post("/api/v1/equipment", headers=h, json={"name": "Гидронасос ЖЕЛТЫЙ"}).json() mk = lambda t, **k: c.post("/api/v1/orders", headers=h, json={"title": t, "checklist_id": cl["id"], **k}).json()["id"] a = mk("Обход Котельной"); b = mk("Смазка 100%_готово"); d = mk("Прочее", equipment_id=eq["id"]) s = lambda q: sorted(o["id"] for o in c.get("/api/v1/orders", headers=h, params={"q": q}).json()) assert s("котел") == [a] and s("КОТЕЛ") == [a] # регистр кириллицы не важен assert s("желтый") == [d] # по названию оборудования assert s("100%_") == [b] and s("%") == [b] # спецсимволы LIKE — буквально assert s("#%d" % d) == [d] and s(str(a)) == [a] # по номеру assert s("нет такого") == [] and s(" ") != [] def test_apk_download(monkeypatch, tmp_path): from pathlib import Path import os, time from app import config bundled = tmp_path / "release" # APK, поставляемый в образе bundled.mkdir() monkeypatch.setattr(config, "RELEASE_DIR", bundled) with TestClient(app) as c: assert c.get("/app.apk").status_code == 404 # нигде нет файла (bundled / "image.apk").write_bytes(b"PK-image") r = c.get("/app.apk") # есть только APK из образа -> раздаётся он assert r.status_code == 200 and r.content == b"PK-image" assert r.headers["content-type"] == "application/vnd.android.package-archive" assert c.head("/app.apk").status_code == 200 d = Path(_DATA) / "apk" # загруженный вручную важнее и берётся самый свежий d.mkdir(exist_ok=True) (d / "old.apk").write_bytes(b"PK-old") os.utime(d / "old.apk", (time.time() - 100, time.time() - 100)) (d / "new.apk").write_bytes(b"PK-new") assert c.get("/app.apk").content == b"PK-new" for f in d.glob("*.apk"): f.unlink() assert c.get("/app.apk").content == b"PK-image" # после удаления снова APK из образа def test_mobile_pwa_assets(): import json with TestClient(app) as c: r = c.get("/m") assert r.status_code == 200 and "text/html" in r.headers["content-type"] and "/static/m/m.js" in r.text assert c.get("/m/").status_code == 200 man = json.loads(c.get("/static/m/manifest.webmanifest").text) assert man["start_url"] == "/m" and man["display"] == "standalone" for icon in man["icons"]: png = c.get(icon["src"]) assert png.status_code == 200 and png.content[:8] == b"\x89PNG\r\n\x1a\n", icon["src"] sw = c.get("/sw.js") assert sw.status_code == 200 and "shell-m" in sw.text and sw.headers["service-worker-allowed"] == "/" # все файлы оболочки из service worker реально существуют import re shell = re.search(r"const SHELL = \[(.*?)\];", sw.text, re.S).group(1) for path in re.findall(r'"([^"]+)"', shell): assert c.get(path).status_code == 200, path