211 lines
9.3 KiB
Python
211 lines
9.3 KiB
Python
"""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()
|
|
# sessione condivisa col client: la collezione version.tasks va riletta
|
|
db_session.expire_all()
|
|
|
|
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")
|