240 lines
11 KiB
Python
240 lines
11 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 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 "Pezzo scuro su fondo chiaro" in html or "Dark part on light" in html
|