110 lines
5.1 KiB
Python
110 lines
5.1 KiB
Python
"""GET /api/production-runs/open - live monitor of the open runs (#57)."""
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy import select
|
|
|
|
from src.backend.models.orm.measurement import Measurement
|
|
from src.backend.models.orm.production import ProductionEvent, ProductionRun
|
|
from src.backend.models.orm.station import Station
|
|
from src.backend.models.orm.task import RecipeSubtask, RecipeTask
|
|
from src.backend.tests.conftest import _create_user, auth_headers, create_test_recipe
|
|
|
|
|
|
async def _run(db, user, code, status="running", overdue=False, open_=True):
|
|
"""A station plus a run on a fresh recipe (limits ltl 9.5 / utl 10.5)."""
|
|
station = Station(code=f"ST-{code}", name=f"Linea {code}", active=True, created_by=user.id)
|
|
db.add(station)
|
|
recipe = await create_test_recipe(db, user_id=user.id, code=f"REC-{code}")
|
|
await db.flush()
|
|
version = recipe.versions[0]
|
|
now = datetime.utcnow()
|
|
run = ProductionRun(
|
|
station_id=station.id, recipe_id=recipe.id, version_id=version.id,
|
|
operator_id=user.id, status=status if open_ else "closed",
|
|
started_at=now - timedelta(hours=1),
|
|
measurement_interval_minutes=30,
|
|
next_measurement_at=now + (timedelta(minutes=-5) if overdue else timedelta(minutes=20)),
|
|
cycle_count=3,
|
|
active_station_id=station.id if open_ else None,
|
|
)
|
|
db.add(run)
|
|
await db.flush()
|
|
db.add(ProductionEvent(run_id=run.id, event_type="start", user_id=user.id))
|
|
sub = (await db.execute(
|
|
select(RecipeSubtask).join(RecipeTask, RecipeTask.id == RecipeSubtask.task_id)
|
|
.where(RecipeTask.version_id == version.id)
|
|
)).scalars().first()
|
|
return run, sub, version
|
|
|
|
|
|
def _m(run, sub, version, user, value, pf, minutes=0):
|
|
return Measurement(
|
|
subtask_id=sub.id, version_id=version.id, measured_by=user.id, value=value,
|
|
pass_fail=pf, production_run_id=run.id,
|
|
measured_at=datetime.utcnow() - timedelta(minutes=minutes),
|
|
)
|
|
|
|
|
|
async def test_open_requires_high_authority(client, db_session, measurement_tec_user):
|
|
resp = await client.get("/api/production-runs/open", headers=auth_headers(measurement_tec_user))
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_open_lists_only_open_runs_with_means(client, db_session, admin_user):
|
|
sup = await _create_user(db_session, "capo", roles=["Supervisor"])
|
|
r1, s1, v1 = await _run(db_session, admin_user, "A", overdue=True)
|
|
r2, s2, v2 = await _run(db_session, admin_user, "B", status="paused")
|
|
await _run(db_session, admin_user, "C", open_=False)
|
|
# Run A: mean 10.0 (inside), run B: mean 10.8 (outside, one fail).
|
|
for i, (val, pf) in enumerate([(9.9, "pass"), (10.1, "pass")]):
|
|
db_session.add(_m(r1, s1, v1, admin_user, val, pf, i))
|
|
for i, (val, pf) in enumerate([(10.6, "fail"), (11.0, "fail"), (10.8, "warning")]):
|
|
db_session.add(_m(r2, s2, v2, admin_user, val, pf, i))
|
|
await db_session.commit()
|
|
|
|
resp = await client.get("/api/production-runs/open", headers=auth_headers(sup))
|
|
assert resp.status_code == 200
|
|
body = {r["station_code"]: r for r in resp.json()}
|
|
assert set(body) == {"ST-A", "ST-B"} # closed run C is absent
|
|
a, b = body["ST-A"], body["ST-B"]
|
|
assert a["overdue"] is True and a["seconds_to_next_measurement"] < 0
|
|
assert b["overdue"] is False and b["status"] == "paused"
|
|
assert a["duration_seconds"] >= 3600 and a["cycle_count"] == 3
|
|
assert a["last_event"]["event_type"] == "start"
|
|
assert a["pass_count"] == 2 and a["fail_count"] == 0
|
|
assert a["subtasks"][0]["mean"] == 10.0 and a["subtasks"][0]["in_tolerance"] is True
|
|
assert b["fail_count"] == 2 and b["warning_count"] == 1
|
|
assert round(b["subtasks"][0]["mean"], 3) == 10.8
|
|
assert b["subtasks"][0]["in_tolerance"] is False
|
|
|
|
|
|
async def test_open_empty_for_admin(client, admin_user):
|
|
resp = await client.get("/api/production-runs/open", headers=auth_headers(admin_user))
|
|
assert resp.status_code == 200 and resp.json() == []
|
|
|
|
|
|
async def test_nan_quotes_in_the_monitor(client, db_session, admin_user):
|
|
"""#118: solo NaN -> la quota resta, mean null e fuori tolleranza; mista -> media dei numeri."""
|
|
sup = await _create_user(db_session, "capo", roles=["Supervisor"])
|
|
r1, s1, v1 = await _run(db_session, admin_user, "N1")
|
|
r2, s2, v2 = await _run(db_session, admin_user, "N2")
|
|
for i in range(2):
|
|
db_session.add(_m(r1, s1, v1, admin_user, None, "fail", i))
|
|
for i, (val, pf) in enumerate([(None, "fail"), (10.0, "pass"), (10.2, "pass")]):
|
|
db_session.add(_m(r2, s2, v2, admin_user, val, pf, i))
|
|
await db_session.commit()
|
|
|
|
resp = await client.get("/api/production-runs/open", headers=auth_headers(sup))
|
|
assert resp.status_code == 200
|
|
body = {r["station_code"]: r for r in resp.json()}
|
|
only_nan = body["ST-N1"]["subtasks"]
|
|
assert len(only_nan) == 1
|
|
assert only_nan[0]["mean"] is None
|
|
assert only_nan[0]["count"] == 2
|
|
assert only_nan[0]["in_tolerance"] is False
|
|
assert body["ST-N1"]["fail_count"] == 2
|
|
mixed = body["ST-N2"]["subtasks"][0]
|
|
assert mixed["count"] == 3
|
|
assert round(mixed["mean"], 6) == 10.1
|
|
assert mixed["in_tolerance"] is True
|