290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""Tests for statistics router (/api/statistics)."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.backend.models.orm.user import User
|
|
from src.backend.models.orm.measurement import Measurement
|
|
from src.backend.tests.conftest import auth_headers, create_test_recipe
|
|
|
|
|
|
async def _seed_measurements(
|
|
client: AsyncClient,
|
|
tec_user: User,
|
|
recipe_id: int,
|
|
values: list[float],
|
|
) -> tuple[int, int]:
|
|
"""Create measurements and return (subtask_id, version_id)."""
|
|
tasks_resp = await client.get(
|
|
f"/api/recipes/{recipe_id}/tasks", headers=auth_headers(tec_user)
|
|
)
|
|
task = tasks_resp.json()[0]
|
|
subtask_id = task["subtasks"][0]["id"]
|
|
version_id = task["version_id"]
|
|
|
|
for val in values:
|
|
await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": val,
|
|
},
|
|
)
|
|
return subtask_id, version_id
|
|
|
|
|
|
class TestSummary:
|
|
"""GET /api/statistics/summary tests."""
|
|
|
|
async def test_get_statistics_by_recipe(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Metrologist can get summary statistics."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
await _seed_measurements(
|
|
client, measurement_tec_user, recipe.id,
|
|
[10.0, 10.1, 10.4, 10.6, 9.4], # pass, pass, warning, fail, fail
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/statistics/summary",
|
|
headers=auth_headers(metrologist_user),
|
|
params={"recipe_id": recipe.id},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 5
|
|
assert data["pass_count"] >= 0
|
|
assert data["warning_count"] >= 0
|
|
assert data["fail_count"] >= 0
|
|
assert data["pass_rate"] >= 0
|
|
# Total rates should sum to ~100
|
|
total_rate = data["pass_rate"] + data["warning_rate"] + data["fail_rate"]
|
|
assert abs(total_rate - 100.0) < 0.1
|
|
|
|
async def test_statistics_require_metrologist(
|
|
self,
|
|
client: AsyncClient,
|
|
maker_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Non-Metrologist cannot access statistics."""
|
|
recipe = await create_test_recipe(db_session, maker_user.id)
|
|
resp = await client.get(
|
|
"/api/statistics/summary",
|
|
headers=auth_headers(maker_user),
|
|
params={"recipe_id": recipe.id},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
async def test_statistics_empty(
|
|
self,
|
|
client: AsyncClient,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Statistics on recipe with no measurements returns zeros."""
|
|
recipe = await create_test_recipe(
|
|
db_session, metrologist_user.id
|
|
)
|
|
resp = await client.get(
|
|
"/api/statistics/summary",
|
|
headers=auth_headers(metrologist_user),
|
|
params={"recipe_id": recipe.id},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 0
|
|
assert data["pass_count"] == 0
|
|
|
|
|
|
class TestCapability:
|
|
"""GET /api/statistics/capability tests."""
|
|
|
|
async def test_capability_indices(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Capability endpoint returns Cp/Cpk/Pp/Ppk indices."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, _ = await _seed_measurements(
|
|
client, measurement_tec_user, recipe.id,
|
|
[9.9, 10.0, 10.1, 9.8, 10.2, 10.0, 9.95, 10.05, 10.1, 9.9],
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/statistics/capability",
|
|
headers=auth_headers(metrologist_user),
|
|
params={
|
|
"recipe_id": recipe.id,
|
|
"subtask_id": subtask_id,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["n"] == 10
|
|
assert data["mean"] is not None
|
|
assert data["std_dev"] is not None
|
|
# With UTL=10.5 and LTL=9.5, spread=1.0, Cp should be > 0
|
|
if data["cp"] is not None:
|
|
assert data["cp"] > 0
|
|
|
|
|
|
class TestSPCData:
|
|
"""GET /api/statistics/control-chart tests."""
|
|
|
|
async def test_spc_data(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Control chart returns values, mean, UCL, LCL."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, _ = await _seed_measurements(
|
|
client, measurement_tec_user, recipe.id,
|
|
[10.0, 10.1, 9.9, 10.05, 9.95],
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/statistics/control-chart",
|
|
headers=auth_headers(metrologist_user),
|
|
params={
|
|
"recipe_id": recipe.id,
|
|
"subtask_id": subtask_id,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["values"]) == 5
|
|
assert data["mean"] is not None
|
|
assert data["ucl"] is not None
|
|
assert data["lcl"] is not None
|
|
assert data["ucl"] > data["mean"]
|
|
assert data["lcl"] < data["mean"]
|
|
|
|
|
|
class TestDateFilter:
|
|
"""Date filter tests."""
|
|
|
|
async def test_statistics_date_filter(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Date filter parameters are accepted without error."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
await _seed_measurements(
|
|
client, measurement_tec_user, recipe.id,
|
|
[10.0, 10.1],
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/statistics/summary",
|
|
headers=auth_headers(metrologist_user),
|
|
params={
|
|
"recipe_id": recipe.id,
|
|
"date_from": "2020-01-01T00:00:00",
|
|
"date_to": "2099-12-31T23:59:59",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
# All measurements should be within this wide range
|
|
assert resp.json()["total"] >= 2
|
|
|
|
|
|
class TestRunFilter:
|
|
"""#58: statistiche per sessione di produzione."""
|
|
|
|
async def _setup(self, db_session, user, metro):
|
|
from datetime import datetime
|
|
from src.backend.models.orm.production import ProductionRun
|
|
from src.backend.models.orm.station import Station
|
|
from src.backend.models.orm.recipe import RecipeVersion
|
|
from sqlalchemy import select
|
|
recipe = await create_test_recipe(db_session, user.id)
|
|
version = (await db_session.execute(
|
|
select(RecipeVersion).where(RecipeVersion.recipe_id == recipe.id)
|
|
)).scalars().first()
|
|
station = Station(code="ST-58", name="Stazione 58", active=True, created_by=user.id)
|
|
db_session.add(station)
|
|
await db_session.commit()
|
|
runs = []
|
|
for i, lot in enumerate(["L1", "L2"]):
|
|
run = ProductionRun(
|
|
station_id=station.id, recipe_id=recipe.id, version_id=version.id,
|
|
operator_id=user.id, status="closed", lot_number=lot,
|
|
started_at=datetime(2026, 1, 1 + i),
|
|
)
|
|
db_session.add(run)
|
|
runs.append(run)
|
|
await db_session.commit()
|
|
return recipe, runs
|
|
|
|
async def test_filter_by_run_and_unfiltered(
|
|
self, client, measurement_tec_user, metrologist_user, db_session,
|
|
):
|
|
from src.backend.models.orm.measurement import Measurement
|
|
recipe, runs = await self._setup(db_session, measurement_tec_user, metrologist_user)
|
|
subtask_id, version_id = await _seed_measurements(
|
|
client, measurement_tec_user, recipe.id, [10.0, 10.1, 10.2],
|
|
)
|
|
# Assegna le prime due misure al run 1, la terza al run 2.
|
|
from sqlalchemy import select
|
|
ms = (await db_session.execute(
|
|
select(Measurement).order_by(Measurement.id)
|
|
)).scalars().all()
|
|
ms[0].production_run_id = runs[0].id
|
|
ms[1].production_run_id = runs[0].id
|
|
ms[2].production_run_id = runs[1].id
|
|
await db_session.commit()
|
|
|
|
h = auth_headers(metrologist_user)
|
|
r1 = await client.get("/api/statistics/summary", headers=h,
|
|
params={"recipe_id": recipe.id, "production_run_id": runs[0].id})
|
|
assert r1.json()["total"] == 2
|
|
r2 = await client.get("/api/statistics/summary", headers=h,
|
|
params={"recipe_id": recipe.id, "production_run_id": runs[1].id})
|
|
assert r2.json()["total"] == 1
|
|
r_all = await client.get("/api/statistics/summary", headers=h,
|
|
params={"recipe_id": recipe.id})
|
|
assert r_all.json()["total"] == 3
|
|
|
|
runs_resp = await client.get("/api/statistics/runs", headers=h,
|
|
params={"recipe_id": recipe.id})
|
|
assert runs_resp.status_code == 200
|
|
data = runs_resp.json()
|
|
assert [r["id"] for r in data] == [runs[1].id, runs[0].id] # più recente prima
|
|
assert [r["measurement_count"] for r in data] == [1, 2]
|
|
assert data[0]["lot_number"] == "L2"
|
|
assert data[0]["station_name"] == "Stazione 58"
|
|
assert data[0]["closed_at"] is None
|
|
|
|
async def test_runs_require_metrologist(
|
|
self, client, measurement_tec_user, db_session,
|
|
):
|
|
resp = await client.get("/api/statistics/runs",
|
|
headers=auth_headers(measurement_tec_user),
|
|
params={"recipe_id": 1})
|
|
assert resp.status_code == 403
|