Adriano/TieMeasureFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e2aae88858b879ce0340993e49b1bb87cbf2aba5
TieMeasureFlow/src/backend/tests/test_out_of_tolerance.py
T
Adriano Dal Pastro e2aae88858 feat(measure): il fuori tolleranza si autorizza, non si chiude 
Il punto 5 c'era a meta': la schermata apriva il modale del capoturno e bloccava
l'avanzamento automatico, ma il modale si chiudeva con Annulla o con un click
sullo sfondo, la misura era gia' salvata, e `pendingAdvance` veniva impostato e
non letto da nessuno. Era una conferma, non uno sbarramento.

Ora l'autorizzazione finisce sulla misura: due colonne, chi ha autorizzato e
quando. Finche' una quota e' fuori tolleranza e nessuno l'ha autorizzata, il
server rifiuta la misura della quota successiva e la chiusura del ciclo. Non e'
la schermata a impedirlo: la schermata risparmia solo il viaggio.

Rimisurare la stessa quota resta possibile — il calibro scivola, il pezzo si
riposiziona — e una seconda lettura in tolleranza libera il blocco, perche' conta
l'ultima lettura di ogni quota. Quante volte si possa riprovare e' il punto 4, e
tutte le letture restano comunque a registro. Un warning non blocca: fuori dai
limiti di attenzione ma dentro la tolleranza e' dentro la tolleranza.

Ricaricare la pagina era il modo piu' semplice per scavalcare il vecchio gate.
Non lo e' piu': la schermata chiede al server, all'apertura, se una quota sta
aspettando, e si ritrova davanti lo stesso blocco.

Sparisce /validate-supervisor, che verificava le credenziali e buttava via la
risposta. Al suo posto un endpoint che le credenziali le usa per scrivere
l'approvazione dove serve. Il controllo del capoturno si sposta in auth_service,
accanto al resto delle credenziali: fermo linea, chiusura e fuori tolleranza
fanno la stessa domanda, e solo una delle tre riguarda la produzione.

Il file di statistica guadagna authorised_by e authorised_at: mostrava il
fallimento e non la decisione, che e' la meta' che un auditor chiede.

Migrazione 010: due colonne nullable sulle misure. Le righe esistenti restano
nulle — retrodatare un'autorizzazione mai avvenuta sarebbe inventarsi un record
di audit, e quelle produzioni sono chiuse da un pezzo.

Test: +18 (309). Coprono il rifiuto della quota successiva, la rimisura ammessa,
il rilascio del blocco con una lettura buona, il ciclo che non si chiude, il
capoturno registrato sulla misura, le credenziali sbagliate e chi capoturno non
e', il file di statistica, e il fatto che una produzione chiusa non blocchi la
successiva. Aggiunto uploads/statistics/ al gitignore: i test che chiudono una
produzione scrivevano nel repository.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-28 20:54:24 +00:00

412 lines
16 KiB
Python
Raw Blame History

