92 lines
6.4 KiB
Python
92 lines
6.4 KiB
Python
"""Анализ надёжности: запись об отказе (форма закрытия), показатели, дрейф параметров, выгрузки."""
|
|
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(), "rl.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"
|
|
PW = "Pass1234567"
|
|
|
|
|
|
def test_failure_record_and_reliability():
|
|
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": "rl_t", "full_name": "Дежурный", "password": PW})
|
|
t = login("rl_t")
|
|
now = config.now()
|
|
eq = c.post(f"{V1}/equipment", headers=adm, json={"name": "RL-Насос"}).json()
|
|
quiet = c.post(f"{V1}/equipment", headers=adm, json={"name": "RL-Тихий"}).json()
|
|
su = c.post(f"{V1}/equipment/{eq['id']}/service-units", headers=adm, json={"name": "RL-Подшипник", "interval_hours": 5000}).json()
|
|
for e, pts in ((eq, ((100, 0), (50, 1000), (1, 2000))), (quiet, ((100, 0), (1, 2000)))):
|
|
for d, h in pts:
|
|
assert c.post(f"{V1}/equipment/{e['id']}/meter", headers=adm,
|
|
json={"hours": h, "at": (now - timedelta(days=d)).isoformat()}).status_code == 201
|
|
|
|
# отказ: снимок наработки при регистрации
|
|
f1 = c.post(f"{V1}/incidents", headers=t, json={"equipment_id": eq["id"], "title": "RL-заклинило",
|
|
"occurred_at": (now - timedelta(hours=10)).isoformat()}).json()
|
|
assert f1["hours_at_open"] == 2000
|
|
# проверка хронологии: ввод раньше начала работ
|
|
bad = c.post(f"{V1}/incidents/{f1['id']}/close", headers=t, json={
|
|
"repair_started_at": (now - timedelta(hours=5)).isoformat(), "restored_at": (now - timedelta(hours=6)).isoformat()})
|
|
assert bad.status_code == 422
|
|
r = c.post(f"{V1}/incidents/{f1['id']}/close", headers=t, json={
|
|
"note": "заменили подшипник", "repair_started_at": (now - timedelta(hours=8)).isoformat(),
|
|
"restored_at": (now - timedelta(hours=4)).isoformat(), "parts_wait_hours": 1.5, "cause_unit_id": su["id"],
|
|
"action": "replace", "nature": "sudden", "cause": "wear", "precursor": "none", "replaced_units": [su["id"]],
|
|
"meter_at_restore": 2001}).json()
|
|
assert r["status"] == "closed" and r["hours_at_failure"] == 2000 and r["cause_unit_name"] == "RL-Подшипник"
|
|
assert r["units"][0]["replaced"] is True and r["meter_at_restore"] == 2001
|
|
units = c.get(f"{V1}/equipment/{eq['id']}/service-units", headers=adm).json()
|
|
assert units[0]["last_service_hours"] == 2001 # ресурс узла сброшен при вводе
|
|
|
|
# второй отказ закрыт старым клиентом (только комментарий) — поля подставляются
|
|
f2 = c.post(f"{V1}/incidents", headers=t, json={"equipment_id": eq["id"], "title": "RL-сбой",
|
|
"occurred_at": (now - timedelta(days=30)).isoformat()}).json()
|
|
r2 = c.post(f"{V1}/incidents/{f2['id']}/close", headers=t, json={"note": "перезапустили"}).json()
|
|
assert r2["restored_at"] and r2["repair_started_at"] and r2["cause"] == "unknown"
|
|
# инженер уточняет закрытую запись; техник — нет
|
|
assert c.patch(f"{V1}/incidents/{f2['id']}/failure", headers=t, json={"action": "restart"}).status_code == 403
|
|
e2 = c.patch(f"{V1}/incidents/{f2['id']}/failure", headers=adm, json={
|
|
"action": "restart", "restored_at": (now - timedelta(days=30) + timedelta(hours=2)).isoformat()}).json()
|
|
assert e2["action"] == "restart"
|
|
|
|
# неисправность → отказ: признак «было сообщение о неисправности»
|
|
fl = c.post(f"{V1}/incidents", headers=t, json={"kind": "fault", "equipment_id": quiet["id"], "title": "RL-шум"}).json()
|
|
x = c.post(f"{V1}/incidents/{fl['id']}/escalate", headers=t, json={"note": "встал"}).json()
|
|
assert x["precursor"] == "fault_report" and x["escalated_at"] and x["hours_at_failure"] == 2000
|
|
|
|
# показатели
|
|
assert c.get(f"{V1}/reliability", headers=t).status_code == 403
|
|
q = {"from": (now - timedelta(days=100)).isoformat(), "to": (now + timedelta(hours=1)).isoformat()}
|
|
rep = c.get(f"{V1}/reliability", headers=adm, params=q).json()
|
|
rows = {x["equipment"]: x for x in rep["rows"]}
|
|
n = rows["RL-Насос"]
|
|
assert n["failures"] == 2 and n["basis"] == "meter" and abs(n["hours"] - 2001) < 1
|
|
assert abs(n["theta"] - 1000.5) < 1 and n["theta_lower"] < n["theta"] and n["mttr"] is not None and 0 < n["availability"] < 1
|
|
assert rows["RL-Тихий"]["failures"] == 1 and rows["RL-Тихий"]["escalated"] == 1
|
|
assert {c_["key"] for c_ in rep["causes"]} >= {"wear"}
|
|
assert c.get(f"{V1}/me", headers=adm).json()["reliability"] is True
|
|
assert c.get(f"{V1}/me", headers=t).json()["reliability"] is False
|
|
|
|
# выгрузки
|
|
csv1 = c.get(f"{V1}/reliability/export", headers=adm, params=q).content.decode("utf-8-sig")
|
|
assert "RL-заклинило" in csv1 and "Износ и старение" in csv1 and "Замена узла" in csv1
|
|
csv2 = c.get(f"{V1}/reliability/export", headers=adm, params={**q, "kind": "service"}).content.decode("utf-8-sig")
|
|
assert "RL-Подшипник" in csv2 and "замена при отказе" in csv2
|
|
|
|
# выключение раздела
|
|
c.put(f"{V1}/settings", headers=adm, json={"reliability": False})
|
|
assert c.get(f"{V1}/reliability", headers=adm).status_code == 404
|
|
c.put(f"{V1}/settings", headers=adm, json={"reliability": True})
|