"""#97: camera e modo di analisi nel task, POST /api/vision/check senza salvataggio. Il worker e l'acquisitore non girano nei test: si fingono (fixture di test_vision_dxf_compare e test_vision_station_worker). """ import pytest from sqlalchemy import func, select from src.backend.config import settings from src.backend.models.orm.measurement import Measurement from src.backend.models.orm.task import RecipeSubtask, RecipeTask from src.backend.models.orm.vision import VisionResult from src.backend.tests.conftest import auth_headers, create_test_recipe from src.backend.tests.test_vision_copy_on_write import ( _current_task_and_subtask, _seed_measurement, ) from src.backend.tests.test_vision_dxf_compare import ( # noqa: F401 (fixture acquisitore) CAL, CAMERA_URL, acquisitore, redirect_uploads, ) from src.backend.tests.test_vision_execute import an_image from src.backend.tests.test_vision_station_worker import ( # noqa: F401 (fixture worker_calls) _camera_task, _execute, _station_with_run, worker_calls, ) TASK_CAMERA = "http://10.0.0.50:8300" ACQ = {"camera_url": TASK_CAMERA, "camera_source": "camera:601020", "modo": "in_linea"} async def _task_with_acquisition(db_session, user_id, acquisition=ACQ): recipe, task = await _camera_task(db_session, user_id) task.acquisition_json = acquisition await db_session.commit() return recipe, task # --- modello e API del task --------------------------------------------------- @pytest.mark.asyncio async def test_create_task_stores_and_normalizes_acquisition(client, db_session, maker_user): recipe = await create_test_recipe(db_session, maker_user.id) resp = await client.post( f"/api/recipes/{recipe.id}/tasks", headers=auth_headers(maker_user), json={ "title": "Camera", "task_type": "camera_measure", "acquisition_json": {"camera_url": " http://h:1/ ", "camera_source": " camera:1 "}, }, ) assert resp.status_code == 201, resp.text body = resp.json() # url normalizzato come nelle stazioni, sorgente senza spazi, modo di default expected = {"camera_url": "http://h:1", "camera_source": "camera:1", "modo": "impulsiva"} assert body["acquisition_json"] == expected db_session.expunge_all() task = (await db_session.execute( select(RecipeTask).where(RecipeTask.id == body["id"]) )).scalar_one() assert task.acquisition_json == expected # sul DB, non solo nella risposta got = await client.get(f"/api/tasks/{body['id']}", headers=auth_headers(maker_user)) assert got.json()["acquisition_json"] == expected @pytest.mark.asyncio async def test_task_without_acquisition_is_null(client, db_session, maker_user): recipe = await create_test_recipe(db_session, maker_user.id) resp = await client.get(f"/api/recipes/{recipe.id}/tasks", headers=auth_headers(maker_user)) assert resp.json()[0]["acquisition_json"] is None @pytest.mark.asyncio @pytest.mark.parametrize("bad", [ {"camera_url": "ftp://h", "camera_source": "camera:1"}, {"camera_url": " ", "camera_source": "camera:1"}, {"camera_url": "http://h", "camera_source": " "}, {"camera_url": "http://h", "camera_source": "x" * 201}, {"camera_url": "http://h", "camera_source": "camera:1", "modo": "continua"}, {"camera_source": "camera:1"}, ]) async def test_invalid_acquisition_is_422(client, db_session, maker_user, bad): recipe = await create_test_recipe(db_session, maker_user.id) resp = await client.post( f"/api/recipes/{recipe.id}/tasks", headers=auth_headers(maker_user), json={"title": "Camera", "task_type": "camera_measure", "acquisition_json": bad}, ) assert resp.status_code == 422, resp.text @pytest.mark.asyncio async def test_update_sets_and_clears_acquisition(client, db_session, maker_user): recipe = await create_test_recipe(db_session, maker_user.id) _, task, _ = await _current_task_and_subtask(db_session, recipe.id) resp = await client.put( f"/api/tasks/{task.id}", headers=auth_headers(maker_user), json={"acquisition_json": ACQ}, ) assert resp.status_code == 200, resp.text assert resp.json()["acquisition_json"] == ACQ resp = await client.put( f"/api/tasks/{task.id}", headers=auth_headers(maker_user), json={"acquisition_json": None}, ) assert resp.status_code == 200 assert resp.json()["acquisition_json"] is None # null = camera della stazione @pytest.mark.asyncio async def test_copy_on_write_preserves_acquisition(client, db_session, maker_user): recipe = await create_test_recipe(db_session, maker_user.id) _, task, _ = await _current_task_and_subtask(db_session, recipe.id) task.acquisition_json = ACQ await db_session.flush() await _seed_measurement(db_session, recipe.id, maker_user.id) db_session.expunge_all() resp = await client.put( f"/api/recipes/{recipe.id}", headers=auth_headers(maker_user), json={"change_notes": "forza il copy-on-write"}, ) assert resp.status_code == 200 db_session.expunge_all() version, new_task, _ = await _current_task_and_subtask(db_session, recipe.id) assert version.version_number == 2 assert new_task.acquisition_json == ACQ # --- calibrazione dalla camera del task ---------------------------------------- @pytest.mark.asyncio async def test_execute_reads_calibration_from_the_task_camera( client, db_session, measurement_tec_user, worker_calls, acquisitore, ): recipe, task = await _task_with_acquisition(db_session, measurement_tec_user.id) # stazione con camera propria: la camera del task prevale station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None) station.camera_url, station.camera_source = CAMERA_URL, "camera:A1" await db_session.commit() await _execute(client, measurement_tec_user, task, run) assert acquisitore["urls"] == [(f"{TASK_CAMERA}/camera/calibration", {"source": "camera:601020"})] assert worker_calls[0]["calibration"] == {"px_per_mm": CAL["px_per_mm"]} result = (await db_session.execute(select(VisionResult))).scalars().one() assert result.calibration_snapshot["px_per_mm"] == CAL["px_per_mm"] @pytest.mark.asyncio async def test_execute_task_camera_works_without_a_run( client, db_session, measurement_tec_user, worker_calls, acquisitore, ): _, task = await _task_with_acquisition(db_session, measurement_tec_user.id) await _execute(client, measurement_tec_user, task) # niente sessione di produzione assert acquisitore["urls"][0][0] == f"{TASK_CAMERA}/camera/calibration" assert "calibration" in worker_calls[0] @pytest.mark.asyncio async def test_execute_falls_back_to_the_station_camera( client, db_session, measurement_tec_user, worker_calls, acquisitore, ): recipe, task = await _camera_task(db_session, measurement_tec_user.id) # acquisition_json null station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None) station.camera_url, station.camera_source = CAMERA_URL, "camera:A1" await db_session.commit() await _execute(client, measurement_tec_user, task, run) assert acquisitore["urls"] == [(f"{CAMERA_URL}/camera/calibration", {"source": "camera:A1"})] assert worker_calls[0]["calibration"] == {"px_per_mm": CAL["px_per_mm"]} # --- POST /api/vision/check ---------------------------------------------------- async def _check(client, user, task, run=None): data = {"task_id": str(task.id)} if run is not None: data["production_run_id"] = str(run.id) return await client.post( "/api/vision/check", headers=auth_headers(user), files={"image": ("p.png", an_image(), "image/png")}, data=data, ) async def _counts(db_session): return tuple([ (await db_session.execute(select(func.count()).select_from(model))).scalar_one() for model in (Measurement, VisionResult) ]) @pytest.mark.asyncio async def test_check_saves_nothing_and_reports_pass_fail( client, db_session, measurement_tec_user, monkeypatch, acquisitore, ): _, task = await _task_with_acquisition(db_session, measurement_tec_user.id) seen = {} async def fake_call(image_bytes, graph, **kwargs): seen.update(kwargs) return {"outputs": {"t1.diameter": 11.0}, "failures": [{"tool_name": "t"}], "overlay": [{"kind": "line"}], "outcome": "fail", "engine_version": "e" * 40, "duration_ms": 3} monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call) before = await _counts(db_session) resp = await _check(client, measurement_tec_user, task) assert resp.status_code == 200, resp.text body = resp.json() quote = (await db_session.execute( select(RecipeSubtask).where(RecipeSubtask.task_id == task.id) )).scalars().one() assert body == { "outcome": "fail", "overlay": [{"kind": "line"}], "failures": [{"tool_name": "t"}], "measurements": [{"subtask_id": quote.id, "value": 11.0, "pass_fail": "fail"}], "calibration": "ok", } assert seen["calibration"] == {"px_per_mm": CAL["px_per_mm"]} # stessa calibrazione di execute assert await _counts(db_session) == before # niente misure ne' VisionResult assert not list(settings.upload_path.rglob("*.overlay.json")) # ne' file @pytest.mark.asyncio @pytest.mark.parametrize("value,expected", [(10.0, "pass"), (10.4, "warning"), (9.0, "fail")]) async def test_check_pass_fail_levels( client, db_session, measurement_tec_user, monkeypatch, value, expected, ): _, task = await _camera_task(db_session, measurement_tec_user.id) async def fake_call(image_bytes, graph, **kwargs): return {"outputs": {"t1.diameter": value}, "failures": [], "engine_version": "e" * 40} monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call) resp = await _check(client, measurement_tec_user, task) assert resp.json()["measurements"][0]["pass_fail"] == expected assert resp.json()["calibration"] == "no_camera" # senza camera ne' stazione @pytest.mark.asyncio async def test_check_calibration_reasons( client, db_session, measurement_tec_user, worker_calls, acquisitore, ): _, task = await _task_with_acquisition(db_session, measurement_tec_user.id) acquisitore["cal"] = None assert (await _check(client, measurement_tec_user, task)).json()["calibration"] == "not_calibrated" acquisitore["cal"] = "down" assert (await _check(client, measurement_tec_user, task)).json()["calibration"] == "unreachable" assert "calibration" not in worker_calls[-1] # senza calibrazione la ricetta usa la sua @pytest.mark.asyncio async def test_check_omits_quotes_the_graph_did_not_produce( client, db_session, measurement_tec_user, monkeypatch, ): _, task = await _camera_task(db_session, measurement_tec_user.id) async def fake_call(image_bytes, graph, **kwargs): return {"outputs": {}, "failures": [{"tool_name": "blob"}], "engine_version": "e" * 40} monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call) resp = await _check(client, measurement_tec_user, task) assert resp.status_code == 200 # un'anteprima non da' 422 a ogni frame assert resp.json()["measurements"] == [] assert resp.json()["failures"] == [{"tool_name": "blob"}] @pytest.mark.asyncio async def test_check_422_without_graph_or_quotes_or_vision_type( client, db_session, measurement_tec_user, worker_calls, ): _, task = await _camera_task(db_session, measurement_tec_user.id) task.vision_json = None await db_session.commit() assert (await _check(client, measurement_tec_user, task)).status_code == 422 task.vision_json = {"schema_version": 1, "name": "p", "tools": []} quote = (await db_session.execute( select(RecipeSubtask).where(RecipeSubtask.task_id == task.id) )).scalars().one() quote.vision_output = None await db_session.commit() assert (await _check(client, measurement_tec_user, task)).status_code == 422 task.task_type = "measure" await db_session.commit() assert (await _check(client, measurement_tec_user, task)).status_code == 422 @pytest.mark.asyncio async def test_check_with_run_uses_station_worker_and_station_camera( client, db_session, measurement_tec_user, worker_calls, acquisitore, ): recipe, task = await _camera_task(db_session, measurement_tec_user.id) station, run = await _station_with_run( db_session, measurement_tec_user, recipe, "http://10.0.0.7:8100") station.camera_url, station.camera_source = CAMERA_URL, "camera:A1" await db_session.commit() before = await _counts(db_session) resp = await _check(client, measurement_tec_user, task, run) assert resp.status_code == 200, resp.text assert worker_calls[0]["worker_url"] == "http://10.0.0.7:8100" assert acquisitore["urls"][0][0] == f"{CAMERA_URL}/camera/calibration" assert await _counts(db_session) == before @pytest.mark.asyncio async def test_check_requires_login(client, db_session, measurement_tec_user, worker_calls): _, task = await _camera_task(db_session, measurement_tec_user.id) resp = await client.post( "/api/vision/check", files={"image": ("p.png", an_image(), "image/png")}, data={"task_id": str(task.id)}, ) assert resp.status_code in (401, 403)