"""Point 5: a quote out of tolerance stops the line until someone says otherwise.
The gate existed on screen and nowhere else - a modal that opened, checked a
password and closed, while the measurement had already been saved. Dismissing it
was enough to carry on. These tests are about the part that cannot be dismissed.
"""
from httpx import AsyncClient
from sqlalchemy import select
from src.backend.models.orm.measurement import Measurement
from src.backend.models.orm.recipe import RecipeVersion
from src.backend.models.orm.station import Station
from src.backend.models.orm.task import RecipeSubtask, RecipeTask
from src.backend.services import auth_service
from src.backend.tests.conftest import auth_headers, create_test_recipe
GOOD = 10.0 # inside every limit of the fixture quote
OUT = 99.0 # far outside
async def _supervisor(db_session, username="capo-tol", password="TurnoPwd1"):
user = await auth_service.create_user(
db_session, username=username, password=password,
display_name="Capo Turno", roles=["Supervisor"],
)
await db_session.commit()
return user, password
async def _two_quotes(db_session, recipe_id: int):
"""The fixture recipe plus a second quote, so there is a 'next' one."""
version = (await db_session.execute(
select(RecipeVersion).where(
RecipeVersion.recipe_id == recipe_id,
RecipeVersion.is_current == True, # noqa: E712
)
)).scalar_one()
task = (await db_session.execute(
select(RecipeTask).where(RecipeTask.version_id == version.id)
)).scalars().first()
first = (await db_session.execute(
select(RecipeSubtask).where(RecipeSubtask.task_id == task.id)
)).scalars().first()
second = RecipeSubtask(
task_id=task.id, marker_number=2, description="Seconda quota",
nominal=10.0, utl=10.5, uwl=10.3, lwl=9.7, ltl=9.5, unit="mm",
)
db_session.add(second)
await db_session.commit()
await db_session.refresh(second)
return version.id, first, second
async def _measure(client, user, subtask_id, version_id, value, run_id=None):
body = {"subtask_id": subtask_id, "version_id": version_id, "value": value}
if run_id is not None:
body["production_run_id"] = run_id
return await client.post(
"/api/measurements/", headers=auth_headers(user), json=body,
)
async def _pending(client, user, version_id, run_id=None):
params = {"version_id": version_id}
if run_id is not None:
params["production_run_id"] = run_id
return await client.get(
"/api/measurements/pending-authorisation",
headers=auth_headers(user), params=params,
)
# ---------------------------------------------------------------------------
# The block
# ---------------------------------------------------------------------------
async def test_next_quote_is_refused_while_one_is_out_of_tolerance(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-NEXT")
version_id, first, second = await _two_quotes(db_session, recipe.id)
bad = await _measure(client, measurement_tec_user, first.id, version_id, OUT)
assert bad.status_code == 200
assert bad.json()["pass_fail"] == "fail"
refused = await _measure(client, measurement_tec_user, second.id, version_id, GOOD)
assert refused.status_code == 409
assert "supervisor" in refused.json()["detail"].lower()
async def test_the_same_quote_can_be_measured_again(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
"""The caliper slips, the piece is reseated. Trying again is not going on."""
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-AGAIN")
version_id, first, _second = await _two_quotes(db_session, recipe.id)
await _measure(client, measurement_tec_user, first.id, version_id, OUT)
again = await _measure(client, measurement_tec_user, first.id, version_id, OUT)
assert again.status_code == 200, "rimisurare la stessa quota deve restare possibile"
async def test_a_good_reading_of_the_same_quote_releases_the_block(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
"""Only the latest reading of a quote counts: it is in tolerance now."""
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-FIXED")
version_id, first, second = await _two_quotes(db_session, recipe.id)
await _measure(client, measurement_tec_user, first.id, version_id, OUT)
await _measure(client, measurement_tec_user, first.id, version_id, GOOD)
ahead = await _measure(client, measurement_tec_user, second.id, version_id, GOOD)
assert ahead.status_code == 200
# Both readings stay on record - how many are allowed is point 4's business.
rows = (await db_session.execute(
select(Measurement).where(Measurement.subtask_id == first.id)
)).scalars().all()
assert len(rows) == 2
async def test_a_warning_does_not_block(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
"""Outside the warning limits but inside tolerance is still inside tolerance."""
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-WARN")
version_id, first, second = await _two_quotes(db_session, recipe.id)
warned = await _measure(client, measurement_tec_user, first.id, version_id, 10.4)
assert warned.json()["pass_fail"] == "warning"
ahead = await _measure(client, measurement_tec_user, second.id, version_id, GOOD)
assert ahead.status_code == 200
async def test_pending_authorisation_names_the_blocking_quote(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
"""What the screen asks on load, so a reload is not a way past the gate."""
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-PEND")
version_id, first, _second = await _two_quotes(db_session, recipe.id)
assert (await _pending(client, measurement_tec_user, version_id)).json() is None
bad = (await _measure(
client, measurement_tec_user, first.id, version_id, OUT,
)).json()
pending = (await _pending(client, measurement_tec_user, version_id)).json()
assert pending["id"] == bad["id"]
assert pending["subtask_id"] == first.id
# ---------------------------------------------------------------------------
# The authorisation
# ---------------------------------------------------------------------------
async def test_supervisor_authorisation_is_recorded_and_releases_the_block(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-AUTH")
version_id, first, second = await _two_quotes(db_session, recipe.id)
supervisor, password = await _supervisor(db_session)
bad = (await _measure(
client, measurement_tec_user, first.id, version_id, OUT,
)).json()
authorised = await client.post(
f"/api/measurements/{bad['id']}/authorise",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": supervisor.username,
"supervisor_password": password,
},
)
assert authorised.status_code == 200, authorised.text
body = authorised.json()
# Who said yes is on the measurement itself, which is the whole point.
assert body["supervisor_id"] == supervisor.id
assert body["authorised_at"] is not None
ahead = await _measure(client, measurement_tec_user, second.id, version_id, GOOD)
assert ahead.status_code == 200
async def test_authorisation_refuses_someone_who_is_not_a_supervisor(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-NOSUP")
version_id, first, second = await _two_quotes(db_session, recipe.id)
bad = (await _measure(
client, measurement_tec_user, first.id, version_id, OUT,
)).json()
resp = await client.post(
f"/api/measurements/{bad['id']}/authorise",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": maker_user.username,
"supervisor_password": "testpassword123",
},
)
assert resp.status_code in (401, 403)
# And the block is still there.
refused = await _measure(client, measurement_tec_user, second.id, version_id, GOOD)
assert refused.status_code == 409
async def test_authorisation_refuses_wrong_credentials(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-BADPW")
version_id, first, _second = await _two_quotes(db_session, recipe.id)
supervisor, _password = await _supervisor(db_session, username="capo-badpw")
bad = (await _measure(
client, measurement_tec_user, first.id, version_id, OUT,
)).json()
resp = await client.post(
f"/api/measurements/{bad['id']}/authorise",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": supervisor.username,
"supervisor_password": "sbagliata",
},
)
assert resp.status_code == 401
async def test_there_is_nothing_to_authorise_on_a_good_measurement(
client: AsyncClient, maker_user, measurement_tec_user, db_session,
):
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-GOOD")
version_id, first, _second = await _two_quotes(db_session, recipe.id)
supervisor, password = await _supervisor(db_session, username="capo-good")
good = (await _measure(
client, measurement_tec_user, first.id, version_id, GOOD,
)).json()
resp = await client.post(
f"/api/measurements/{good['id']}/authorise",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": supervisor.username,
"supervisor_password": password,
},
)
assert resp.status_code == 409
# ---------------------------------------------------------------------------
# The production run
# ---------------------------------------------------------------------------
async def test_cycle_cannot_close_over_an_unauthorised_fail(
client: AsyncClient, maker_user, measurement_tec_user, admin_user, db_session,
):
"""Closing the cycle would be another way of walking past the quote."""
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-CYC")
recipe.measurement_interval_minutes = 10
station = Station(
code="ST-TOL", name="Stazione", active=True, created_by=admin_user.id,
)
db_session.add(station)
await db_session.commit()
version_id, first, _second = await _two_quotes(db_session, recipe.id)
run = (await client.post(
"/api/production-runs",
headers=auth_headers(measurement_tec_user),
json={"station_code": station.code, "recipe_id": recipe.id},
)).json()
await _measure(
client, measurement_tec_user, first.id, version_id, OUT, run_id=run["id"],
)
refused = await client.post(
f"/api/production-runs/{run['id']}/cycle",
headers=auth_headers(measurement_tec_user),
)
assert refused.status_code == 409
supervisor, password = await _supervisor(db_session, username="capo-cyc")
blocking = (await _pending(
client, measurement_tec_user, version_id, run_id=run["id"],
)).json()
await client.post(
f"/api/measurements/{blocking['id']}/authorise",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": supervisor.username,
"supervisor_password": password,
},
)
allowed = await client.post(
f"/api/production-runs/{run['id']}/cycle",
headers=auth_headers(measurement_tec_user),
)
assert allowed.status_code == 200
assert allowed.json()["cycle_count"] == 1
async def test_the_statistics_file_carries_the_decision(
client: AsyncClient, maker_user, measurement_tec_user, admin_user, db_session,
tmp_path, monkeypatch,
):
"""The file has to show the failure and who let it stand, not just the failure."""
from src.backend.config import settings
monkeypatch.setattr(
type(settings), "upload_path", property(lambda self: tmp_path), raising=False,
)
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-CSV")
station = Station(
code="ST-CSV", name="Stazione", active=True, created_by=admin_user.id,
)
db_session.add(station)
await db_session.commit()
version_id, first, _second = await _two_quotes(db_session, recipe.id)
supervisor, password = await _supervisor(db_session, username="capo-csv")
creds = {
"supervisor_username": supervisor.username, "supervisor_password": password,
}
run = (await client.post(
"/api/production-runs",
headers=auth_headers(measurement_tec_user),
json={"station_code": station.code, "recipe_id": recipe.id},
)).json()
bad = (await _measure(
client, measurement_tec_user, first.id, version_id, OUT, run_id=run["id"],
)).json()
await client.post(
f"/api/measurements/{bad['id']}/authorise",
headers=auth_headers(measurement_tec_user), json=creds,
)
closed = (await client.post(
f"/api/production-runs/{run['id']}/close",
headers=auth_headers(measurement_tec_user), json=creds,
)).json()
text = (tmp_path / closed["statistics_path"]).read_text(encoding="utf-8-sig")
header, row = [line for line in text.splitlines() if line.strip()]
assert "authorised_by" in header and "authorised_at" in header
assert str(supervisor.id) in row
async def test_a_fail_from_another_production_does_not_block_this_one(
client: AsyncClient, maker_user, measurement_tec_user, admin_user, db_session,
tmp_path, monkeypatch,
):
"""The run is the boundary: an old production must not stop a new one."""
# Closing a run writes the statistics file; pointed at tmp_path so the test
# does not leave one in the repository's uploads directory.
from src.backend.config import settings
monkeypatch.setattr(
type(settings), "upload_path", property(lambda self: tmp_path), raising=False,
)
recipe = await create_test_recipe(db_session, maker_user.id, code="TOL-SCOPE")
station = Station(
code="ST-SCOPE", name="Stazione", active=True, created_by=admin_user.id,
)
db_session.add(station)
await db_session.commit()
version_id, first, second = await _two_quotes(db_session, recipe.id)
first_run = (await client.post(
"/api/production-runs",
headers=auth_headers(measurement_tec_user),
json={"station_code": station.code, "recipe_id": recipe.id},
)).json()
await _measure(
client, measurement_tec_user, first.id, version_id, OUT,
run_id=first_run["id"],
)
supervisor, password = await _supervisor(db_session, username="capo-scope")
await client.post(
f"/api/production-runs/{first_run['id']}/close",
headers=auth_headers(measurement_tec_user),
json={
"supervisor_username": supervisor.username,
"supervisor_password": password,
},
)
second_run = (await client.post(
"/api/production-runs",
headers=auth_headers(measurement_tec_user),
json={"station_code": station.code, "recipe_id": recipe.id},
)).json()
resp = await _measure(
client, measurement_tec_user, second.id, version_id, GOOD,
run_id=second_run["id"],
)
assert resp.status_code == 200
Reference in New Issue View Git Blame Copy Permalink