"""gitea:Adriano/TieMeasureFlow#97 - camera e modo di analisi nel task visione (lato Maker). api_client e requests sono finti: nessun servizio camera vero. """ import json import re from unittest.mock import MagicMock, patch import requests from tests.conftest import api_get_router ST1 = {"id": 4, "code": "ST-4", "name": "Banco", "camera_url": "http://acq:8765/", "camera_source": "camera:1"} ST2 = {"id": 5, "code": "ST-5", "name": "Linea", "camera_url": "http://acq:8765", "camera_source": "camera:2"} ST3 = {"id": 6, "code": "ST-6", "name": "Altra", "camera_url": "http://altro:9000", "camera_source": "x"} NO_CAMERA = {"id": 7, "code": "ST-7", "name": "Senza", "camera_url": None, "camera_source": None} SOURCES = {"sources": [ {"name": "camera:601020", "kind": "camera", "label": "Basler 601020"}, {"name": "pezzo.png", "kind": "image", "label": "immagine - pezzo.png"}, ]} def _stations(mock_api_client, stations=(ST1, ST2, ST3, NO_CAMERA)): mock_api_client.get.side_effect = api_get_router({"/api/stations": list(stations)}) def _fake(status=200, body=None): resp = MagicMock(status_code=status, text="") if body is None: resp.json.side_effect = ValueError("no json") else: resp.json.return_value = body return resp class TestCameraServizi: def test_distinct_urls_with_stations(self, logged_in_client, mock_api_client): _stations(mock_api_client) resp = logged_in_client.get("/maker/api/camera/servizi") assert resp.status_code == 200 assert resp.get_json() == {"servizi": [ {"camera_url": "http://acq:8765", "stazioni": ["ST-4 - Banco", "ST-5 - Linea"]}, {"camera_url": "http://altro:9000", "stazioni": ["ST-6 - Altra"]}, ]} def test_no_stations_with_camera(self, logged_in_client, mock_api_client): _stations(mock_api_client, [NO_CAMERA]) assert logged_in_client.get("/maker/api/camera/servizi").get_json() == {"servizi": []} class TestCameraSorgenti: def test_lists_sources(self, logged_in_client, mock_api_client): _stations(mock_api_client) with patch("blueprints.maker.http_requests.get", return_value=_fake(200, SOURCES)) as get: resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765/") assert resp.status_code == 200 assert resp.get_json() == SOURCES args, kwargs = get.call_args assert args[0] == "http://acq:8765/camera/sources" assert kwargs["timeout"] == 15 def test_url_not_a_station_service_is_400(self, logged_in_client, mock_api_client): _stations(mock_api_client) with patch("blueprints.maker.http_requests.get") as get: resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://evil:1") assert resp.status_code == 400 get.assert_not_called() def test_missing_url_is_400(self, logged_in_client, mock_api_client): _stations(mock_api_client) with patch("blueprints.maker.http_requests.get") as get: assert logged_in_client.get("/maker/api/camera/sorgenti").status_code == 400 get.assert_not_called() def test_service_down_is_502(self, logged_in_client, mock_api_client): _stations(mock_api_client) with patch("blueprints.maker.http_requests.get", side_effect=requests.ConnectionError("refused")): resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765") assert resp.status_code == 502 assert "refused" in resp.get_json()["detail"] def test_service_error_status_is_502(self, logged_in_client, mock_api_client): _stations(mock_api_client) with patch("blueprints.maker.http_requests.get", return_value=_fake(500, {"detail": "x"})): resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765") assert resp.status_code == 502 class TestTaskProxyForwardsAcquisition: def test_put_forwards_acquisition_json(self, logged_in_client, mock_api_client): acq = {"camera_url": "http://acq:8765", "camera_source": "camera:1", "modo": "in_linea"} mock_api_client.put.return_value = {"id": 11, "title": "T", "acquisition_json": acq} resp = logged_in_client.put("/maker/api/tasks/11", json={"title": "T", "acquisition_json": acq}) assert resp.status_code == 200 assert mock_api_client.put.call_args.kwargs["data"]["acquisition_json"] == acq def test_put_forwards_null(self, logged_in_client, mock_api_client): mock_api_client.put.return_value = {"id": 11, "title": "T", "acquisition_json": None} logged_in_client.put("/maker/api/tasks/11", json={"title": "T", "acquisition_json": None}) assert mock_api_client.put.call_args.kwargs["data"]["acquisition_json"] is None class TestTaskEditorAcquisitionUi: def _html(self, client, mock_api_client, task_type="camera_measure"): mock_api_client.get.side_effect = api_get_router({ "/api/recipes/3": {"id": 3, "code": "R", "name": "R"}, "/api/recipes/3/tasks": [{ "id": 11, "title": "T", "task_type": task_type, "subtasks": [], "file_path": None, "order_index": 1, "acquisition_json": {"camera_url": "http://acq:8765", "camera_source": "camera:1", "modo": "in_linea"}, }], }) return client.get("/maker/recipes/3/tasks").get_data(as_text=True) def test_section_and_fields(self, logged_in_client, mock_api_client): html = self._html(logged_in_client, mock_api_client) assert "taskAcquisition(" in html assert "js/task_acquisition.js" in html assert "Servizio di acquisizione" in html assert "Sorgente" in html assert "Analisi impulsiva" in html and "Analisi in linea" in html assert 'value="impulsiva"' in html and 'value="in_linea"' in html assert "Camera della stazione dell'operatore" in html or \ "Camera della stazione dell'operatore" in html def test_payload_carries_acquisition_json(self, logged_in_client, mock_api_client): html = self._html(logged_in_client, mock_api_client) assert "acquisition_json: acq.value" in html assert "_acquisitionPayload(this.editTaskData)" in html # il task salvato riempie lo stato di modifica assert "task.acquisition_json.camera_url" in html def test_static_js_endpoints(self): import os path = os.path.join(os.path.dirname(__file__), "..", "static", "js", "task_acquisition.js") src = open(path, encoding="utf-8").read() assert "/maker/api/camera/servizi" in src assert "/maker/api/camera/sorgenti?url=" in src