fix(csv): neutralizza injection di formule nei campi testuali export (#22)
lot_number, serial_number (export manuale e di produzione), station.code, recipe.code, subtask.description finivano nel CSV cosi' come arrivavano: un lotto '=HYPERLINK(...)' apriva un link vivo in Excel/LibreOffice. Nuova neutralise_csv_cell() in production_export_service.py, importata da measurements.py: se il campo (spazi iniziali esclusi) comincia con =, +, -, @, TAB o CR antepone un apice, cosi' resta testo letterale. Applicata solo ai campi testuali, mai ai numeri. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,11 +1,15 @@
|
||||
"""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 auth_headers, create_test_recipe
|
||||
|
||||
|
||||
@@ -223,6 +227,281 @@ class TestMeasurementPassFail:
|
||||
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."""
|
||||
|
||||
@@ -366,3 +645,76 @@ class TestMeasurementStatistics:
|
||||
# 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)."""
|
||||
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user