87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""gitea:Adriano/TieMeasureFlow#46 - operator side of the vision tasks.
|
|
|
|
The backend of contract B is not merged yet: api_client is mocked.
|
|
"""
|
|
import io
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from tests.conftest import api_get_router
|
|
|
|
TASK = {
|
|
"id": 12, "recipe_id": 3, "version_id": 5, "task_type": "camera_measure",
|
|
"title": "Foto", "directive": None, "description": None, "file_path": None,
|
|
"file_type": None, "annotations_json": None, "subtasks": [],
|
|
}
|
|
RECIPE = {
|
|
"id": 3, "code": "REC-1", "name": "Ricetta",
|
|
"measurement_interval_minutes": 30, "allow_manual_input": True,
|
|
"current_version": {"id": 5, "version_number": 1},
|
|
}
|
|
|
|
|
|
def test_execute_page_has_acquire_input(logged_in_client, mock_api_client, monkeypatch):
|
|
"""camera_measure screen offers the file/camera input and the acquire entry point."""
|
|
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
|
|
mock_api_client.get_station_recipes.return_value = {"items": [{"id": 3}]}
|
|
mock_api_client.get.side_effect = api_get_router({
|
|
"/api/tasks/12": TASK,
|
|
"/api/recipes/3/tasks": [{"id": 12, "order_index": 0}],
|
|
"/api/recipes/3": RECIPE,
|
|
})
|
|
html = logged_in_client.get("/measure/execute/12").get_data(as_text=True)
|
|
assert 'capture="environment"' in html
|
|
assert 'accept="image/*"' in html
|
|
assert "visionExecute()" in html
|
|
assert "/measure/api/vision/execute" in html
|
|
|
|
|
|
def test_vision_execute_proxy_forwards_multipart(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {
|
|
"vision_result_id": 7, "outcome": "ok", "overlay": [], "failures": [],
|
|
"measurements": [{"subtask_id": 1, "value": 9.9, "pass_fail": "pass"}],
|
|
}
|
|
resp = logged_in_client.post(
|
|
"/measure/api/vision/execute",
|
|
data={
|
|
"task_id": "12", "lot_number": "L1", "serial_number": "S1",
|
|
"production_run_id": "4",
|
|
"image": (io.BytesIO(b"\xff\xd8\xff fake"), "shot.jpg", "image/jpeg"),
|
|
},
|
|
content_type="multipart/form-data",
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["vision_result_id"] == 7
|
|
args, kwargs = mock_api_client.post.call_args
|
|
assert args[0] == "/api/vision/execute"
|
|
assert kwargs["data"]["task_id"] == "12"
|
|
assert kwargs["data"]["lot_number"] == "L1"
|
|
assert kwargs["data"]["production_run_id"] == "4"
|
|
name, content, mime = kwargs["files"]["image"]
|
|
assert name == "shot.jpg" and mime == "image/jpeg" and content.startswith(b"\xff\xd8")
|
|
|
|
|
|
def test_vision_execute_proxy_requires_image(logged_in_client, mock_api_client):
|
|
resp = logged_in_client.post("/measure/api/vision/execute", data={"task_id": "12"})
|
|
assert resp.status_code == 400
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
|
|
def test_vision_execute_proxy_relays_backend_error(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {"error": True, "status_code": 422, "detail": "bad"}
|
|
resp = logged_in_client.post(
|
|
"/measure/api/vision/execute",
|
|
data={"task_id": "12", "image": (io.BytesIO(b"x"), "a.jpg", "image/jpeg")},
|
|
content_type="multipart/form-data",
|
|
)
|
|
assert resp.status_code == 422
|
|
assert resp.get_json()["detail"] == "bad"
|
|
|
|
|
|
def test_vision_result_image_proxy(logged_in_client):
|
|
fake = MagicMock(status_code=200, content=b"IMG", headers={"content-type": "image/jpeg"})
|
|
with patch("blueprints.measure.http_requests.get", return_value=fake) as g:
|
|
resp = logged_in_client.get("/measure/api/vision/results/7/image")
|
|
assert resp.status_code == 200 and resp.data == b"IMG"
|
|
assert g.call_args[0][0].endswith("/api/vision/results/7/image")
|
|
assert g.call_args[1]["headers"]["X-API-Key"] == "test-api-key-12345"
|