Adriano
1d5f24510c
Il controllo ruolo di GET /api/measurements/export/csv ignorava is_admin: un admin senza MeasurementTec/Metrologist assegnato prendeva 403, a differenza di ogni altro endpoint che passa da require_role (ruolo o admin). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
748 lines
26 KiB
Python
748 lines
26 KiB
Python
"""Tests for measurements router (/api/measurements)."""
|
|
import csv
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.backend.models.orm.station import Station
|
|
from src.backend.models.orm.user import User
|
|
from src.backend.models.orm.recipe import RecipeVersion
|
|
from src.backend.models.orm.task import RecipeTask, RecipeSubtask
|
|
from src.backend.services import auth_service
|
|
from src.backend.tests.conftest import _create_user, auth_headers, create_test_recipe
|
|
|
|
|
|
async def _get_subtask_and_version(
|
|
client: AsyncClient, user: User, recipe_id: int
|
|
) -> tuple[int, int]:
|
|
"""Helper: return (subtask_id, version_id) from the recipe's current version."""
|
|
tasks_resp = await client.get(
|
|
f"/api/recipes/{recipe_id}/tasks", headers=auth_headers(user)
|
|
)
|
|
task = tasks_resp.json()[0]
|
|
subtask_id = task["subtasks"][0]["id"]
|
|
version_id = task["version_id"]
|
|
return subtask_id, version_id
|
|
|
|
|
|
class TestCreateMeasurement:
|
|
"""POST /api/measurements/ tests."""
|
|
|
|
async def test_create_measurement(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""MeasurementTec can create a measurement."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.1,
|
|
"input_method": "manual",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["value"] == pytest.approx(10.1, rel=1e-4)
|
|
assert data["pass_fail"] in ("pass", "warning", "fail")
|
|
assert data["measured_by"] == measurement_tec_user.id
|
|
|
|
async def test_create_measurement_with_input_duration(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""input_duration_ms is stored and returned when provided."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"input_duration_ms": 4250,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["input_duration_ms"] == 4250
|
|
|
|
async def test_create_measurement_without_input_duration(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""input_duration_ms defaults to null when omitted."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["input_duration_ms"] is None
|
|
|
|
async def test_measurement_requires_auth(self, client: AsyncClient):
|
|
"""Creating measurement without auth returns 401."""
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
json={
|
|
"subtask_id": 1,
|
|
"version_id": 1,
|
|
"value": 10.0,
|
|
},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
async def test_measurement_validation(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
):
|
|
"""Measurement with non-existent subtask returns 400."""
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": 99999,
|
|
"version_id": 99999,
|
|
"value": 10.0,
|
|
},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestMeasurementPassFail:
|
|
"""Pass/fail calculation tests."""
|
|
|
|
async def test_measurement_pass(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Value within tolerance -> pass."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
# nominal=10.0, utl=10.5, uwl=10.3, lwl=9.7, ltl=9.5
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["pass_fail"] == "pass"
|
|
|
|
async def test_measurement_warning(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Value outside warning but inside tolerance -> warning."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
# uwl=10.3 < 10.4 < utl=10.5 -> warning
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.4,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["pass_fail"] == "warning"
|
|
|
|
async def test_measurement_fail(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Value outside tolerance limits -> fail."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
# value > utl=10.5 -> fail
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.6,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["pass_fail"] == "fail"
|
|
|
|
|
|
class TestMeasurementValueBounds:
|
|
"""value rejects NaN/Infinity and out-of-DECIMAL(12,6)-scale numbers (#20)."""
|
|
|
|
@pytest.mark.parametrize("bad_value", [1e300, 1_000_000])
|
|
async def test_out_of_scale_value_is_422(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
bad_value: float,
|
|
):
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": bad_value,
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
@pytest.mark.parametrize("raw_token", ["NaN", "Infinity", "-Infinity"])
|
|
async def test_nan_and_infinity_are_422(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
raw_token: str,
|
|
):
|
|
"""NaN/Infinity are not valid JSON: httpx's own encoder refuses to send
|
|
them via ``json=``, so the raw body is built by hand here - exactly
|
|
what a client that is not httpx could still put on the wire."""
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
body = (
|
|
'{"subtask_id": %d, "version_id": %d, "value": %s}'
|
|
% (subtask_id, version_id, raw_token)
|
|
)
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers={**auth_headers(measurement_tec_user), "Content-Type": "application/json"},
|
|
content=body,
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
async def test_value_is_rounded_before_the_verdict(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""10.5000004 rounds to 10.5 (=utl) first, so it is not a fail."""
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.5000004,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["value"] == pytest.approx(10.5)
|
|
# 10.5 is > uwl(10.3) but not > utl(10.5): warning, not fail.
|
|
assert data["pass_fail"] == "warning"
|
|
|
|
|
|
class TestSaveMeasurementCoherence:
|
|
"""save_measurement checks quote/version/run coherence before insert (#5)."""
|
|
|
|
async def _station(self, db_session, user_id, code):
|
|
station = Station(
|
|
code=code, name=f"Stazione {code}", active=True, created_by=user_id,
|
|
)
|
|
db_session.add(station)
|
|
await db_session.commit()
|
|
await db_session.refresh(station)
|
|
return station
|
|
|
|
async def _supervisor(self, db_session, username):
|
|
user = await auth_service.create_user(
|
|
db_session, username=username, password="TurnoPwd1",
|
|
display_name="Capo Turno", roles=["Supervisor"],
|
|
)
|
|
await db_session.commit()
|
|
return user, "TurnoPwd1"
|
|
|
|
async def test_quota_of_another_recipe_version_is_422(
|
|
self, client: AsyncClient, measurement_tec_user: User, db_session: AsyncSession,
|
|
):
|
|
recipe_a = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-A")
|
|
recipe_b = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-B")
|
|
subtask_a, _version_a = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe_a.id
|
|
)
|
|
_subtask_b, version_b = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe_b.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={"subtask_id": subtask_a, "version_id": version_b, "value": 10.0},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
async def test_measurement_of_other_recipe_in_run_is_422(
|
|
self, client: AsyncClient, measurement_tec_user: User, admin_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
recipe_a = await create_test_recipe(
|
|
db_session, measurement_tec_user.id, code="COH-RUNA"
|
|
)
|
|
recipe_b = await create_test_recipe(
|
|
db_session, measurement_tec_user.id, code="COH-RUNB"
|
|
)
|
|
station = await self._station(db_session, admin_user.id, "ST-COH1")
|
|
|
|
run = (await client.post(
|
|
"/api/production-runs",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={"station_code": station.code, "recipe_id": recipe_a.id},
|
|
)).json()
|
|
|
|
subtask_b, version_b = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe_b.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_b,
|
|
"version_id": version_b,
|
|
"value": 10.0,
|
|
"production_run_id": run["id"],
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
async def test_measurement_on_closed_run_is_409(
|
|
self, client: AsyncClient, measurement_tec_user: User, admin_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-CLOSED")
|
|
station = await self._station(db_session, admin_user.id, "ST-COH2")
|
|
run = (await client.post(
|
|
"/api/production-runs",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={"station_code": station.code, "recipe_id": recipe.id},
|
|
)).json()
|
|
supervisor, password = await self._supervisor(db_session, "capo-coh-close")
|
|
await client.post(
|
|
f"/api/production-runs/{run['id']}/close",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"supervisor_username": supervisor.username,
|
|
"supervisor_password": password,
|
|
},
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"production_run_id": run["id"],
|
|
},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
async def test_measurement_on_paused_run_is_409(
|
|
self, client: AsyncClient, measurement_tec_user: User, admin_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-PAUSED")
|
|
station = await self._station(db_session, admin_user.id, "ST-COH3")
|
|
run = (await client.post(
|
|
"/api/production-runs",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={"station_code": station.code, "recipe_id": recipe.id},
|
|
)).json()
|
|
supervisor, password = await self._supervisor(db_session, "capo-coh-pause")
|
|
await client.post(
|
|
f"/api/production-runs/{run['id']}/pause",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"supervisor_username": supervisor.username,
|
|
"supervisor_password": password,
|
|
},
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"production_run_id": run["id"],
|
|
},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
async def test_valid_measurement_in_matching_run_is_200(
|
|
self, client: AsyncClient, measurement_tec_user: User, admin_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-OK")
|
|
station = await self._station(db_session, admin_user.id, "ST-COH4")
|
|
run = (await client.post(
|
|
"/api/production-runs",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={"station_code": station.code, "recipe_id": recipe.id},
|
|
)).json()
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"production_run_id": run["id"],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
async def test_measurement_on_unknown_run_is_404(
|
|
self, client: AsyncClient, measurement_tec_user: User, db_session: AsyncSession,
|
|
):
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id, code="COH-404")
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"production_run_id": 999999,
|
|
},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestListMeasurements:
|
|
"""GET /api/measurements/ tests."""
|
|
|
|
async def test_list_measurements(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Metrologist can list measurements with pagination."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
# Create some measurements
|
|
for val in [9.8, 10.0, 10.2]:
|
|
await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": val,
|
|
},
|
|
)
|
|
|
|
# List as Metrologist
|
|
resp = await client.get(
|
|
"/api/measurements/",
|
|
headers=auth_headers(metrologist_user),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
assert data["total"] >= 3
|
|
|
|
async def test_measurement_tec_can_list_measurements(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Regression: MeasurementTec must be able to list measurements.
|
|
|
|
The Flask client renders task_complete.html by calling GET
|
|
/api/measurements?version_id=X right after the operator finishes a
|
|
recipe. The endpoint used to require Metrologist, which silently
|
|
produced an empty riepilogo for every operator without that role.
|
|
"""
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
for val in [9.8, 10.0, 10.2]:
|
|
await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": val,
|
|
},
|
|
)
|
|
|
|
resp = await client.get(
|
|
f"/api/measurements/?version_id={version_id}",
|
|
headers=auth_headers(measurement_tec_user),
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["total"] >= 3
|
|
|
|
async def test_measurement_by_recipe_filter(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
metrologist_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Filtering measurements by recipe_id returns only matching records."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
},
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/measurements/",
|
|
headers=auth_headers(metrologist_user),
|
|
params={"recipe_id": recipe.id},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["total"] >= 1
|
|
|
|
|
|
class TestMeasurementStatistics:
|
|
"""Measurement deviation / statistics edge case tests."""
|
|
|
|
async def test_measurement_has_deviation(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Measurement response includes deviation from nominal."""
|
|
recipe = await create_test_recipe(
|
|
db_session, measurement_tec_user.id
|
|
)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.2,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# deviation = 10.2 - nominal(10.0) = 0.2
|
|
assert data["deviation"] is not None
|
|
assert abs(data["deviation"] - 0.2) < 0.001
|
|
|
|
|
|
class TestExportCsv:
|
|
"""GET /api/measurements/export/csv tests (gitea#22, gitea#36 point 3)."""
|
|
|
|
async def test_admin_without_explicit_role_can_export(
|
|
self, client: AsyncClient, db_session: AsyncSession,
|
|
):
|
|
"""An admin with no MeasurementTec/Metrologist role must not get 403 (#36.3).
|
|
|
|
admin_user (conftest) already has both roles, which would pass even
|
|
without the fix: a bare admin (roles=[]) is the actual regression case.
|
|
"""
|
|
bare_admin = await _create_user(
|
|
db_session, username="bare_admin", roles=[], is_admin=True,
|
|
)
|
|
resp = await client.get(
|
|
"/api/measurements/export/csv",
|
|
headers=auth_headers(bare_admin),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
async def test_maker_without_role_still_forbidden(
|
|
self, client: AsyncClient, maker_user: User,
|
|
):
|
|
"""A plain Maker (no MeasurementTec/Metrologist/admin) still gets 403."""
|
|
resp = await client.get(
|
|
"/api/measurements/export/csv",
|
|
headers=auth_headers(maker_user),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
async def test_lot_number_formula_is_neutralised(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""A lot_number starting with '=' must not reach the CSV as a live formula (#22)."""
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": 10.0,
|
|
"lot_number": '=HYPERLINK("http://x","y")',
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
export_resp = await client.get(
|
|
"/api/measurements/export/csv",
|
|
headers=auth_headers(measurement_tec_user),
|
|
)
|
|
assert export_resp.status_code == 200
|
|
rows = list(csv.reader(export_resp.text.splitlines()))
|
|
header, row = rows[0], rows[1]
|
|
lot_cell = row[header.index("lot_number")]
|
|
# Quoted with a leading ' so a spreadsheet reads it as literal text,
|
|
# not the raw formula that would otherwise be live in Excel/LibreOffice.
|
|
assert lot_cell == '\'=HYPERLINK("http://x","y")'
|
|
|
|
async def test_negative_value_stays_numeric(
|
|
self,
|
|
client: AsyncClient,
|
|
measurement_tec_user: User,
|
|
db_session: AsyncSession,
|
|
):
|
|
"""Neutralisation must never touch numeric cells like the measured value."""
|
|
recipe = await create_test_recipe(db_session, measurement_tec_user.id)
|
|
subtask_id, version_id = await _get_subtask_and_version(
|
|
client, measurement_tec_user, recipe.id
|
|
)
|
|
|
|
resp = await client.post(
|
|
"/api/measurements/",
|
|
headers=auth_headers(measurement_tec_user),
|
|
json={
|
|
"subtask_id": subtask_id,
|
|
"version_id": version_id,
|
|
"value": -1.25,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
export_resp = await client.get(
|
|
"/api/measurements/export/csv",
|
|
headers=auth_headers(measurement_tec_user),
|
|
)
|
|
assert export_resp.status_code == 200
|
|
rows = list(csv.reader(export_resp.text.splitlines()))
|
|
header, row = rows[0], rows[1]
|
|
value_cell = row[header.index("value")]
|
|
assert value_cell == "-1.250000"
|