Task results: upload allegati per MeasurementTec, percorsi in task_results/, test e migrazione 018 (#55 #53 #54)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -168,3 +168,68 @@ class TestUserPinMigration:
|
||||
assert not (new & cols())
|
||||
command.upgrade(cfg, "head") # e di nuovo su
|
||||
assert new <= cols()
|
||||
|
||||
|
||||
_SEED_USER_RECIPE_VERSION = [
|
||||
"INSERT INTO users (username, password_hash, display_name, roles, is_admin, active,"
|
||||
" language_pref, theme_pref) VALUES ('u', 'x', 'U', '[]', 1, 1, 'en', 'light')",
|
||||
"INSERT INTO recipes (code, name, active, created_by) VALUES ('R', 'R', 1, 1)",
|
||||
"INSERT INTO recipe_versions (recipe_id, version_number, is_current, created_by)"
|
||||
" VALUES (1, 1, 1, 1)",
|
||||
]
|
||||
|
||||
|
||||
class TestTaskLoopResultsMigration:
|
||||
"""018_task_loop_results: in_loop/at_close/fields, tipo check, task_results (#53 #54 #55)."""
|
||||
|
||||
def test_018_upgrade_and_downgrade(self, sqlite_database_url):
|
||||
db_path, _ = sqlite_database_url
|
||||
cfg = _alembic_config()
|
||||
|
||||
def cols():
|
||||
return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
|
||||
|
||||
command.upgrade(cfg, "head")
|
||||
assert {"in_loop", "at_close", "fields"} <= cols()
|
||||
assert "task_results" in _inspect(db_path).get_table_names()
|
||||
command.downgrade(cfg, "017_user_pin")
|
||||
assert not ({"in_loop", "at_close", "fields"} & cols())
|
||||
assert "task_results" not in _inspect(db_path).get_table_names()
|
||||
command.upgrade(cfg, "head") # e di nuovo su
|
||||
|
||||
def test_018_marks_existing_measure_tasks_in_loop(self, sqlite_database_url):
|
||||
db_path, _ = sqlite_database_url
|
||||
cfg = _alembic_config()
|
||||
command.upgrade(cfg, "017_user_pin")
|
||||
engine = create_engine(f"sqlite:///{db_path}")
|
||||
with engine.begin() as conn:
|
||||
for stmt in _SEED_USER_RECIPE_VERSION:
|
||||
conn.execute(text(stmt))
|
||||
for i, t in enumerate(("measure", "camera_measure", "dxf_compare", "note", "drawing")):
|
||||
conn.execute(text(
|
||||
"INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
|
||||
f" VALUES (1, {i}, '{t}', '{t}')"))
|
||||
engine.dispose()
|
||||
|
||||
command.upgrade(cfg, "head")
|
||||
engine = create_engine(f"sqlite:///{db_path}")
|
||||
with engine.connect() as conn:
|
||||
rows = dict(conn.execute(text("SELECT task_type, in_loop FROM recipe_tasks")).all())
|
||||
engine.dispose()
|
||||
assert rows == {"measure": 1, "camera_measure": 1, "dxf_compare": 1, "note": 0, "drawing": 0}
|
||||
|
||||
def test_018_downgrade_refused_when_a_check_task_exists(self, sqlite_database_url):
|
||||
db_path, _ = sqlite_database_url
|
||||
cfg = _alembic_config()
|
||||
command.upgrade(cfg, "head")
|
||||
engine = create_engine(f"sqlite:///{db_path}")
|
||||
with engine.begin() as conn:
|
||||
for stmt in _SEED_USER_RECIPE_VERSION:
|
||||
conn.execute(text(stmt))
|
||||
conn.execute(text(
|
||||
"INSERT INTO recipe_tasks (version_id, order_index, task_type, title)"
|
||||
" VALUES (1, 0, 'check', 'c')"))
|
||||
engine.dispose()
|
||||
|
||||
with pytest.raises(RuntimeError, match="check"):
|
||||
command.downgrade(cfg, "017_user_pin")
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
"""Test di in_loop / at_close / fields e del tipo 'check' sui task (#53 #54 #55),
|
||||
di close_task_ids nella risposta del run e del bucket di rate limit del login PIN."""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from src.backend.api.middleware.rate_limit import RateLimitMiddleware
|
||||
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.tests.conftest import auth_headers, create_test_recipe
|
||||
|
||||
FIELDS = [
|
||||
{"key": "peso", "label": "Peso", "type": "number", "default": "1.5", "attachment": False},
|
||||
{"key": "foto", "label": "Foto", "type": "text", "default": None, "attachment": True},
|
||||
]
|
||||
|
||||
|
||||
async def _post_task(client, user, recipe_id, **body):
|
||||
return await client.post(
|
||||
f"/api/recipes/{recipe_id}/tasks", headers=auth_headers(user), json=body,
|
||||
)
|
||||
|
||||
|
||||
async def test_create_defaults_in_loop_from_type(client: AsyncClient, maker_user, db_session):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L1")
|
||||
await db_session.commit()
|
||||
expected = {"note": False, "drawing": False, "measure": True,
|
||||
"camera_measure": True, "dxf_compare": True, "check": True}
|
||||
for task_type, in_loop in expected.items():
|
||||
resp = await _post_task(client, maker_user, recipe.id, title=task_type, task_type=task_type)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["in_loop"] is in_loop, task_type
|
||||
assert body["at_close"] is False
|
||||
assert body["fields"] is None
|
||||
|
||||
|
||||
async def test_create_explicit_in_loop_and_note_fields(client: AsyncClient, maker_user, db_session):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L2")
|
||||
await db_session.commit()
|
||||
resp = await _post_task(
|
||||
client, maker_user, recipe.id, title="Nota", task_type="note",
|
||||
in_loop=True, at_close=True, fields=FIELDS,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["in_loop"] is True and body["at_close"] is True
|
||||
assert body["fields"] == FIELDS
|
||||
|
||||
|
||||
@pytest.mark.parametrize("task_type", ["measure", "check", "drawing"])
|
||||
async def test_create_at_close_or_fields_rejected_for_other_types(
|
||||
client: AsyncClient, maker_user, db_session, task_type,
|
||||
):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code=f"REC-L3{task_type}")
|
||||
await db_session.commit()
|
||||
resp = await _post_task(client, maker_user, recipe.id, title="x", task_type=task_type, at_close=True)
|
||||
assert resp.status_code == 422
|
||||
assert isinstance(resp.json()["detail"], str)
|
||||
resp = await _post_task(client, maker_user, recipe.id, title="x", task_type=task_type, fields=FIELDS)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [
|
||||
[{"key": "A b", "label": "x", "type": "text"}], # chiave non slug
|
||||
[{"key": "k", "label": "", "type": "text"}], # etichetta vuota
|
||||
[{"key": "k", "label": "x", "type": "date"}], # tipo ignoto
|
||||
[{"key": "k", "label": "x", "type": "text"}, {"key": "k", "label": "y", "type": "text"}], # duplicata
|
||||
[{"key": "k", "label": "x", "type": "number", "default": "abc"}], # default non numerico
|
||||
[{"key": f"k{i}", "label": "x", "type": "text"} for i in range(21)], # troppi
|
||||
])
|
||||
async def test_create_invalid_fields_rejected(client: AsyncClient, maker_user, db_session, bad):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L4")
|
||||
await db_session.commit()
|
||||
resp = await _post_task(client, maker_user, recipe.id, title="n", task_type="note", fields=bad)
|
||||
assert resp.status_code == 422
|
||||
assert isinstance(resp.json()["detail"], str)
|
||||
|
||||
|
||||
async def test_update_fields_null_clears_and_type_rules(client: AsyncClient, maker_user, db_session):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L5")
|
||||
await db_session.commit()
|
||||
created = (await _post_task(
|
||||
client, maker_user, recipe.id, title="Nota", task_type="note",
|
||||
at_close=True, fields=FIELDS,
|
||||
)).json()
|
||||
headers = auth_headers(maker_user)
|
||||
|
||||
# il frontend manda sempre i tre campi: fields null = nessun campo
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=headers,
|
||||
json={"in_loop": True, "at_close": False, "fields": None},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["in_loop"] is True and body["at_close"] is False and body["fields"] is None
|
||||
|
||||
# tipo diverso da note con at_close o fields: 422
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=headers,
|
||||
json={"task_type": "measure", "at_close": True, "fields": None},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=headers,
|
||||
json={"task_type": "check", "at_close": False, "fields": FIELDS},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
# un task che ha gia' at_close non puo' passare a measure senza toglierlo
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=headers,
|
||||
json={"at_close": True, "fields": FIELDS},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=headers, json={"task_type": "measure"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
async def test_copy_on_write_carries_the_three_fields(
|
||||
client: AsyncClient, maker_user, db_session,
|
||||
):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L6")
|
||||
await db_session.commit()
|
||||
created = (await _post_task(
|
||||
client, maker_user, recipe.id, title="Nota", task_type="note",
|
||||
in_loop=True, at_close=True, fields=FIELDS,
|
||||
)).json()
|
||||
|
||||
# una misura sulla versione corrente forza il copy-on-write alla prossima modifica
|
||||
version = (await db_session.execute(
|
||||
select(RecipeVersion).where(
|
||||
RecipeVersion.recipe_id == recipe.id, RecipeVersion.is_current == True, # noqa: E712
|
||||
)
|
||||
)).scalar_one()
|
||||
subtask_id = (await db_session.execute(
|
||||
select(RecipeSubtask.id).join(RecipeSubtask.task)
|
||||
.where(RecipeSubtask.task.has(version_id=version.id))
|
||||
)).scalars().first()
|
||||
db_session.add(Measurement(
|
||||
subtask_id=subtask_id, version_id=version.id, measured_by=maker_user.id,
|
||||
value=10.0, pass_fail="pass", deviation=0.0, input_method="manual",
|
||||
))
|
||||
await db_session.commit()
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/tasks/{created['id']}", headers=auth_headers(maker_user),
|
||||
json={"title": "Nota rinominata"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["version_id"] != version.id # e' la copia nella nuova versione
|
||||
assert body["in_loop"] is True and body["at_close"] is True
|
||||
assert body["fields"] == FIELDS
|
||||
|
||||
|
||||
async def test_run_response_loop_and_close_task_ids(
|
||||
client: AsyncClient, maker_user, measurement_tec_user, admin_user, db_session,
|
||||
):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code="REC-L7")
|
||||
db_session.add(Station(code="ST-L7", name="S", active=True, created_by=admin_user.id))
|
||||
await db_session.commit()
|
||||
|
||||
async def mk(**body):
|
||||
resp = await _post_task(client, maker_user, recipe.id, **body)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()
|
||||
|
||||
check = await mk(title="Check", task_type="check")
|
||||
note_loop = await mk(title="Nota ciclo", task_type="note", in_loop=True)
|
||||
note_out = await mk(title="Nota fuori", task_type="note")
|
||||
close_a = await mk(title="Chiusura A", task_type="note", at_close=True)
|
||||
measure_off = await mk(title="Misura fuori", task_type="measure", in_loop=False)
|
||||
close_b = await mk(title="Chiusura B", task_type="note", at_close=True)
|
||||
|
||||
first_id = (await db_session.execute(
|
||||
select(RecipeTask.id).where(RecipeTask.version_id == check["version_id"])
|
||||
.order_by(RecipeTask.order_index)
|
||||
)).scalars().first()
|
||||
|
||||
resp = await client.post(
|
||||
"/api/production-runs", headers=auth_headers(measurement_tec_user),
|
||||
json={"station_code": "ST-L7", "recipe_id": recipe.id},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
# ciclo: solo in_loop, in ordine (misura della fixture, check, nota in ciclo)
|
||||
assert body["measurement_task_ids"] == [first_id, check["id"], note_loop["id"]]
|
||||
assert note_out["id"] not in body["measurement_task_ids"]
|
||||
assert measure_off["id"] not in body["measurement_task_ids"]
|
||||
assert body["close_task_ids"] == [close_a["id"], close_b["id"]]
|
||||
|
||||
|
||||
class TestLoginPinRateLimitBucket:
|
||||
"""#59: il login con PIN sta nel bucket 'login' come il login con password."""
|
||||
|
||||
def test_login_pin_is_login_bucket(self):
|
||||
assert RateLimitMiddleware._is_login_bucket_request("POST", "/api/auth/login-pin")
|
||||
|
||||
def test_login_pin_get_is_not_login_bucket(self):
|
||||
assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/auth/login-pin")
|
||||
|
||||
def test_pin_users_list_is_not_login_bucket(self):
|
||||
assert not RateLimitMiddleware._is_login_bucket_request("GET", "/api/auth/pin-users")
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Test di POST/GET /api/task-results (#54 check OK/NOK, #55 note con campi)."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from src.backend.models.orm.production import ProductionRun
|
||||
from src.backend.models.orm.station import Station
|
||||
from src.backend.tests.conftest import auth_headers, create_test_recipe
|
||||
|
||||
FIELDS = [
|
||||
{"key": "peso", "label": "Peso", "type": "number", "default": None, "attachment": False},
|
||||
{"key": "foto", "label": "Foto", "type": "text", "default": None, "attachment": True},
|
||||
]
|
||||
|
||||
|
||||
async def _setup(client, db_session, maker_user, admin_user, code):
|
||||
recipe = await create_test_recipe(db_session, user_id=maker_user.id, code=code)
|
||||
db_session.add(Station(code=f"ST-{code}", name="S", active=True, created_by=admin_user.id))
|
||||
await db_session.commit()
|
||||
h = auth_headers(maker_user)
|
||||
check = (await client.post(
|
||||
f"/api/recipes/{recipe.id}/tasks", headers=h, json={"title": "C", "task_type": "check"},
|
||||
)).json()
|
||||
note = (await client.post(
|
||||
f"/api/recipes/{recipe.id}/tasks", headers=h,
|
||||
json={"title": "N", "task_type": "note", "fields": FIELDS},
|
||||
)).json()
|
||||
return recipe, check, note
|
||||
|
||||
|
||||
async def _open_run(client, user, recipe, code):
|
||||
resp = await client.post(
|
||||
"/api/production-runs", headers=auth_headers(user),
|
||||
json={"station_code": f"ST-{code}", "recipe_id": recipe.id},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()
|
||||
|
||||
|
||||
async def _post(client, user, **body):
|
||||
return await client.post("/api/task-results", headers=auth_headers(user), json=body)
|
||||
|
||||
|
||||
async def test_requires_auth_and_role(client: AsyncClient, maker_user, admin_user, db_session):
|
||||
_, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR0")
|
||||
resp = await client.post("/api/task-results", json={"task_id": check["id"], "outcome": "ok"})
|
||||
assert resp.status_code == 401
|
||||
# un Maker senza ruolo di reparto non registra esiti
|
||||
resp = await _post(client, maker_user, task_id=check["id"], outcome="ok")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
async def test_check_ok_and_nok_rules(
|
||||
client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
|
||||
):
|
||||
recipe, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR1")
|
||||
run = await _open_run(client, measurement_tec_user, recipe, "TR1")
|
||||
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
production_run_id=run["id"], outcome="ok")
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["outcome"] == "ok"
|
||||
assert body["version_id"] == check["version_id"]
|
||||
assert body["user_id"] == measurement_tec_user.id
|
||||
assert body["production_run_id"] == run["id"]
|
||||
assert body["created_at"]
|
||||
|
||||
for note in (None, "", " "):
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
outcome="nok", note=note)
|
||||
assert resp.status_code == 422
|
||||
assert resp.json()["detail"] == "Per un NOK serve la nota"
|
||||
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
outcome="nok", note=" graffio ")
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["note"] == "graffio"
|
||||
|
||||
# un check non si chiude con 'done'
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"], outcome="done")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
async def test_note_values_and_attachments(
|
||||
client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
|
||||
):
|
||||
recipe, _, note = await _setup(client, db_session, maker_user, admin_user, "TR2")
|
||||
ok = await _post(
|
||||
client, measurement_tec_user, task_id=note["id"], outcome="done",
|
||||
values={"peso": "12,5"},
|
||||
attachments=[{"key": "foto", "path": "task_results/2026/10/abc.jpg", "name": "foto.jpg"}],
|
||||
)
|
||||
assert ok.status_code == 201, ok.text
|
||||
assert ok.json()["values"] == {"peso": "12,5"}
|
||||
assert ok.json()["attachments"] == [{"key": "foto", "path": "task_results/2026/10/abc.jpg", "name": "foto.jpg"}]
|
||||
|
||||
# solo done
|
||||
assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="ok")).status_code == 422
|
||||
# chiave fuori dai campi
|
||||
assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
|
||||
values={"altro": "1"})).status_code == 422
|
||||
# numero non convertibile
|
||||
assert (await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
|
||||
values={"peso": "abc"})).status_code == 422
|
||||
# allegati con percorso non sicuro
|
||||
for bad in ("../x.jpg", "/etc/passwd", "a/../b.jpg", "C:/x.jpg", "a\\b.jpg"):
|
||||
resp = await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
|
||||
attachments=[{"key": None, "path": bad, "name": "x"}])
|
||||
assert resp.status_code == 422, bad
|
||||
# chiave allegato fuori dai campi
|
||||
resp = await _post(client, measurement_tec_user, task_id=note["id"], outcome="done",
|
||||
attachments=[{"key": "boh", "path": "task_results/a.jpg", "name": "x"}])
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
async def test_other_task_types_rejected_and_unknown_task(
|
||||
client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
|
||||
):
|
||||
recipe, _, _ = await _setup(client, db_session, maker_user, admin_user, "TR3")
|
||||
measure = (await client.post(
|
||||
f"/api/recipes/{recipe.id}/tasks", headers=auth_headers(maker_user),
|
||||
json={"title": "M", "task_type": "measure"},
|
||||
)).json()
|
||||
resp = await _post(client, measurement_tec_user, task_id=measure["id"], outcome="ok")
|
||||
assert resp.status_code == 422
|
||||
resp = await _post(client, measurement_tec_user, task_id=999999, outcome="ok")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
async def test_run_must_be_open_and_same_version(
|
||||
client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
|
||||
):
|
||||
recipe, check, _ = await _setup(client, db_session, maker_user, admin_user, "TR4")
|
||||
_, other_check, _ = await _setup(client, db_session, maker_user, admin_user, "TR4b")
|
||||
run = await _open_run(client, measurement_tec_user, recipe, "TR4")
|
||||
|
||||
# task di un'altra versione
|
||||
resp = await _post(client, measurement_tec_user, task_id=other_check["id"],
|
||||
production_run_id=run["id"], outcome="ok")
|
||||
assert resp.status_code == 422
|
||||
# run inesistente
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
production_run_id=999999, outcome="ok")
|
||||
assert resp.status_code == 404
|
||||
|
||||
# run chiuso: non registra piu'
|
||||
row = (await db_session.execute(
|
||||
select(ProductionRun).where(ProductionRun.id == run["id"])
|
||||
)).scalar_one()
|
||||
row.status = "closed"
|
||||
await db_session.commit()
|
||||
resp = await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
production_run_id=run["id"], outcome="ok")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
async def test_list_filters_and_order(
|
||||
client: AsyncClient, maker_user, admin_user, measurement_tec_user, db_session,
|
||||
):
|
||||
recipe, check, note = await _setup(client, db_session, maker_user, admin_user, "TR5")
|
||||
run = await _open_run(client, measurement_tec_user, recipe, "TR5")
|
||||
a = (await _post(client, measurement_tec_user, task_id=check["id"],
|
||||
production_run_id=run["id"], outcome="ok")).json()
|
||||
b = (await _post(client, measurement_tec_user, task_id=note["id"],
|
||||
production_run_id=run["id"], outcome="done")).json()
|
||||
c = (await _post(client, measurement_tec_user, task_id=check["id"], outcome="ok")).json()
|
||||
h = auth_headers(measurement_tec_user)
|
||||
|
||||
resp = await client.get(f"/api/task-results?production_run_id={run['id']}", headers=h)
|
||||
assert [r["id"] for r in resp.json()] == [b["id"], a["id"]] # piu' recente prima
|
||||
resp = await client.get(f"/api/task-results?task_id={check['id']}", headers=h)
|
||||
assert [r["id"] for r in resp.json()] == [c["id"], a["id"]]
|
||||
resp = await client.get(
|
||||
f"/api/task-results?task_id={check['id']}&production_run_id={run['id']}", headers=h)
|
||||
assert [r["id"] for r in resp.json()] == [a["id"]]
|
||||
assert (await client.get("/api/task-results")).status_code == 401
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/task-results/attachments: l'operatore carica gli allegati (#55)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# firma PNG valida (magic bytes) piu' un po' di zeri
|
||||
PNG = bytes([0x89]) + b"PNG" + bytes([13, 10, 26, 10]) + bytes(32)
|
||||
|
||||
|
||||
async def test_attachment_upload_ok_for_measurement_tec(
|
||||
client: AsyncClient, measurement_tec_user, tmp_path,
|
||||
):
|
||||
with patch("src.backend.api.routers.task_results.settings") as mock_settings:
|
||||
mock_settings.upload_path = tmp_path
|
||||
mock_settings.max_upload_size_mb = 50
|
||||
resp = await client.post(
|
||||
"/api/task-results/attachments", headers=auth_headers(measurement_tec_user),
|
||||
files={"file": ("foto bella (1).png", PNG, "image/png")},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["name"] == "fotobella1.png"
|
||||
assert body["file_path"].startswith("task_results/")
|
||||
assert body["file_path"].endswith(".png")
|
||||
assert (tmp_path / body["file_path"]).read_bytes() == PNG
|
||||
|
||||
|
||||
async def test_attachment_upload_rejects_wrong_type_size_and_role(
|
||||
client: AsyncClient, measurement_tec_user, maker_user, tmp_path,
|
||||
):
|
||||
h = auth_headers(measurement_tec_user)
|
||||
with patch("src.backend.api.routers.task_results.settings") as mock_settings:
|
||||
mock_settings.upload_path = tmp_path
|
||||
mock_settings.max_upload_size_mb = 1
|
||||
# tipo non ammesso (HTML dichiarato come png): come upload_file, 400
|
||||
resp = await client.post(
|
||||
"/api/task-results/attachments", headers=h,
|
||||
files={"file": ("x.png", b"<html></html>", "image/png")},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
# oltre il limite
|
||||
resp = await client.post(
|
||||
"/api/task-results/attachments", headers=h,
|
||||
files={"file": ("x.png", PNG + bytes(1024 * 1024 + 1), "image/png")},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
# un Maker senza ruolo di reparto non carica
|
||||
resp = await client.post(
|
||||
"/api/task-results/attachments", headers=auth_headers(maker_user),
|
||||
files={"file": ("x.png", PNG, "image/png")},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert not any(tmp_path.rglob("*.png"))
|
||||
|
||||
|
||||
async def test_task_results_dir_is_undeletable(client: AsyncClient, admin_user):
|
||||
resp = await client.delete(
|
||||
"/api/files/task_results/2026/10/abc.png", headers=auth_headers(admin_user),
|
||||
)
|
||||
assert resp.status_code in (403, 404)
|
||||
assert resp.status_code == 403
|
||||
Reference in New Issue
Block a user