334 lines
26 KiB
Python
334 lines
26 KiB
Python
"""Портал: инвентарь и склады, заявки на инструмент, инциденты, суточные задания, карта, карточка сотрудника, отчёт."""
|
||
import os
|
||
import tempfile
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
|
||
os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "portal.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 # noqa: E402
|
||
from app.main import app # noqa: E402
|
||
|
||
V1 = "/api/v1"
|
||
PNG = b"\x89PNG\r\n\x1a\n" + b"\0" * 64
|
||
PW = "Pass1234567"
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def env():
|
||
with TestClient(app) as c:
|
||
def login(u, p=PW):
|
||
r = c.post(f"{V1}/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")
|
||
tech_pos = c.post(f"{V1}/positions", headers=adm, json={"name": "П-Слесарь", "kind": "technician"}).json()
|
||
eng_pos = c.post(f"{V1}/positions", headers=adm, json={"name": "П-Инженер", "kind": "engineer"}).json()
|
||
ids = {}
|
||
for name, pos, extra in (("p_tech", tech_pos, {}), ("p_tech2", tech_pos, {}), ("p_eng", eng_pos, {}),
|
||
("p_store", tech_pos, {"is_storekeeper": True})):
|
||
r = c.post(f"{V1}/users", headers=adm, json={"username": name, "password": PW, "full_name": name.upper(),
|
||
"position_id": pos["id"], **extra})
|
||
assert r.status_code == 201, r.text
|
||
ids[name] = r.json()["id"]
|
||
eq = c.post(f"{V1}/equipment", headers=adm, json={"name": "П-Насос", "icon": "builtin:pump"}).json()
|
||
unit = c.post(f"{V1}/equipment/{eq['id']}/service-units", headers=adm,
|
||
json={"name": "Подшипник", "mode": "hours", "interval_hours": 100}).json()
|
||
yield {"c": c, "adm": adm, "tech": login("p_tech"), "tech2": login("p_tech2"), "eng": login("p_eng"),
|
||
"store": login("p_store"), "ids": ids, "eq": eq, "unit": unit}
|
||
|
||
|
||
def test_inventory_roles_and_moves(env):
|
||
c, store, tech, adm = env["c"], env["store"], env["tech"], env["adm"]
|
||
tid = env["ids"]["p_tech"]
|
||
assert c.get(f"{V1}/me", headers=store).json()["is_storekeeper"] is True
|
||
# сотрудник не может править каталог и склады
|
||
assert c.post(f"{V1}/warehouses", headers=tech, json={"name": "Х"}).status_code == 403
|
||
assert c.post(f"{V1}/inventory/items", headers=tech, json={"name": "Х"}).status_code == 403
|
||
w1 = c.post(f"{V1}/warehouses", headers=store, json={"name": "Склад №1"}).json()
|
||
w2 = c.post(f"{V1}/warehouses", headers=store, json={"name": "Склад №2"}).json()
|
||
drill = c.post(f"{V1}/inventory/items", headers=store, json={"name": "Дрель", "category": "Электроинструмент",
|
||
"inventory_no": "ИН-1"}).json()
|
||
keys = c.post(f"{V1}/inventory/items", headers=store, json={"name": "Набор ключей"}).json()
|
||
mv = lambda body, h=store: c.post(f"{V1}/inventory/moves", headers=h, json=body)
|
||
assert mv({"kind": "receive", "item_id": drill["id"], "qty": 2, "to": {"warehouse_id": w1["id"]}}).status_code == 201
|
||
assert mv({"kind": "receive", "item_id": keys["id"], "qty": 5, "to": {"warehouse_id": w1["id"]}}).status_code == 201
|
||
# сотрудник не может выдавать
|
||
assert mv({"kind": "move", "item_id": drill["id"], "from": {"warehouse_id": w1["id"]}, "to": {"user_id": tid}}, tech).status_code == 403
|
||
# выдача лично сотруднику
|
||
r = mv({"kind": "move", "item_id": drill["id"], "qty": 1, "from": {"warehouse_id": w1["id"]}, "to": {"user_id": tid}})
|
||
assert r.status_code == 201, r.text
|
||
# нельзя выдать больше, чем есть
|
||
assert mv({"kind": "move", "item_id": drill["id"], "qty": 5, "from": {"warehouse_id": w1["id"]}, "to": {"user_id": tid}}).status_code == 422
|
||
# перемещение между складами, списание
|
||
assert mv({"kind": "move", "item_id": keys["id"], "qty": 2, "from": {"warehouse_id": w1["id"]}, "to": {"warehouse_id": w2["id"]}}).status_code == 201
|
||
assert mv({"kind": "writeoff", "item_id": keys["id"], "qty": 1, "from": {"warehouse_id": w2["id"]}, "note": "сломан"}).status_code == 201
|
||
# «откуда» должно быть ровно одно
|
||
assert mv({"kind": "move", "item_id": keys["id"], "from": {"warehouse_id": w1["id"], "user_id": tid}, "to": {"warehouse_id": w2["id"]}}).status_code == 422
|
||
# сотрудник видит только своё
|
||
mine = c.get(f"{V1}/me/inventory", headers=tech).json()
|
||
assert [(x["item_name"], x["qty"]) for x in mine] == [("Дрель", 1)]
|
||
assert c.get(f"{V1}/inventory/stock", headers=tech, params={"warehouse_id": w1["id"]}).status_code == 403
|
||
assert c.get(f"{V1}/inventory/stock", headers=tech).json() == mine
|
||
st = {x["item_name"]: x["qty"] for x in c.get(f"{V1}/inventory/stock", headers=store, params={"warehouse_id": w1["id"]}).json()}
|
||
assert st == {"Дрель": 1, "Набор ключей": 3}
|
||
items = {x["name"]: x for x in c.get(f"{V1}/inventory/items", headers=store).json()}
|
||
assert items["Дрель"]["in_stock"] == 1 and items["Дрель"]["issued"] == 1
|
||
whs = {x["name"]: x for x in c.get(f"{V1}/warehouses", headers=tech).json()}
|
||
assert whs["Склад №2"]["total_qty"] == 1
|
||
moves = c.get(f"{V1}/inventory/moves", headers=store, params={"item_id": drill["id"]}).json()
|
||
assert [m["kind"] for m in moves] == ["issue", "receive"]
|
||
assert [m["kind"] for m in c.get(f"{V1}/inventory/moves", headers=tech).json()] == ["issue"]
|
||
# возврат на склад
|
||
assert mv({"kind": "move", "item_id": drill["id"], "from": {"user_id": tid}, "to": {"warehouse_id": w1["id"]}}).status_code == 201
|
||
assert c.get(f"{V1}/me/inventory", headers=tech).json() == []
|
||
# склад с инвентарём не удаляется; зона на карте
|
||
assert c.delete(f"{V1}/warehouses/{w1['id']}", headers=store).status_code == 409
|
||
assert c.put(f"{V1}/warehouses/{w1['id']}/zone", headers=store, json={"zone": {"x": 10, "y": 10, "w": 20, "h": 15}}).status_code == 200
|
||
assert c.put(f"{V1}/warehouses/{w1['id']}/zone", headers=store, json={"zone": {"x": 90, "y": 10, "w": 20, "h": 15}}).status_code == 422
|
||
assert c.put(f"{V1}/warehouses/{w1['id']}/zone", headers=tech, json={"zone": None}).status_code == 403
|
||
# обычный администратор тоже может всё
|
||
assert c.post(f"{V1}/warehouses", headers=adm, json={"name": "Склад админа"}).status_code == 201
|
||
|
||
|
||
def test_tool_requests(env):
|
||
c, tech, tech2, store = env["c"], env["tech"], env["tech2"], env["store"]
|
||
bad = c.post(f"{V1}/tool-requests", headers=tech, json={"what": "Мультиметр", "why": "Старый сломан", "link": "javascript:alert(1)"})
|
||
assert bad.status_code == 422
|
||
assert c.post(f"{V1}/tool-requests", headers=tech, json={"what": "Мультиметр"}).status_code == 422 # «зачем» обязательно
|
||
r = c.post(f"{V1}/tool-requests", headers=tech, json={"what": "Мультиметр Fluke", "why": "Старый сломан",
|
||
"link": "https://example.com/item/1", "qty": 1}).json()
|
||
assert r["status"] == "new" and r["author_name"] == "P_TECH"
|
||
c.post(f"{V1}/tool-requests", headers=tech2, json={"what": "Перчатки", "why": "Износились"})
|
||
assert [x["what"] for x in c.get(f"{V1}/tool-requests", headers=tech).json()] == ["Мультиметр Fluke"]
|
||
assert len(c.get(f"{V1}/tool-requests", headers=store).json()) >= 2
|
||
assert c.post(f"{V1}/tool-requests/{r['id']}/answer", headers=tech, json={"status": "approved"}).status_code == 403
|
||
assert c.post(f"{V1}/tool-requests/{r['id']}/answer", headers=store, json={"status": "rejected"}).status_code == 422
|
||
a = c.post(f"{V1}/tool-requests/{r['id']}/answer", headers=store, json={"status": "ordered", "answer": "Заказан"}).json()
|
||
assert a["status"] == "ordered" and a["handled_by_name"] == "P_STORE"
|
||
assert c.delete(f"{V1}/tool-requests/{r['id']}", headers=tech).status_code == 403 # уже не новая
|
||
|
||
|
||
def test_incidents_affect_state(env):
|
||
c, tech, eng, adm, eq, unit = env["c"], env["tech"], env["eng"], env["adm"], env["eq"], env["unit"]
|
||
c.post(f"{V1}/equipment/{eq['id']}/meter", headers=eng, json={"hours": 95}) # узел «скоро» (осталось 5 из 100)
|
||
h = c.get(f"{V1}/equipment-health", headers=tech).json()[str(eq["id"])]
|
||
assert h["state"] == "soon" and h["soon"] == 1
|
||
other = c.post(f"{V1}/equipment", headers=adm, json={"name": "П-Другая"}).json()
|
||
other_unit = c.post(f"{V1}/equipment/{other['id']}/service-units", headers=adm, json={"name": "X", "mode": "hours", "interval_hours": 10}).json()
|
||
assert c.post(f"{V1}/incidents", headers=tech, json={"equipment_id": eq["id"], "title": "Течь",
|
||
"units": [{"unit_id": other_unit["id"]}]}).status_code == 422
|
||
r = c.post(f"{V1}/incidents", headers=tech, json={"equipment_id": eq["id"], "title": "Заклинил подшипник",
|
||
"description": "Шум, нагрев", "units": [{"unit_id": unit["id"], "replaced": True}],
|
||
"replacement_note": "Поставлен новый"})
|
||
assert r.status_code == 201, r.text
|
||
inc = r.json()
|
||
assert inc["replaced"] is True and inc["units"][0]["unit_name"] == "Подшипник" and inc["status"] == "open"
|
||
# замена сбросила узел, открытый инцидент — «Авария»
|
||
u = c.get(f"{V1}/equipment/{eq['id']}/service-units", headers=tech).json()[0]
|
||
assert u["state"] == "ok" and u["used_hours"] == 0
|
||
assert c.get(f"{V1}/service-units/{unit['id']}/events", headers=tech).json()[0]["kind"] == "incident"
|
||
m = {x["id"]: x for x in c.get(f"{V1}/map", headers=tech).json()["equipment"]}
|
||
assert m[eq["id"]]["state"] == "alarm" and m[eq["id"]]["open_incidents"] == 1
|
||
assert c.post(f"{V1}/incidents/{inc['id']}/close", headers=env["tech2"], json={}).status_code == 403
|
||
assert c.post(f"{V1}/incidents/{inc['id']}/close", headers=tech, json={"note": "Устранено"}).json()["status"] == "closed"
|
||
assert c.get(f"{V1}/equipment-health", headers=tech).json()[str(eq["id"])]["state"] == "ok"
|
||
assert len(c.get(f"{V1}/incidents", headers=tech, params={"equipment_id": eq["id"]}).json()) == 1
|
||
|
||
|
||
def test_daily_tasks(env):
|
||
c, tech, tech2, eng, eq, unit = env["c"], env["tech"], env["tech2"], env["eng"], env["eq"], env["unit"]
|
||
ids = env["ids"]
|
||
# техник не может ставить задания другим, но может себе
|
||
assert c.post(f"{V1}/tasks", headers=tech, json={"title": "Чужое", "assignee_id": ids["p_tech2"]}).status_code == 403
|
||
own = c.post(f"{V1}/tasks", headers=tech, json={"title": "Себе задание"}).json()
|
||
assert own["assignee_id"] == ids["p_tech"] and own["status"] == "assigned"
|
||
t = c.post(f"{V1}/tasks", headers=eng, json={"title": "Проверить насос", "assignee_id": ids["p_tech"],
|
||
"equipment_id": eq["id"], "unit_id": unit["id"]}).json()
|
||
assert t["equipment_name"] == "П-Насос" and t["unit_name"] == "Подшипник"
|
||
st = lambda s, cm="", h=tech: c.post(f"{V1}/tasks/{t['id']}/status", headers=h, json={"status": s, "comment": cm})
|
||
assert st("in_progress").status_code == 200
|
||
assert st("paused").status_code == 422 # без комментария нельзя
|
||
assert st("paused", "Жду запчасть").status_code == 200
|
||
assert st("done").status_code == 422
|
||
assert st("in_progress").status_code == 200
|
||
# чужой техник не видит задание
|
||
assert c.get(f"{V1}/tasks/{t['id']}", headers=tech2).status_code == 404
|
||
# фото
|
||
r = c.post(f"{V1}/tasks/{t['id']}/photos", headers=tech, content=PNG)
|
||
assert r.status_code == 201 and len(r.json()["photos"]) == 1
|
||
assert c.get(f"{V1}/task-photos/{r.json()['photos'][0]}", headers=eng).status_code == 200
|
||
assert c.post(f"{V1}/tasks/{t['id']}/photos", headers=tech, content=b"not an image").status_code == 415
|
||
done = st("done", "Подшипник смазан").json()
|
||
assert done["status"] == "done" and done["finished_at"] and done["last_comment"] == "Подшипник смазан"
|
||
assert [e["to_status"] for e in done["events"]] == ["assigned", "in_progress", "paused", "in_progress", "done"]
|
||
assert st("in_progress").status_code == 403 # завершённое техник уже не меняет
|
||
assert st("in_progress", "x", eng).status_code == 422 # из «выполнен» — только вернуть в «назначен»
|
||
assert st("assigned", "Переделать").status_code == 403 # техник не возвращает завершённое
|
||
assert st("assigned", "Переделать", eng).status_code == 200
|
||
mine = c.get(f"{V1}/tasks", headers=tech, params={"mine": True}).json()
|
||
assert {x["title"] for x in mine} >= {"Себе задание", "Проверить насос"}
|
||
assert all(x["assignee_id"] == ids["p_tech"] for x in c.get(f"{V1}/tasks", headers=tech).json())
|
||
|
||
|
||
def test_icons_map_card(env):
|
||
c, adm, tech, store, eng, eq = env["c"], env["adm"], env["tech"], env["store"], env["eng"], env["eq"]
|
||
icons = c.get(f"{V1}/icons", headers=tech).json()
|
||
assert len(icons) >= 12 and all(i["id"].startswith("builtin:") for i in icons)
|
||
assert c.get(icons[0]["url"]).status_code == 200
|
||
assert c.patch(f"{V1}/equipment/{eq['id']}", headers=adm, json={"icon": "file:../../etc/passwd"}).status_code == 422
|
||
assert c.patch(f"{V1}/equipment/{eq['id']}", headers=adm, json={"icon": "builtin:motor"}).json()["icon"] == "builtin:motor"
|
||
r = c.put(f"{V1}/equipment/{eq['id']}/icon", headers=eng, content=PNG)
|
||
assert r.status_code == 200 and r.json()["icon"].startswith("file:")
|
||
assert c.get(f"{V1}/equipment/{eq['id']}/icon", headers=tech).content == PNG
|
||
# карта: картинку и иконки ставит админ, зоны складов — кладовщик
|
||
assert c.put(f"{V1}/map/image", headers=store, content=PNG).status_code == 403
|
||
assert c.put(f"{V1}/map/image", headers=adm, content=PNG).status_code == 204
|
||
assert c.get(f"{V1}/map/image", headers=tech).content == PNG
|
||
assert c.put(f"{V1}/equipment/{eq['id']}/map", headers=store, json={"x": 10, "y": 20}).status_code == 403
|
||
assert c.put(f"{V1}/equipment/{eq['id']}/map", headers=adm, json={"x": 10, "y": None}).status_code == 422
|
||
assert c.put(f"{V1}/equipment/{eq['id']}/map", headers=adm, json={"x": 10.5, "y": 20}).json()["map_x"] == 10.5
|
||
m = c.get(f"{V1}/map", headers=store).json()
|
||
assert m["has_image"] and m["can_edit_warehouses"] and not m["can_edit_equipment"]
|
||
assert any(e["id"] == eq["id"] and e["x"] == 10.5 for e in m["equipment"])
|
||
# карточка сотрудника
|
||
tid = env["ids"]["p_tech"]
|
||
assert c.patch(f"{V1}/users/{tid}/profile", headers=tech, json={"skills": "Сварка, электрика", "phone": "+7 900"}).status_code == 200
|
||
assert c.patch(f"{V1}/users/{env['ids']['p_tech2']}/profile", headers=tech, json={"skills": "x"}).status_code == 403
|
||
assert c.put(f"{V1}/users/{tid}/photo", headers=tech, content=PNG).json()["has_photo"] is True
|
||
card = c.get(f"{V1}/users/{tid}/card", headers=env["tech2"]).json()
|
||
assert card["user"]["skills"] == "Сварка, электрика" and card["user"]["position"]["name"] == "П-Слесарь"
|
||
assert any(t["title"] == "Проверить насос" for t in card["tasks"])
|
||
assert card["inventory"] is None and card["can_edit"] is False # чужой инвентарь коллеге не показываем
|
||
assert c.get(f"{V1}/users/{tid}/card", headers=store).json()["inventory"] == []
|
||
assert c.get(f"{V1}/users/{tid}/photo", headers=env["tech2"]).content == PNG
|
||
|
||
|
||
def test_execution_report(env):
|
||
c, adm, eng, tech, eq = env["c"], env["adm"], env["eng"], env["tech"], env["eq"]
|
||
cl = c.post(f"{V1}/checklists", headers=adm, json={"name": "П-Осмотр", "items": [
|
||
{"text": "Давление", "answer_type": "number", "min_value": 1, "max_value": 5, "unit": "бар"}]}).json()
|
||
o = c.post(f"{V1}/orders", headers=eng, json={"title": "Осмотр насоса", "checklist_id": cl["id"],
|
||
"equipment_id": eq["id"], "assignee_user_id": env["ids"]["p_tech"]}).json()
|
||
item = c.get(f"{V1}/orders/{o['id']}", headers=tech).json()["items"][0]
|
||
c.put(f"{V1}/orders/{o['id']}/items/{item['id']}", headers=tech, json={"value": "9"})
|
||
assert c.post(f"{V1}/orders/{o['id']}/submit", headers=tech).json()["status"] == "completed"
|
||
o2 = c.post(f"{V1}/orders", headers=eng, json={"title": "Второй осмотр", "checklist_id": cl["id"], "equipment_id": eq["id"]}).json()
|
||
c.post(f"{V1}/orders/{o2['id']}/cancel", headers=eng)
|
||
now = config.now()
|
||
q = {"from": (now - timedelta(days=1)).isoformat(), "to": (now + timedelta(days=1)).isoformat(), "equipment_id": eq["id"]}
|
||
assert c.get(f"{V1}/reports/execution", headers=tech, params=q).status_code == 403
|
||
rep = c.get(f"{V1}/reports/execution", headers=eng, params=q).json()
|
||
row = rep["rows"][0]
|
||
assert row["equipment"] == "П-Насос"
|
||
assert row["cats"]["inspection"] == {"planned": 2, "done": 1, "percent": 50}
|
||
assert row["cats"]["tasks"]["planned"] >= 1
|
||
texts = [d["text"] for d in row["deviations"]]
|
||
assert any("Давление: 9 бар" in t for t in texts) and any("Наряд отменён" in t for t in texts)
|
||
assert any("Жду запчасть" in t for t in texts)
|
||
pdf = c.get(f"{V1}/reports/execution", headers=eng, params={**q, "format": "pdf"})
|
||
assert pdf.status_code == 200 and pdf.content[:5] == b"%PDF-"
|
||
csv = c.get(f"{V1}/reports/execution", headers=eng, params={**q, "format": "csv"})
|
||
assert csv.status_code == 200 and "Суточные осмотры" in csv.content.decode("utf-8-sig")
|
||
|
||
|
||
def test_task_transfer(env):
|
||
c, tech, tech2, eng, ids = env["c"], env["tech"], env["tech2"], env["eng"], env["ids"]
|
||
t = c.post(f"{V1}/tasks", headers=eng, json={"title": "Передаваемое", "assignee_id": ids["p_tech"]}).json()
|
||
assert t["can_transfer"] is False or True
|
||
c.post(f"{V1}/tasks/{t['id']}/status", headers=tech, json={"status": "in_progress"})
|
||
tr = lambda h, to, cm="Занят на аварии": c.post(f"{V1}/tasks/{t['id']}/transfer", headers=h, json={"assignee_id": to, "comment": cm})
|
||
assert tr(tech2, ids["p_tech2"]).status_code == 404 # чужое задание не видно
|
||
assert tr(tech, ids["p_tech2"], "").status_code == 422 # причина обязательна
|
||
r = tr(tech, ids["p_tech2"]) # исполнитель передаёт
|
||
assert r.status_code == 200, r.text
|
||
d = r.json()
|
||
assert d["assignee_id"] == ids["p_tech2"] and d["status"] == "assigned"
|
||
ev = d["events"][-1]
|
||
assert ev["kind"] == "transfer" and ev["from_user_name"] == "P_TECH" and ev["to_user_name"] == "P_TECH2"
|
||
assert tr(tech2, ids["p_tech2"]).status_code == 409
|
||
assert tr(eng, ids["p_tech"]).status_code == 200 # заказчик (поставивший) возвращает
|
||
c.post(f"{V1}/tasks/{t['id']}/status", headers=tech, json={"status": "done", "comment": "ok"})
|
||
assert tr(eng, ids["p_tech2"]).status_code == 403 # завершённое не передаётся
|
||
|
||
|
||
def test_public_and_external_incidents(env):
|
||
c, adm, tech, eq = env["c"], env["adm"], env["tech"], env["eq"]
|
||
loc = c.post(f"{V1}/locations", headers=adm, json={"name": "П-Цех"}).json()
|
||
sub = c.post(f"{V1}/locations", headers=adm, json={"name": "П-Участок", "parent_id": loc["id"]}).json()
|
||
other = c.post(f"{V1}/locations", headers=adm, json={"name": "П-Склад"}).json()
|
||
c.patch(f"{V1}/equipment/{eq['id']}", headers=adm, json={"location_id": sub["id"], "inventory_no": "П-ИНВ-1"})
|
||
o = c.get(f"{V1}/public/report-options").json() # без входа
|
||
assert o["enabled"] and any(e["id"] == eq["id"] and e["location_id"] == sub["id"] for e in o["equipment"])
|
||
base = {"location_id": loc["id"], "equipment_id": eq["id"], "description": "Сильная течь масла под машиной",
|
||
"reporter_name": "Подрядчик Петров", "reporter_contact": "+7 900 123-45-67"}
|
||
assert c.post(f"{V1}/public/incidents", json={**base, "reporter_contact": ""}).status_code == 422
|
||
assert c.post(f"{V1}/public/incidents", json={**base, "description": ""}).status_code == 422
|
||
assert c.post(f"{V1}/public/incidents", json={**base, "location_id": other["id"]}).status_code == 422 # машина не в этом месте
|
||
r = c.post(f"{V1}/public/incidents", json=base) # родительское место включает вложенное
|
||
assert r.status_code == 201 and r.json()["number"]
|
||
inc = c.get(f"{V1}/incidents/{r.json()['number']}", headers=tech).json()
|
||
assert inc["source"] == "public" and inc["reporter_contact"] == "+7 900 123-45-67" and inc["title"].startswith("Сильная течь")
|
||
assert c.get("/report").status_code == 200
|
||
# выключение формы
|
||
c.put(f"{V1}/settings", headers=adm, json={"public_incidents": False})
|
||
assert c.get(f"{V1}/public/report-options").json()["enabled"] is False
|
||
assert c.post(f"{V1}/public/incidents", json=base).status_code == 403
|
||
c.put(f"{V1}/settings", headers=adm, json={"public_incidents": True})
|
||
# внешняя система по ключу
|
||
assert c.post(f"{V1}/api-keys", headers=tech, json={"name": "x"}).status_code == 403
|
||
k = c.post(f"{V1}/api-keys", headers=adm, json={"name": "SCADA"}).json()
|
||
assert k["key"].startswith("cl_")
|
||
H = {"X-API-Key": k["key"]}
|
||
body = {"inventory_no": "П-ИНВ-1", "title": "Перегрев двигателя", "external_id": "SCADA-77"}
|
||
assert c.post(f"{V1}/external/incidents", json=body).status_code == 401
|
||
assert c.post(f"{V1}/external/incidents", headers={"X-API-Key": "cl_wrong"}, json=body).status_code == 401
|
||
r1 = c.post(f"{V1}/external/incidents", headers=H, json=body)
|
||
assert r1.status_code == 201 and r1.json()["source"] == "api" and r1.json()["reporter_name"] == "SCADA"
|
||
assert c.post(f"{V1}/external/incidents", headers=H, json=body).json()["id"] == r1.json()["id"] # повтор — без дубликата
|
||
assert c.post(f"{V1}/external/incidents", headers=H, json={"inventory_no": "нет", "title": "x1"}).status_code == 404
|
||
assert any(e["inventory_no"] == "П-ИНВ-1" for e in c.get(f"{V1}/external/equipment", headers=H).json())
|
||
assert c.get(f"{V1}/api-keys", headers=adm).json()[0]["last_used_at"]
|
||
c.delete(f"{V1}/api-keys/{k['id']}", headers=adm)
|
||
assert c.post(f"{V1}/external/incidents", headers=H, json=body).status_code == 401
|
||
|
||
|
||
def test_equipment_history(env):
|
||
c, tech, eq = env["c"], env["tech"], env["eq"]
|
||
h = c.get(f"{V1}/equipment/{eq['id']}/history", headers=tech).json()
|
||
kinds = {r["type"] for r in h["rows"]}
|
||
assert {"incident", "service", "meter", "order", "task"} <= kinds
|
||
assert h["rows"] == sorted(h["rows"], key=lambda r: r["at"], reverse=True)
|
||
only = c.get(f"{V1}/equipment/{eq['id']}/history", headers=tech, params={"types": "incident"}).json()
|
||
assert only["rows"] and all(r["type"] == "incident" for r in only["rows"])
|
||
assert c.get(f"{V1}/equipment/{eq['id']}/history", headers=tech, params={"types": ""}).json()["total"] == 0
|
||
srch = c.get(f"{V1}/equipment/{eq['id']}/history", headers=tech, params={"q": "заклинил"}).json()
|
||
assert srch["rows"] and all("аклинил" in r["title"] + r["text"] for r in srch["rows"])
|
||
old = c.get(f"{V1}/equipment/{eq['id']}/history", headers=tech, params={"to": "2000-01-01"}).json()
|
||
assert old["total"] == 0
|
||
assert any("заменён при отказе" in r["title"] for r in h["rows"])
|
||
|
||
|
||
def test_departments(env):
|
||
c, adm, tech, ids = env["c"], env["adm"], env["tech"], env["ids"]
|
||
assert c.post(f"{V1}/departments", headers=tech, json={"name": "Х"}).status_code == 403
|
||
top = c.post(f"{V1}/departments", headers=adm, json={"name": "П-Служба ГИ"}).json()
|
||
mech = c.post(f"{V1}/departments", headers=adm, json={"name": "П-Мехслужба", "parent_id": top["id"], "head_user_id": ids["p_eng"]}).json()
|
||
assert c.patch(f"{V1}/departments/{top['id']}", headers=adm, json={"parent_id": mech["id"]}).status_code == 422 # цикл
|
||
assert c.post(f"{V1}/departments", headers=adm, json={"name": "П-Мехслужба"}).status_code == 409 # дубль
|
||
assert c.patch(f"{V1}/users/{ids['p_tech']}", headers=adm, json={"department_id": 99999}).status_code == 422
|
||
u = c.patch(f"{V1}/users/{ids['p_tech']}", headers=adm, json={"department_id": mech["id"]}).json()
|
||
assert u["department"]["name"] == "П-Мехслужба"
|
||
deps = {d["name"]: d for d in c.get(f"{V1}/departments", headers=tech).json()}
|
||
assert deps["П-Мехслужба"]["members"] == 1 and deps["П-Мехслужба"]["parent_id"] == top["id"]
|
||
assert c.get(f"{V1}/users/{ids['p_tech']}/card", headers=tech).json()["user"]["department"]["id"] == mech["id"]
|
||
assert c.delete(f"{V1}/departments/{mech['id']}", headers=adm).status_code == 204
|
||
assert c.get(f"{V1}/me", headers=tech).json()["department_id"] is None
|