57 lines
4.2 KiB
Python
57 lines
4.2 KiB
Python
"""Сообщения о неисправности: машина работает, состояние «Неисправность», перевод в инцидент, публичная форма и внешний API."""
|
|
import os
|
|
import tempfile
|
|
|
|
os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "fl.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.main import app # noqa: E402
|
|
|
|
V1 = "/api/v1"
|
|
PW = "Pass1234567"
|
|
|
|
|
|
def test_fault_reports():
|
|
with TestClient(app) as c:
|
|
login = lambda u, p=PW: {"Authorization": "Bearer " + c.post(f"{V1}/auth/login", json={"username": u, "password": p}).json()["access_token"]}
|
|
adm = login("admin", "admin")
|
|
c.post(f"{V1}/users", headers=adm, json={"username": "fl_t", "full_name": "Техник Неисправностей", "password": PW})
|
|
t = login("fl_t")
|
|
loc = c.post(f"{V1}/locations", headers=adm, json={"name": "FL-Цех"}).json()
|
|
eq = c.post(f"{V1}/equipment", headers=adm, json={"name": "FL-Насос", "location_id": loc["id"]}).json()
|
|
su = c.post(f"{V1}/equipment/{eq['id']}/service-units", headers=adm, json={"name": "Подшипник", "interval_hours": 1000}).json()
|
|
|
|
# сообщение о неисправности: состояние «fault», замены нет (ресурс узла не сбрасывается)
|
|
f = c.post(f"{V1}/incidents", headers=t, json={"kind": "fault", "equipment_id": eq["id"], "title": "FL-шум",
|
|
"units": [{"unit_id": su["id"], "replaced": True}], "replaced": True}).json()
|
|
assert f["kind"] == "fault" and f["replaced"] is False and f["units"][0]["replaced"] is False
|
|
h = c.get(f"{V1}/equipment-health", headers=t).json()[str(eq["id"])]
|
|
assert h["state"] == "fault" and h["open_faults"] == 1 and h["open_incidents"] == 0
|
|
assert [i["title"] for i in c.get(f"{V1}/incidents", headers=t, params={"kind": "fault", "q": "fl-"}).json()] == ["FL-шум"]
|
|
assert c.get(f"{V1}/incidents", headers=t, params={"kind": "incident", "q": "fl-"}).json() == []
|
|
hist = c.get(f"{V1}/equipment/{eq['id']}/history", headers=t).json()["rows"]
|
|
assert any(r["title"] == "Неисправность: FL-шум" for r in hist)
|
|
|
|
# ухудшилось — перевод в инцидент: машина в «Аварии»
|
|
e = c.post(f"{V1}/incidents/{f['id']}/escalate", headers=t, json={"note": "остановили"}).json()
|
|
assert e["kind"] == "incident" and "остановили" in e["description"]
|
|
assert c.get(f"{V1}/equipment-health", headers=t).json()[str(eq["id"])]["state"] == "alarm"
|
|
assert c.post(f"{V1}/incidents/{f['id']}/escalate", headers=t, json={}).status_code == 409
|
|
c.post(f"{V1}/incidents/{f['id']}/close", headers=t, json={"note": "ok"})
|
|
assert c.get(f"{V1}/equipment-health", headers=t).json()[str(eq["id"])]["state"] in ("ok", "due", "soon")
|
|
|
|
# публичная форма и внешний API принимают kind
|
|
r = c.post(f"{V1}/public/incidents", json={"location_id": loc["id"], "equipment_id": eq["id"], "kind": "fault",
|
|
"description": "Подтекает сальник", "reporter_name": "Подрядчик", "reporter_contact": "+7 900 000"})
|
|
assert r.status_code == 201
|
|
pub = c.get(f"{V1}/incidents", headers=adm, params={"source": "public", "kind": "fault"}).json()
|
|
assert pub and pub[0]["kind"] == "fault"
|
|
key = c.post(f"{V1}/api-keys", headers=adm, json={"name": "FL-SCADA"}).json()["key"]
|
|
x = c.post(f"{V1}/external/incidents", headers={"X-API-Key": key}, json={"equipment_id": eq["id"], "kind": "fault", "title": "FL-вибрация"})
|
|
assert x.status_code == 201 and x.json()["kind"] == "fault"
|
|
assert c.post(f"{V1}/incidents", headers=t, json={"kind": "nonsense", "equipment_id": eq["id"], "title": "x1"}).status_code == 422
|