362 lines
24 KiB
Python
362 lines
24 KiB
Python
"""Регламентное обслуживание: узлы машин, наработка, сброс, регламентные наряды, отчёт «Текущее состояние»."""
|
|
import os
|
|
import tempfile
|
|
from datetime import date, datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("DATABASE_URL", os.environ.get("TEST_DATABASE_URL") or "sqlite:///" + os.path.join(tempfile.mkdtemp(), "svc.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, service # noqa: E402
|
|
from app.main import app # noqa: E402
|
|
from app.models import ServiceUnit # noqa: E402
|
|
|
|
V1 = "/api/v1"
|
|
|
|
|
|
def bearer(t):
|
|
return {"Authorization": "Bearer " + t}
|
|
|
|
|
|
# ---------------------------------------------------------------- расчёт состояния (без БД)
|
|
def unit(**kw):
|
|
base = dict(mode="hours", interval_hours=500, interval_value=None, interval_unit=None, warn_percent=10,
|
|
last_service_at=datetime(2026, 1, 1), last_service_hours=1000, created_at=datetime(2026, 1, 1))
|
|
base.update(kw)
|
|
return ServiceUnit(**base)
|
|
|
|
|
|
NOW = datetime(2026, 6, 1, 12, 0)
|
|
|
|
|
|
def test_hours_rule_states():
|
|
u = unit()
|
|
assert service.evaluate(u, 1100, now=NOW)["state"] == "ok"
|
|
e = service.evaluate(u, 1100, now=NOW)
|
|
assert e["used_hours"] == 100 and e["remaining_hours"] == 400 and e["percent"] == 20
|
|
assert service.evaluate(u, 1460, now=NOW)["state"] == "soon" # осталось 40 из 500 (8% ≤ 10%)
|
|
assert service.evaluate(u, 1450, now=NOW)["state"] == "soon" # ровно 10%
|
|
assert service.evaluate(u, 1449, now=NOW)["state"] == "ok"
|
|
assert service.evaluate(u, 1500, now=NOW)["state"] == "due" # ресурс выработан
|
|
over = service.evaluate(u, 1620, now=NOW)
|
|
assert over["state"] == "due" and over["remaining_hours"] == -120 and over["percent"] == 124
|
|
|
|
|
|
def test_calendar_rule_states_and_months():
|
|
u = unit(mode="calendar", interval_hours=None, interval_value=1, interval_unit="month", last_service_at=datetime(2026, 5, 1))
|
|
e = service.evaluate(u, 0, now=NOW)
|
|
assert e["due_date"] == "2026-06-01" and e["state"] == "due" and e["remaining_days"] == 0
|
|
assert service.evaluate(u, 0, now=datetime(2026, 5, 10))["state"] == "ok"
|
|
assert service.evaluate(u, 0, now=datetime(2026, 5, 29))["state"] == "soon" # осталось 3 из 31 дня
|
|
# конец месяца: 31 января + 1 месяц = 28 февраля
|
|
assert service.add_months(date(2026, 1, 31), 1) == date(2026, 2, 28)
|
|
assert service.add_interval(date(2026, 1, 1), 2, "week") == date(2026, 1, 15)
|
|
assert service.rule_text(u) == "раз в месяц"
|
|
assert service.rule_text(unit(mode="calendar", interval_value=3, interval_unit="month")) == "раз в 3 месяца"
|
|
assert service.rule_text(unit(mode="both", interval_value=6, interval_unit="month")) == "каждые 500 м/ч или раз в 6 месяцев"
|
|
|
|
|
|
def test_both_rule_takes_the_worse_and_forecast():
|
|
u = unit(mode="both", interval_value=12, interval_unit="month", last_service_at=datetime(2026, 5, 1))
|
|
e = service.evaluate(u, 1480, rate=10, now=NOW) # по часам осталось 20 (4%) — «скоро», по календарю далеко
|
|
assert e["state"] == "soon" and e["percent"] == 96
|
|
assert e["forecast_date"] == (NOW.date() + timedelta(days=2)).isoformat() # 20 м/ч при 10 м/ч в сутки
|
|
assert service.evaluate(u, 1600, now=NOW)["state"] == "due"
|
|
assert service.evaluate(unit(mode="hours", interval_hours=None), 5, now=NOW)["state"] == "nodata"
|
|
|
|
|
|
# ---------------------------------------------------------------- API
|
|
@pytest.fixture(scope="module")
|
|
def env():
|
|
with TestClient(app) as c:
|
|
def login(u, p):
|
|
r = c.post(f"{V1}/auth/login", json={"username": u, "password": p})
|
|
assert r.status_code == 200, r.text
|
|
return bearer(r.json()["access_token"])
|
|
|
|
adm = login("admin", "admin")
|
|
always = c.post(f"{V1}/schedules", headers=adm, json={"name": "svc-always", "kind": "weekly",
|
|
"weekdays": [0, 1, 2, 3, 4, 5, 6], "start_time": "00:00:00", "end_time": "23:59:00"}).json()
|
|
tech = c.post(f"{V1}/positions", headers=adm, json={"name": "svc-tech", "kind": "technician", "schedule_id": always["id"]}).json()
|
|
eng = c.post(f"{V1}/positions", headers=adm, json={"name": "svc-eng", "kind": "engineer"}).json()
|
|
for name, pos in (("svc_t", tech), ("svc_e", eng)):
|
|
assert c.post(f"{V1}/users", headers=adm, json={"username": name, "password": "Passw0rd!x", "position_id": pos["id"]}).status_code == 201
|
|
h = {"admin": adm, "t": login("svc_t", "Passw0rd!x"), "e": login("svc_e", "Passw0rd!x")}
|
|
users = {u["username"]: u["id"] for u in c.get(f"{V1}/users", headers=adm).json()}
|
|
loc = c.post(f"{V1}/locations", headers=h["e"], json={"name": "svc-цех"}).json()
|
|
yield {"c": c, "h": h, "users": users, "loc": loc, "tech": tech, "eng": eng}
|
|
|
|
|
|
def new_machine(env, name="Компрессор", hours=1000):
|
|
c, h = env["c"], env["h"]
|
|
eq = c.post(f"{V1}/equipment", headers=h["e"], json={"name": name, "location_id": env["loc"]["id"], "inventory_no": "K-1"}).json()
|
|
if hours:
|
|
r = c.post(f"{V1}/equipment/{eq['id']}/meter", headers=h["e"], json={"hours": hours})
|
|
assert r.status_code == 201, r.text
|
|
return eq["id"]
|
|
|
|
|
|
def test_unit_crud_validation_and_rbac(env):
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env)
|
|
url = f"{V1}/equipment/{eid}/service-units"
|
|
# правило должно быть задано
|
|
assert c.post(url, headers=h["e"], json={"name": "Масло", "mode": "hours"}).status_code == 422
|
|
assert c.post(url, headers=h["e"], json={"name": "Масло", "mode": "calendar", "interval_value": 1}).status_code == 422
|
|
assert c.post(url, headers=h["e"], json={"name": "Масло", "mode": "hours", "interval_hours": 0}).status_code == 422
|
|
# техник не может менять, но может читать
|
|
body = {"name": "Масло", "mode": "hours", "interval_hours": 500}
|
|
assert c.post(url, headers=h["t"], json=body).status_code == 403
|
|
u = c.post(url, headers=h["e"], json=body)
|
|
assert u.status_code == 201, u.text
|
|
u = u.json()
|
|
assert u["state"] == "ok" and u["last_service_hours"] == 1000 and u["remaining_hours"] == 500 # старт: «обслужено сейчас»
|
|
assert c.get(url, headers=h["t"]).status_code == 200
|
|
assert c.patch(f"{V1}/service-units/{u['id']}", headers=h["t"], json={"name": "x"}).status_code == 403
|
|
# правка: добавляем календарное правило
|
|
p = c.patch(f"{V1}/service-units/{u['id']}", headers=h["e"], json={"mode": "both", "interval_value": 6, "interval_unit": "month"}).json()
|
|
assert p["rule"] == "каждые 500 м/ч или раз в 6 месяцев" and p["due_date"]
|
|
assert c.patch(f"{V1}/service-units/{u['id']}", headers=h["e"], json={"mode": "calendar", "interval_unit": None}).status_code == 422
|
|
# начальное состояние не может быть «из будущего» по наработке
|
|
r = c.post(url, headers=h["e"], json={"name": "Фильтр", "mode": "hours", "interval_hours": 200, "last_service_hours": 5000})
|
|
assert r.status_code == 422
|
|
# удаление
|
|
assert c.delete(f"{V1}/service-units/{u['id']}", headers=h["e"]).status_code == 204
|
|
assert c.get(url, headers=h["e"]).json() == []
|
|
|
|
|
|
def test_meter_rules(env):
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env, "Насос-м", 100)
|
|
m = f"{V1}/equipment/{eid}/meter"
|
|
assert c.post(m, headers=h["e"], json={"hours": 90}).status_code == 422 # назад нельзя…
|
|
assert c.post(m, headers=h["e"], json={"hours": 5, "correction": True, "note": "замена счётчика"}).status_code == 201 # …кроме коррекции
|
|
assert c.get(f"{V1}/equipment/{eid}", headers=h["e"]).json()["meter_hours"] == 5
|
|
assert c.post(m, headers=h["e"], json={"hours": 10, "at": (config.now() + timedelta(days=1)).isoformat()}).status_code == 422
|
|
assert c.post(m, headers=h["t"], json={"hours": 50}).status_code == 403
|
|
hist = c.get(m, headers=h["t"]).json()
|
|
assert [x["source"] for x in hist][:2] == ["correction", "manual"] and hist[0]["note"] == "замена счётчика"
|
|
|
|
|
|
def test_manual_reset_flow(env):
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env, "Дизель", 2000)
|
|
u = c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json={"name": "Масло", "mode": "hours", "interval_hours": 250}).json()
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 2260})
|
|
u = c.get(f"{V1}/equipment/{eid}/service-units", headers=h["e"]).json()[0]
|
|
assert u["state"] == "due" and u["remaining_hours"] == -10
|
|
# сброс: масло заменили отдельно, на 2255 м/ч
|
|
r = c.post(f"{V1}/service-units/{u['id']}/reset", headers=h["e"], json={"hours": 2255, "note": "Замена вне наряда"})
|
|
assert r.status_code == 200, r.text
|
|
r = r.json()
|
|
assert r["state"] == "ok" and r["last_service_hours"] == 2255 and r["remaining_hours"] == 245
|
|
# без параметров — «обслужено сейчас, при текущей наработке»
|
|
r = c.post(f"{V1}/service-units/{u['id']}/reset", headers=h["e"], json={}).json()
|
|
assert r["last_service_hours"] == 2260 and r["remaining_hours"] == 250
|
|
# валидация и права
|
|
assert c.post(f"{V1}/service-units/{u['id']}/reset", headers=h["e"], json={"hours": 99999}).status_code == 422
|
|
assert c.post(f"{V1}/service-units/{u['id']}/reset", headers=h["e"], json={"at": (config.now() + timedelta(days=2)).isoformat()}).status_code == 422
|
|
assert c.post(f"{V1}/service-units/{u['id']}/reset", headers=h["t"], json={}).status_code == 403
|
|
ev = c.get(f"{V1}/service-units/{u['id']}/events", headers=h["t"]).json()
|
|
assert [e["kind"] for e in ev] == ["reset", "reset", "initial"] and ev[1]["note"] == "Замена вне наряда" and ev[0]["user_name"]
|
|
|
|
|
|
def make_regular(env, only_due=False, items=()):
|
|
c, h = env["c"], env["h"]
|
|
return c.post(f"{V1}/checklists", headers=h["e"], json={"name": f"Регламент {datetime.now().timestamp()}", "kind": "regular",
|
|
"only_due": only_due, "items": list(items)}).json()
|
|
|
|
|
|
def test_regular_order_lifecycle(env):
|
|
"""Регламентный наряд: пункты узлов подставляются из машины; по завершении узлы сбрасываются, наработка обновляется."""
|
|
c, h, users = env["c"], env["h"], env["users"]
|
|
eid = new_machine(env, "Погрузчик", 3000)
|
|
mk = lambda body: c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json=body).json()
|
|
oil = mk({"name": "Масло двигателя", "mode": "hours", "interval_hours": 500})
|
|
flt = mk({"name": "Воздушный фильтр", "mode": "hours", "interval_hours": 1000})
|
|
belt = mk({"name": "Ремень", "mode": "calendar", "interval_value": 3, "interval_unit": "month"})
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 3480}) # у масла осталось 20 м/ч
|
|
cl = make_regular(env, items=[{"text": "Осмотр на течи", "answer_type": "status"}])
|
|
assert cl["kind"] == "regular"
|
|
|
|
# без оборудования регламентный наряд не создать
|
|
r = c.post(f"{V1}/orders", headers=h["e"], json={"title": "Рег.", "checklist_id": cl["id"], "assignee_user_id": users["svc_t"]})
|
|
assert r.status_code == 422
|
|
o = c.post(f"{V1}/orders", headers=h["e"], json={"title": "ТО погрузчика", "checklist_id": cl["id"], "equipment_id": eid,
|
|
"assignee_user_id": users["svc_t"]})
|
|
assert o.status_code == 201, o.text
|
|
o = o.json()
|
|
items = o["items"]
|
|
assert items[0]["is_meter"] and items[0]["answer_type"] == "number" and items[0]["unit"] == "м/ч" and items[0]["min_value"] == 3480
|
|
assert items[1]["text"] == "Осмотр на течи" and not items[1]["service_unit_id"]
|
|
by_unit = {i["service_unit_id"]: i for i in items if i["service_unit_id"]}
|
|
assert set(by_unit) == {oil["id"], flt["id"], belt["id"]}
|
|
assert "осталось 20 м/ч" in by_unit[oil["id"]]["text"] and by_unit[oil["id"]]["answer_type"] == "status"
|
|
|
|
# техник: наработка + «обслужено» для масла и осмотра; фильтр и ремень не трогаем
|
|
t = h["t"]
|
|
put = lambda iid, v: c.put(f"{V1}/orders/{o['id']}/items/{iid}", headers=t, json={"value": v})
|
|
assert put(items[0]["id"], "3490").status_code == 200
|
|
assert put(items[1]["id"], "done").status_code == 200
|
|
assert put(by_unit[oil["id"]]["id"], "done").status_code == 200
|
|
# до завершения ничего не сброшено
|
|
assert c.get(f"{V1}/equipment/{eid}", headers=h["e"]).json()["meter_hours"] == 3480
|
|
assert c.post(f"{V1}/orders/{o['id']}/submit", headers=t).json()["status"] == "completed" # проверяющего нет
|
|
|
|
assert c.get(f"{V1}/equipment/{eid}", headers=h["e"]).json()["meter_hours"] == 3490
|
|
st = {u["name"]: u for u in c.get(f"{V1}/equipment/{eid}/service-units", headers=h["e"]).json()}
|
|
assert st["Масло двигателя"]["last_service_hours"] == 3490 and st["Масло двигателя"]["remaining_hours"] == 500
|
|
assert st["Воздушный фильтр"]["last_service_hours"] == 3000 # не отмечен — не сброшен
|
|
ev = c.get(f"{V1}/service-units/{oil['id']}/events", headers=h["e"]).json()
|
|
assert ev[0]["kind"] == "order" and ev[0]["order_id"] == o["id"]
|
|
# повторная обработка того же наряда ничего не дублирует
|
|
from app.db import SessionLocal
|
|
from app.models import Order, ServiceEvent
|
|
with SessionLocal() as s:
|
|
service.apply_completed_order(s, s.get(Order, o["id"]))
|
|
s.commit()
|
|
assert s.query(ServiceEvent).filter(ServiceEvent.unit_id == oil["id"], ServiceEvent.order_id == o["id"]).count() == 1
|
|
|
|
|
|
def test_regular_order_with_review_applies_only_on_approval(env):
|
|
c, h, users = env["c"], env["h"], env["users"]
|
|
eid = new_machine(env, "Генератор", 100)
|
|
u = c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json={"name": "Свечи", "mode": "hours", "interval_hours": 100}).json()
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 195})
|
|
cl = make_regular(env)
|
|
o = c.post(f"{V1}/orders", headers=h["e"], json={"title": "ТО ген.", "checklist_id": cl["id"], "equipment_id": eid,
|
|
"assignee_user_id": users["svc_t"], "reviewer_user_id": users["svc_e"]}).json()
|
|
unit_item = next(i for i in o["items"] if i["service_unit_id"])
|
|
c.put(f"{V1}/orders/{o['id']}/items/{o['items'][0]['id']}", headers=h["t"], json={"value": "200"})
|
|
c.put(f"{V1}/orders/{o['id']}/items/{unit_item['id']}", headers=h["t"], json={"value": "done"})
|
|
assert c.post(f"{V1}/orders/{o['id']}/submit", headers=h["t"]).json()["status"] == "review"
|
|
assert c.get(f"{V1}/equipment/{eid}/service-units", headers=h["e"]).json()[0]["last_service_hours"] == 100 # ещё не сброшено
|
|
assert c.post(f"{V1}/orders/{o['id']}/review", headers=h["e"], json={"approve": True, "comment": ""}).json()["status"] == "completed"
|
|
r = c.get(f"{V1}/equipment/{eid}/service-units", headers=h["e"]).json()[0]
|
|
assert r["last_service_hours"] == 200 and r["state"] == "ok"
|
|
# возврат на доработку ничего не сбрасывает
|
|
eid2 = new_machine(env, "Генератор-2", 10)
|
|
c.post(f"{V1}/equipment/{eid2}/service-units", headers=h["e"], json={"name": "Свечи", "mode": "hours", "interval_hours": 100})
|
|
o2 = c.post(f"{V1}/orders", headers=h["e"], json={"title": "ТО-2", "checklist_id": cl["id"], "equipment_id": eid2,
|
|
"assignee_user_id": users["svc_t"], "reviewer_user_id": users["svc_e"]}).json()
|
|
for it in o2["items"]:
|
|
c.put(f"{V1}/orders/{o2['id']}/items/{it['id']}", headers=h["t"], json={"value": "50" if it["is_meter"] else "done"})
|
|
c.post(f"{V1}/orders/{o2['id']}/submit", headers=h["t"])
|
|
c.post(f"{V1}/orders/{o2['id']}/review", headers=h["e"], json={"approve": False, "comment": "переделать"})
|
|
assert c.get(f"{V1}/equipment/{eid2}/service-units", headers=h["e"]).json()[0]["last_service_hours"] == 10
|
|
|
|
|
|
def test_only_due_filter_and_empty_cases(env):
|
|
c, h, users = env["c"], env["h"], env["users"]
|
|
eid = new_machine(env, "Пресс", 500)
|
|
c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json={"name": "Гидравлика", "mode": "hours", "interval_hours": 1000})
|
|
cl = make_regular(env, only_due=True)
|
|
body = {"title": "ТО пресса", "checklist_id": cl["id"], "equipment_id": eid, "assignee_user_id": users["svc_t"]}
|
|
r = c.post(f"{V1}/orders", headers=h["e"], json=body)
|
|
assert r.status_code == 422 # узлы в норме и своих пунктов нет — наряд не нужен
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 1450})
|
|
assert c.post(f"{V1}/orders", headers=h["e"], json=body).status_code == 201 # 50 из 1000 (5%) — «скоро»
|
|
# шаблон автозапуска на регламентный чек-лист требует машину
|
|
t = c.post(f"{V1}/order-templates", headers=h["e"], json={"title": "Авто", "checklist_id": cl["id"], "recurrence": "daily",
|
|
"assignee_user_id": users["svc_t"]})
|
|
assert t.status_code == 422
|
|
t = c.post(f"{V1}/order-templates", headers=h["e"], json={"title": "Авто", "checklist_id": cl["id"], "equipment_id": eid,
|
|
"recurrence": "daily", "assignee_user_id": users["svc_t"]})
|
|
assert t.status_code == 201
|
|
|
|
|
|
def test_scheduler_skips_templates_that_cannot_create_orders(env):
|
|
"""Шаблон регламентного ТО у машины без узлов не должен ломать планировщик остальных шаблонов."""
|
|
from app import logic
|
|
from app.db import SessionLocal
|
|
from app.models import OrderTemplate
|
|
c, h, users = env["c"], env["h"], env["users"]
|
|
eid = new_machine(env, "Пустая", 0)
|
|
cl = make_regular(env)
|
|
ok_cl = c.post(f"{V1}/checklists", headers=h["e"], json={"name": "обычный", "items": [{"text": "a", "answer_type": "status"}]}).json()
|
|
with SessionLocal() as s:
|
|
past = config.now() - timedelta(hours=1)
|
|
bad = OrderTemplate(title="плохой", checklist_id=cl["id"], equipment_id=eid, assignee_user_id=users["svc_t"], recurrence="daily", is_active=True, next_run=past,
|
|
start_date=past.date())
|
|
good = OrderTemplate(title="хороший", checklist_id=ok_cl["id"], assignee_user_id=users["svc_t"], recurrence="daily", is_active=True, next_run=past,
|
|
start_date=past.date())
|
|
s.add_all([bad, good])
|
|
s.commit()
|
|
created = logic.run_due_templates(s)
|
|
assert [o.title for o in created] == ["хороший"]
|
|
s.refresh(bad)
|
|
assert bad.next_run > config.now()
|
|
|
|
|
|
def test_state_report(env):
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env, "Отчётная машина", 100)
|
|
mk = lambda body: c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json=body).json()
|
|
mk({"name": "=Масло", "mode": "hours", "interval_hours": 50}) # просрочится
|
|
mk({"name": "Фильтр", "mode": "hours", "interval_hours": 1000}) # в норме
|
|
mk({"name": "Ремень", "mode": "hours", "interval_hours": 100}) # скоро
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 195})
|
|
base = f"{V1}/reports/equipment-state"
|
|
j = c.get(base, headers=h["e"], params={"equipment_id": eid}).json()
|
|
assert [r["unit"] for r in j["rows"]] == ["=Масло", "Ремень", "Фильтр"] # сначала «Требуется», потом «Скоро»
|
|
assert [r["state"] for r in j["rows"]] == ["due", "soon", "ok"] and j["summary"] == {"due": 1, "soon": 1, "ok": 1, "nodata": 0}
|
|
assert j["rows"][0]["remaining_hours"] == -45
|
|
att = c.get(base, headers=h["e"], params={"equipment_id": eid, "state": "attention"}).json()
|
|
assert [r["state"] for r in att["rows"]] == ["due", "soon"]
|
|
assert [r["unit"] for r in c.get(base, headers=h["e"], params={"equipment_id": eid, "state": "ok"}).json()["rows"]] == ["Фильтр"]
|
|
assert c.get(base, headers=h["e"], params={"state": "bad"}).status_code == 422
|
|
assert c.get(base, headers=h["t"]).status_code == 403
|
|
pdf = c.get(base, headers=h["e"], params={"equipment_id": eid, "format": "pdf"})
|
|
assert pdf.status_code == 200 and pdf.content[:4] == b"%PDF"
|
|
csv_ = c.get(base, headers=h["e"], params={"equipment_id": eid, "format": "csv"})
|
|
text = csv_.content.decode("utf-8-sig")
|
|
assert csv_.status_code == 200 and "Состояние" in text and "'=Масло" in text and ";=Масло" not in text # формулы экранированы
|
|
# неактивный узел скрыт, пока не попросят
|
|
u_id = j["rows"][1]["unit_id"]
|
|
c.patch(f"{V1}/service-units/{u_id}", headers=h["e"], json={"is_active": False})
|
|
assert len(c.get(base, headers=h["e"], params={"equipment_id": eid}).json()["rows"]) == 2
|
|
assert len(c.get(base, headers=h["e"], params={"equipment_id": eid, "include_inactive": True}).json()["rows"]) == 3
|
|
|
|
|
|
def test_forecast_uses_meter_history(env):
|
|
from app.db import SessionLocal
|
|
from app.models import MeterReading, Equipment
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env, "Прогноз", 0)
|
|
c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json={"name": "Масло", "mode": "hours", "interval_hours": 400})
|
|
with SessionLocal() as s: # 10 суток назад — 0 м/ч, сейчас — 100 м/ч → 10 м/ч в сутки
|
|
eq = s.get(Equipment, eid)
|
|
s.add(MeterReading(equipment_id=eid, hours=0, at=config.now() - timedelta(days=10), source="manual"))
|
|
s.add(MeterReading(equipment_id=eid, hours=100, at=config.now(), source="manual"))
|
|
eq.meter_hours, eq.meter_updated_at = 100, config.now()
|
|
s.commit()
|
|
row = c.get(f"{V1}/reports/equipment-state", headers=h["e"], params={"equipment_id": eid}).json()["rows"][0]
|
|
assert row["daily_rate"] == 10.0
|
|
assert row["forecast_date"] == (config.now().date() + timedelta(days=30)).isoformat() # осталось 300 м/ч / 10 в сутки
|
|
|
|
|
|
def test_checklist_kind_roundtrip(env):
|
|
c, h = env["c"], env["h"]
|
|
cl = c.post(f"{V1}/checklists", headers=h["e"], json={"name": "рег-1", "kind": "regular", "only_due": True}).json()
|
|
got = c.get(f"{V1}/checklists/{cl['id']}", headers=h["t"]).json()
|
|
assert got["kind"] == "regular" and got["only_due"] is True
|
|
upd = c.put(f"{V1}/checklists/{cl['id']}", headers=h["e"], json={"name": "рег-1", "kind": "inspection", "items": [{"text": "x"}]}).json()
|
|
assert upd["kind"] == "inspection" and upd["only_due"] is False
|
|
assert c.post(f"{V1}/checklists", headers=h["e"], json={"name": "x", "kind": "weird"}).status_code == 422
|
|
|
|
|
|
def test_report_summary_ignores_state_filter(env):
|
|
c, h = env["c"], env["h"]
|
|
eid = new_machine(env, "Сводка", 100)
|
|
for n, hrs_ in (("a", 50), ("b", 1000)):
|
|
c.post(f"{V1}/equipment/{eid}/service-units", headers=h["e"], json={"name": n, "mode": "hours", "interval_hours": hrs_})
|
|
c.post(f"{V1}/equipment/{eid}/meter", headers=h["e"], json={"hours": 160})
|
|
j = c.get(f"{V1}/reports/equipment-state", headers=h["e"], params={"equipment_id": eid, "state": "due"}).json()
|
|
assert [r["unit"] for r in j["rows"]] == ["a"] and j["total"] == 1
|
|
assert j["summary"]["due"] == 1 and j["summary"]["ok"] == 1 # сводка — по всем узлам машины
|