Adriano
9b8e729927
Due proxy Flask (/source e /files, byte) e wizard che all'init legge la ricetta esistente: etichetta del bottone dinamica, apertura con banda, soglia, polarita', mm/px, DXF e immagine di riferimento; risposte tardive non sovrascrivono le scelte manuali. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
360 lines
17 KiB
Python
360 lines
17 KiB
Python
"""gitea:Adriano/TieMeasureFlow#83 - ricetta DXF compare da camera: rotte Flask nuove.
|
|
|
|
api_client e requests sono finti: backend (B3) e acquisitore (C1) rispondono
|
|
come da contratto del piano.
|
|
"""
|
|
import io
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import requests
|
|
|
|
from tests.conftest import api_get_router
|
|
|
|
STATION = {
|
|
"id": 4, "code": "ST-4", "name": "Banco",
|
|
"camera_url": "http://acq:8765/", "camera_source": "camera:folder",
|
|
}
|
|
NO_CAMERA = {"id": 5, "code": "ST-5", "name": "Senza", "camera_url": None, "camera_source": None}
|
|
PNG = b"\x89PNG\r\n\x1a\n fake"
|
|
CAL = {
|
|
"source": "camera:folder", "px_per_mm": 20.03, "mm_per_px": 0.0499, "rms_px": 0.2,
|
|
"n_corners": 54, "cols": 9, "rows": 6, "square_mm": 5.0,
|
|
"image_width": 2448, "image_height": 2048, "scale_spread_pct": 0.3,
|
|
"calibrated_at": "2026-10-23T10:00:00",
|
|
}
|
|
|
|
|
|
def _fake(status=200, content=PNG, json_body=None):
|
|
resp = MagicMock(status_code=status, content=content, text=content.decode("latin-1"))
|
|
if json_body is None:
|
|
resp.json.side_effect = ValueError("no json")
|
|
else:
|
|
resp.json.return_value = json_body
|
|
return resp
|
|
|
|
|
|
def _stations(mock_api_client, stations=(STATION, NO_CAMERA)):
|
|
mock_api_client.get.side_effect = api_get_router({"/api/stations": list(stations)})
|
|
|
|
|
|
class TestStationList:
|
|
def test_only_stations_with_camera(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
resp = logged_in_client.get("/maker/api/stations")
|
|
assert resp.status_code == 200
|
|
assert [s["id"] for s in resp.get_json()] == [4]
|
|
|
|
def test_non_admin_list_falls_back_to_current_station(
|
|
self, logged_in_client, mock_api_client, monkeypatch):
|
|
# /api/stations e' solo admin: il Maker vede la stazione corrente (by-code)
|
|
monkeypatch.setattr("blueprints.maker.Config.STATION_CODE", "ST-4")
|
|
mock_api_client.get.side_effect = api_get_router(
|
|
{"/api/stations": {"error": True, "status_code": 403, "detail": "no"}})
|
|
mock_api_client.get_station_by_code.return_value = STATION
|
|
resp = logged_in_client.get("/maker/api/stations")
|
|
assert [s["id"] for s in resp.get_json()] == [4]
|
|
mock_api_client.get_station_by_code.assert_called_once_with("ST-4")
|
|
|
|
def test_requires_maker_or_admin(self, logged_in_client, mock_api_client):
|
|
with logged_in_client.session_transaction() as sess:
|
|
sess["user"] = {**sess["user"], "roles": ["MeasurementTec"], "is_admin": False}
|
|
assert logged_in_client.get("/maker/api/stations").status_code == 403
|
|
|
|
|
|
class TestStationCameraGrab:
|
|
def test_png_pass_through(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
with patch("blueprints.maker.http_requests.post", return_value=_fake()) as post:
|
|
resp = logged_in_client.post("/maker/api/stations/4/camera/grab")
|
|
assert resp.status_code == 200
|
|
assert resp.mimetype == "image/png"
|
|
assert resp.data == PNG
|
|
args, kwargs = post.call_args
|
|
assert args[0] == "http://acq:8765/camera/grab" # slash finale tolto
|
|
assert kwargs["json"] == {"source": "camera:folder", "trigger": "continuo", "timeout_ms": 5000}
|
|
|
|
def test_get_also_works(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
with patch("blueprints.maker.http_requests.post", return_value=_fake()):
|
|
assert logged_in_client.get("/maker/api/stations/4/camera/grab").status_code == 200
|
|
|
|
def test_station_without_camera_is_409(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
with patch("blueprints.maker.http_requests.post") as post:
|
|
resp = logged_in_client.post("/maker/api/stations/5/camera/grab")
|
|
assert resp.status_code == 409
|
|
post.assert_not_called()
|
|
|
|
def test_unknown_station_is_409(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
assert logged_in_client.post("/maker/api/stations/99/camera/grab").status_code == 409
|
|
|
|
def test_service_down_is_502(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
with patch("blueprints.maker.http_requests.post",
|
|
side_effect=requests.ConnectionError("refused")):
|
|
resp = logged_in_client.post("/maker/api/stations/4/camera/grab")
|
|
assert resp.status_code == 502
|
|
assert resp.get_json()["error"]
|
|
|
|
def test_service_503_is_502_with_detail(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
fake = _fake(503, b"{}", {"detail": "camera occupata"})
|
|
with patch("blueprints.maker.http_requests.post", return_value=fake):
|
|
resp = logged_in_client.post("/maker/api/stations/4/camera/grab")
|
|
assert resp.status_code == 502
|
|
assert "camera occupata" in resp.get_json()["detail"]
|
|
|
|
|
|
class TestStationCameraCalibration:
|
|
def _get(self, client, fake=None, exc=None):
|
|
with patch("blueprints.maker.http_requests.get", return_value=fake,
|
|
side_effect=exc) as get:
|
|
resp = client.get("/maker/api/stations/4/camera/calibration")
|
|
return resp, get
|
|
|
|
def test_calibrated(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
resp, get = self._get(logged_in_client, _fake(200, b"{}", CAL))
|
|
body = resp.get_json()
|
|
assert resp.status_code == 200
|
|
assert body["state"] == "calibrated"
|
|
assert body["calibration"]["px_per_mm"] == 20.03
|
|
assert body["calibration_url"] == "http://acq:8765/acquisizione"
|
|
args, kwargs = get.call_args
|
|
assert args[0] == "http://acq:8765/camera/calibration"
|
|
assert kwargs["params"] == {"source": "camera:folder"}
|
|
|
|
def test_404_is_uncalibrated(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
resp, _ = self._get(logged_in_client, _fake(404, b"{}", {"detail": "no"}))
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["state"] == "uncalibrated"
|
|
assert resp.get_json()["calibration_url"] == "http://acq:8765/acquisizione"
|
|
|
|
def test_unreachable(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
resp, _ = self._get(logged_in_client, exc=requests.ConnectionError("down"))
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["state"] == "unreachable"
|
|
|
|
def test_server_error_is_unreachable(self, logged_in_client, mock_api_client):
|
|
_stations(mock_api_client)
|
|
resp, _ = self._get(logged_in_client, _fake(500, b"boom"))
|
|
assert resp.get_json()["state"] == "unreachable"
|
|
|
|
def test_station_without_camera_is_409(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/stations/5/camera/calibration")
|
|
assert resp.status_code == 409
|
|
get.assert_not_called()
|
|
|
|
|
|
class TestDxfCompareProxy:
|
|
URL = "/maker/api/vision/tasks/7/recipe/dxf-compare"
|
|
OK = {
|
|
"result": {"outputs": {"dxf.points_in_pct": 99.1}, "overlay": [], "outcome": "pass"},
|
|
"saved": False, "recipe_dir": None, "outputs": None,
|
|
}
|
|
|
|
def _post(self, client, **form):
|
|
data = {"image": (io.BytesIO(PNG), "p.png"), "dxf": (io.BytesIO(b"0\nEOF"), "d.dxf"), **form}
|
|
return client.post(self.URL, data=data, content_type="multipart/form-data")
|
|
|
|
def test_forwards_files_and_fields(self, logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = self.OK
|
|
resp = self._post(logged_in_client, station_id="4", band_mm="0.1",
|
|
min_points_pct="95", polarity="dark", save="false", px_per_mm="")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["result"]["outputs"]["dxf.points_in_pct"] == 99.1
|
|
args, kwargs = mock_api_client.post.call_args
|
|
assert args[0] == "/api/vision/tasks/7/recipe/dxf-compare"
|
|
# campi del contratto B3; px_per_mm vuoto non si inoltra
|
|
assert kwargs["data"] == {
|
|
"station_id": "4", "band_mm": "0.1", "min_points_pct": "95",
|
|
"polarity": "dark", "save": "false",
|
|
}
|
|
assert set(kwargs["files"]) == {"image", "dxf"}
|
|
assert kwargs["files"]["image"][1] == PNG
|
|
|
|
def test_manual_px_per_mm_forwarded(self, logged_in_client, mock_api_client):
|
|
# #98: scala a mano dal wizard (immagine da file): inoltrata al backend
|
|
mock_api_client.post.return_value = self.OK
|
|
self._post(logged_in_client, station_id="4", px_per_mm="71.1237")
|
|
assert mock_api_client.post.call_args.kwargs["data"]["px_per_mm"] == "71.1237"
|
|
|
|
def test_save_true_forwarded(self, logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {**self.OK, "saved": True, "recipe_dir": "vision/recipes/7/x"}
|
|
resp = self._post(logged_in_client, save="true")
|
|
assert resp.get_json()["saved"] is True
|
|
assert mock_api_client.post.call_args.kwargs["data"]["save"] == "true"
|
|
|
|
def test_missing_files_is_400(self, logged_in_client, mock_api_client):
|
|
resp = logged_in_client.post(self.URL, data={"station_id": "4"},
|
|
content_type="multipart/form-data")
|
|
assert resp.status_code == 400
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
def test_backend_422_propagates(self, logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {
|
|
"error": True, "status_code": 422, "detail": "stazione non calibrata"}
|
|
resp = self._post(logged_in_client)
|
|
assert resp.status_code == 422
|
|
assert resp.get_json()["detail"] == "stazione non calibrata"
|
|
|
|
def test_requires_maker_role(self, logged_in_client, mock_api_client):
|
|
with logged_in_client.session_transaction() as sess:
|
|
sess["user"] = {**sess["user"], "roles": ["MeasurementTec"]}
|
|
assert self._post(logged_in_client).status_code == 403
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
|
|
class TestDxfPreviewProxy:
|
|
"""#101: anteprima del DXF scelto nel wizard (proxy verso il backend)."""
|
|
URL = "/maker/api/vision/dxf-preview"
|
|
OK = {
|
|
"contorni": [[[0.0, -5.0], [10.0, -5.0], [10.0, 0.0], [0.0, 0.0]]],
|
|
"bbox": {"xmin": 0.0, "ymin": -5.0, "xmax": 10.0, "ymax": 0.0},
|
|
"larghezza_mm": 10.0, "altezza_mm": 5.0, "entita": 1, "punti": 4,
|
|
}
|
|
|
|
def _post(self, client, with_file=True):
|
|
data = {"dxf": (io.BytesIO(b"0\nEOF"), "d.dxf")} if with_file else {}
|
|
return client.post(self.URL, data=data, content_type="multipart/form-data")
|
|
|
|
def test_forwards_file(self, logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = self.OK
|
|
resp = self._post(logged_in_client)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["larghezza_mm"] == 10.0
|
|
args, kwargs = mock_api_client.post.call_args
|
|
assert args[0] == "/api/vision/dxf-preview"
|
|
assert set(kwargs["files"]) == {"dxf"}
|
|
assert kwargs["files"]["dxf"][0] == "d.dxf"
|
|
assert kwargs["files"]["dxf"][1] == b"0\nEOF"
|
|
|
|
def test_missing_file_is_400(self, logged_in_client, mock_api_client):
|
|
resp = self._post(logged_in_client, with_file=False)
|
|
assert resp.status_code == 400
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
def test_backend_422_propagates(self, logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {
|
|
"error": True, "status_code": 422, "detail": "DXF illeggibile: x"}
|
|
resp = self._post(logged_in_client)
|
|
assert resp.status_code == 422
|
|
assert resp.get_json()["detail"] == "DXF illeggibile: x"
|
|
|
|
def test_requires_maker_role(self, logged_in_client, mock_api_client):
|
|
with logged_in_client.session_transaction() as sess:
|
|
sess["user"] = {**sess["user"], "roles": ["MeasurementTec"]}
|
|
assert self._post(logged_in_client).status_code == 403
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
|
|
class TestDxfSourceProxy:
|
|
"""#105: parametri della ricetta DXF gia' salvata (proxy verso il backend)."""
|
|
URL = "/maker/api/vision/tasks/7/recipe/dxf-compare/source"
|
|
OK = {"band_mm": 0.3, "min_points_pct": 95.0, "polarity": "dark", "px_per_mm": 71.12,
|
|
"dxf": "disegno.dxf", "image": "riferimento.bmp"}
|
|
|
|
def test_forwards_body(self, logged_in_client, mock_api_client):
|
|
mock_api_client.get.return_value = self.OK
|
|
resp = logged_in_client.get(self.URL)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == self.OK
|
|
assert mock_api_client.get.call_args.args[0] == "/api/vision/tasks/7/recipe/dxf-compare/source"
|
|
|
|
def test_404_propagates(self, logged_in_client, mock_api_client):
|
|
mock_api_client.get.return_value = {
|
|
"error": True, "status_code": 404, "detail": "nessuna ricetta da DXF su questo task"}
|
|
resp = logged_in_client.get(self.URL)
|
|
assert resp.status_code == 404
|
|
assert resp.get_json()["detail"] == "nessuna ricetta da DXF su questo task"
|
|
|
|
def test_requires_maker_role(self, logged_in_client, mock_api_client):
|
|
with logged_in_client.session_transaction() as sess:
|
|
sess["user"] = {**sess["user"], "roles": ["MeasurementTec"]}
|
|
assert logged_in_client.get(self.URL).status_code == 403
|
|
mock_api_client.get.assert_not_called()
|
|
|
|
|
|
class TestRecipeFileProxy:
|
|
"""#105: file della cartella ricetta (DXF, immagine di riferimento) passati come byte."""
|
|
URL = "/maker/api/vision/tasks/7/recipe/files/riferimento.bmp"
|
|
BMP = b"BM 1 fake bmp"
|
|
|
|
def _fake_file(self, status=200, content=BMP, ctype="image/bmp"):
|
|
resp = MagicMock(status_code=status, content=content)
|
|
resp.headers = {"content-type": ctype}
|
|
return resp
|
|
|
|
def test_bytes_and_content_type_pass_through(self, logged_in_client, mock_api_client):
|
|
with patch("blueprints.maker.http_requests.get", return_value=self._fake_file()) as get:
|
|
resp = logged_in_client.get(self.URL)
|
|
assert resp.status_code == 200
|
|
assert resp.data == self.BMP
|
|
assert resp.mimetype == "image/bmp"
|
|
args, kwargs = get.call_args
|
|
assert args[0].endswith("/api/vision/tasks/7/recipe/files/riferimento.bmp")
|
|
assert kwargs["headers"]["X-API-Key"] == "test-api-key-12345"
|
|
|
|
def test_backend_404_propagates(self, logged_in_client, mock_api_client):
|
|
body = b'{"detail":"file non trovato"}'
|
|
with patch("blueprints.maker.http_requests.get",
|
|
return_value=self._fake_file(404, body, "application/json")):
|
|
resp = logged_in_client.get(self.URL)
|
|
assert resp.status_code == 404
|
|
assert resp.get_json()["detail"] == "file non trovato"
|
|
|
|
def test_unsafe_path_is_404_without_calling_backend(self, logged_in_client, mock_api_client):
|
|
with patch("blueprints.maker.http_requests.get") as get:
|
|
for path in ("a/../../x", "..%5Cx", "%2e%2e/x"):
|
|
assert logged_in_client.get(f"/maker/api/vision/tasks/7/recipe/files/{path}").status_code == 404
|
|
get.assert_not_called()
|
|
|
|
def test_service_down_is_502(self, logged_in_client, mock_api_client):
|
|
with patch("blueprints.maker.http_requests.get", side_effect=requests.ConnectionError("giu")):
|
|
assert logged_in_client.get(self.URL).status_code == 502
|
|
|
|
def test_requires_maker_role(self, logged_in_client, mock_api_client):
|
|
with logged_in_client.session_transaction() as sess:
|
|
sess["user"] = {**sess["user"], "roles": ["MeasurementTec"]}
|
|
with patch("blueprints.maker.http_requests.get") as get:
|
|
assert logged_in_client.get(self.URL).status_code == 403
|
|
get.assert_not_called()
|
|
|
|
|
|
class TestPages:
|
|
def test_stations_page_has_calibration_cell(self, logged_in_client):
|
|
with patch("blueprints.admin.api_client") as admin_api:
|
|
admin_api.get.side_effect = api_get_router({"/api/stations": [STATION], "/api/recipes": []})
|
|
html = logged_in_client.get("/admin/stations").get_data(as_text=True)
|
|
assert "js/station_calibration.js" in html
|
|
assert "stationCalibrationCell(" in html
|
|
# niente piu' pannello di calibrazione nella pagina: solo il link all'acquisitore
|
|
assert "calibration-panel" not in html
|
|
|
|
def _editor(self, client, mock_api_client, task_type):
|
|
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,
|
|
}],
|
|
})
|
|
return client.get("/maker/recipes/3/tasks").get_data(as_text=True)
|
|
|
|
def test_editor_has_dxf_wizard(self, logged_in_client, mock_api_client):
|
|
html = self._editor(logged_in_client, mock_api_client, "dxf_compare")
|
|
assert "js/dxf_recipe_wizard.js" in html
|
|
assert "dxfRecipeWizard(" in html
|
|
assert "@dxf-recipe-saved" in html
|
|
assert 'x-model="mmPx"' in html # #98: campo mm/px a mano
|
|
assert "dxfPreviewD()" in html and "view-box.camel" in html # #101: anteprima SVG del DXF
|
|
assert "imageSource === 'file'" in html and "zoomBy(" in html and "fit()" in html # #104: mm/px solo da file, zoom e Fit
|
|
assert "imagePreviewUrl" in html and "imagePreviewSummary" in html # #102: anteprima immagine
|
|
assert "existing ? msgs.editLabel : msgs.createLabel" in html and "Modifica ricetta da DXF" in html # #105: etichetta dinamica
|
|
assert "Pezzo scuro su fondo chiaro" in html or "Dark part on light" in html
|