Adriano
390d6210d0
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
328 lines
13 KiB
Python
328 lines
13 KiB
Python
"""Tests for the admin Stations management UI and proxy endpoints."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_admin_api():
|
|
"""Patch api_client used by the admin blueprint."""
|
|
mock = MagicMock()
|
|
with patch("blueprints.admin.api_client", mock):
|
|
yield mock
|
|
|
|
|
|
def test_station_list_page_requires_admin(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = []
|
|
resp = logged_in_client.get("/admin/stations")
|
|
assert resp.status_code == 200
|
|
assert b"Gestione Stazioni" in resp.data or b"stations" in resp.data.lower()
|
|
|
|
|
|
def test_station_list_page_calls_correct_endpoints(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.side_effect = [
|
|
[{"id": 1, "code": "ST-001", "name": "Linea A", "location": None,
|
|
"notes": None, "active": True, "created_by": 1, "created_at": "2026-04-25T10:00:00"}],
|
|
[{"id": 1, "code": "REC-001", "name": "Test Recipe", "active": True}],
|
|
]
|
|
resp = logged_in_client.get("/admin/stations")
|
|
assert resp.status_code == 200
|
|
calls = [c.args[0] for c in mock_admin_api.get.call_args_list]
|
|
assert "/api/stations" in calls
|
|
assert "/api/recipes" in calls
|
|
|
|
|
|
def test_create_station_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.post.return_value = {
|
|
"id": 5, "code": "ST-005", "name": "Nuova",
|
|
"location": None, "notes": None, "active": True,
|
|
"created_by": 1, "created_at": "2026-04-25T10:00:00",
|
|
}
|
|
resp = logged_in_client.post(
|
|
"/admin/api/stations",
|
|
json={"code": "ST-005", "name": "Nuova", "active": True},
|
|
)
|
|
assert resp.status_code == 201
|
|
body = resp.get_json()
|
|
assert body["code"] == "ST-005"
|
|
mock_admin_api.post.assert_called_once_with(
|
|
"/api/stations",
|
|
data={"code": "ST-005", "name": "Nuova", "active": True},
|
|
)
|
|
|
|
|
|
def test_create_station_propagates_error(logged_in_client, mock_admin_api):
|
|
mock_admin_api.post.return_value = {
|
|
"error": True, "status_code": 409, "detail": "code already exists",
|
|
}
|
|
resp = logged_in_client.post(
|
|
"/admin/api/stations", json={"code": "ST-001", "name": "x"},
|
|
)
|
|
assert resp.status_code == 409
|
|
assert resp.get_json()["detail"] == "code already exists"
|
|
|
|
|
|
def test_update_station_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.put.return_value = {
|
|
"id": 3, "code": "ST-003", "name": "Aggiornata",
|
|
"location": "Reparto B", "notes": None, "active": False,
|
|
"created_by": 1, "created_at": "2026-04-25T10:00:00",
|
|
}
|
|
resp = logged_in_client.put(
|
|
"/admin/api/stations/3",
|
|
json={"name": "Aggiornata", "location": "Reparto B", "active": False},
|
|
)
|
|
assert resp.status_code == 200
|
|
mock_admin_api.put.assert_called_once_with(
|
|
"/api/stations/3",
|
|
data={"name": "Aggiornata", "location": "Reparto B", "active": False},
|
|
)
|
|
|
|
|
|
def test_delete_station_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.delete.return_value = {}
|
|
resp = logged_in_client.delete("/admin/api/stations/7")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"deleted": True}
|
|
mock_admin_api.delete.assert_called_once_with("/api/stations/7")
|
|
|
|
|
|
def test_assign_recipe_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.post.return_value = {
|
|
"id": 1, "station_id": 2, "recipe_id": 10,
|
|
"assigned_by": 1, "assigned_at": "2026-04-25T10:00:00",
|
|
}
|
|
resp = logged_in_client.post(
|
|
"/admin/api/stations/2/recipes", json={"recipe_id": 10},
|
|
)
|
|
assert resp.status_code == 201
|
|
mock_admin_api.post.assert_called_once_with(
|
|
"/api/stations/2/recipes", data={"recipe_id": 10},
|
|
)
|
|
|
|
|
|
def test_unassign_recipe_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.delete.return_value = {}
|
|
resp = logged_in_client.delete("/admin/api/stations/2/recipes/10")
|
|
assert resp.status_code == 200
|
|
mock_admin_api.delete.assert_called_once_with("/api/stations/2/recipes/10")
|
|
|
|
|
|
def test_list_station_recipes_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = [
|
|
{"id": 1, "code": "REC-001", "name": "R1", "active": True},
|
|
]
|
|
resp = logged_in_client.get("/admin/api/stations/2/recipes")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == [{"id": 1, "code": "REC-001", "name": "R1", "active": True}]
|
|
mock_admin_api.get.assert_called_once_with("/api/stations/2/recipes")
|
|
|
|
|
|
def test_non_admin_cannot_access(client, mock_admin_api):
|
|
"""Non-admin user gets redirected away from station management."""
|
|
with client.session_transaction() as sess:
|
|
sess["api_key"] = "test-key"
|
|
sess["user"] = {
|
|
"id": 2,
|
|
"username": "operator",
|
|
"roles": ["MeasurementTec"],
|
|
"is_admin": False,
|
|
"active": True,
|
|
}
|
|
sess["user_id"] = 2
|
|
resp = client.get("/admin/stations", follow_redirects=False)
|
|
assert resp.status_code in (301, 302)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-row station reset (point 7)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_reset_station_proxy(logged_in_client, mock_admin_api):
|
|
mock_admin_api.delete.return_value = {"station_id": 3, "removed": 2}
|
|
resp = logged_in_client.delete("/admin/api/stations/3/recipes")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"station_id": 3, "removed": 2}
|
|
mock_admin_api.delete.assert_called_once_with("/api/stations/3/recipes")
|
|
|
|
|
|
def test_reset_station_propagates_error(logged_in_client, mock_admin_api):
|
|
mock_admin_api.delete.return_value = {
|
|
"error": True, "status_code": 404, "detail": "Station not found",
|
|
}
|
|
resp = logged_in_client.delete("/admin/api/stations/999/recipes")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_reset_station_distinct_from_single_unassign(logged_in_client, mock_admin_api):
|
|
"""The collection route must not swallow the per-recipe delete."""
|
|
mock_admin_api.delete.return_value = {}
|
|
logged_in_client.delete("/admin/api/stations/3/recipes/7")
|
|
mock_admin_api.delete.assert_called_once_with("/api/stations/3/recipes/7")
|
|
|
|
|
|
def test_station_list_page_renders_assigned_recipes(logged_in_client, mock_admin_api):
|
|
"""Recipes assigned to a station reach the page so the row can show them."""
|
|
mock_admin_api.get.side_effect = [
|
|
[{
|
|
"id": 1, "code": "ST-001", "name": "Linea A", "location": "Reparto A",
|
|
"notes": None, "active": True, "created_by": 1,
|
|
"created_at": "2026-04-25T10:00:00",
|
|
"recipes": [{"id": 9, "code": "REC-ABC", "name": "Perno", "active": True}],
|
|
}],
|
|
[{"id": 9, "code": "REC-ABC", "name": "Perno", "active": True}],
|
|
]
|
|
resp = logged_in_client.get("/admin/stations")
|
|
assert resp.status_code == 200
|
|
assert b"REC-ABC" in resp.data
|
|
|
|
|
|
# --- Worker visione per stazione (#45) ---
|
|
|
|
def test_station_form_has_the_vision_worker_field(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = []
|
|
resp = logged_in_client.get("/admin/stations")
|
|
assert resp.status_code == 200
|
|
assert b"form.vision_worker_url" in resp.data
|
|
# etichetta in italiano o in inglese, secondo il catalogo compilato
|
|
assert b"Worker visione (URL)" in resp.data or b"Vision worker (URL)" in resp.data
|
|
|
|
|
|
def test_station_list_exposes_the_vision_worker_url(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.side_effect = [
|
|
[{"id": 1, "code": "ST-001", "name": "Linea A", "location": None,
|
|
"notes": None, "active": True, "vision_worker_url": "http://10.0.0.7:8100",
|
|
"created_by": 1, "created_at": "2026-04-25T10:00:00"}],
|
|
[],
|
|
]
|
|
resp = logged_in_client.get("/admin/stations")
|
|
assert b"http://10.0.0.7:8100" in resp.data # nel JSON iniziale della pagina
|
|
|
|
|
|
def test_station_proxies_pass_the_vision_worker_url_untouched(logged_in_client, mock_admin_api):
|
|
mock_admin_api.post.return_value = {"id": 7, "code": "ST-7", "name": "n"}
|
|
mock_admin_api.put.return_value = {"id": 7, "code": "ST-7", "name": "n"}
|
|
logged_in_client.post(
|
|
"/admin/api/stations",
|
|
json={"code": "ST-7", "name": "n", "vision_worker_url": "http://w:8100"},
|
|
)
|
|
mock_admin_api.post.assert_called_once_with(
|
|
"/api/stations",
|
|
data={"code": "ST-7", "name": "n", "vision_worker_url": "http://w:8100"},
|
|
)
|
|
logged_in_client.put("/admin/api/stations/7", json={"vision_worker_url": None})
|
|
mock_admin_api.put.assert_called_once_with(
|
|
"/api/stations/7", data={"vision_worker_url": None},
|
|
)
|
|
|
|
|
|
# --- Camera per stazione (#72) ---
|
|
|
|
def test_station_form_has_the_camera_fields(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = []
|
|
html = logged_in_client.get("/admin/stations").get_data(as_text=True)
|
|
assert "form.camera_url" in html
|
|
assert "form.camera_source" in html
|
|
|
|
|
|
def test_station_list_exposes_the_camera_fields(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.side_effect = [
|
|
[{"id": 1, "code": "ST-001", "name": "Linea A", "location": None,
|
|
"notes": None, "active": True,
|
|
"camera_url": "http://10.0.0.9:8765", "camera_source": "camera:folder",
|
|
"created_by": 1, "created_at": "2026-04-25T10:00:00"}],
|
|
[],
|
|
]
|
|
html = logged_in_client.get("/admin/stations").get_data(as_text=True)
|
|
assert "http://10.0.0.9:8765" in html
|
|
assert "camera:folder" in html
|
|
|
|
|
|
def test_station_payload_script_sends_the_camera_fields(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = []
|
|
html = logged_in_client.get("/admin/stations").get_data(as_text=True)
|
|
# default, reset, carica, crea, aggiorna: ogni punto che già porta vision_worker_url
|
|
assert html.count("camera_url") >= html.count("vision_worker_url")
|
|
assert html.count("camera_source") >= html.count("vision_worker_url")
|
|
|
|
|
|
def test_create_station_proxy_forwards_the_camera_fields(logged_in_client, mock_admin_api):
|
|
mock_admin_api.post.return_value = {"id": 5, "code": "ST-005", "name": "N"}
|
|
body = {"code": "ST-005", "name": "N",
|
|
"camera_url": "http://h:8765", "camera_source": "camera:folder"}
|
|
resp = logged_in_client.post("/admin/api/stations", json=body)
|
|
assert resp.status_code == 201
|
|
mock_admin_api.post.assert_called_once_with("/api/stations", data=body)
|
|
|
|
|
|
# --- Sorgenti camera lette dall'acquisitore (#88) ---
|
|
|
|
def _fake_sources_response(status=200, payload=None):
|
|
resp = MagicMock()
|
|
resp.status_code = status
|
|
resp.json.return_value = payload
|
|
return resp
|
|
|
|
|
|
def test_camera_sources_proxy_keeps_only_the_three_fields(logged_in_client, mock_admin_api):
|
|
payload = {"sources": [
|
|
{"name": "camera:folder", "label": "Cartella", "kind": "folder", "extra": 1},
|
|
{"name": "camera:123", "label": "Basler", "kind": "gige"},
|
|
]}
|
|
with patch("blueprints.admin.http_requests.get",
|
|
return_value=_fake_sources_response(200, payload)) as get:
|
|
resp = logged_in_client.get("/admin/api/camera/sources?url=http://acq:8765/")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["sources"][0] == {
|
|
"name": "camera:folder", "label": "Cartella", "kind": "folder"}
|
|
assert len(resp.get_json()["sources"]) == 2
|
|
assert get.call_args.args[0] == "http://acq:8765/camera/sources"
|
|
assert get.call_args.kwargs["timeout"] == 5
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["ftp://x", "", "http://", "acq:8765"])
|
|
def test_camera_sources_proxy_rejects_invalid_url(logged_in_client, mock_admin_api, url):
|
|
with patch("blueprints.admin.http_requests.get") as get:
|
|
resp = logged_in_client.get("/admin/api/camera/sources", query_string={"url": url})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()["error"] is True
|
|
get.assert_not_called()
|
|
|
|
|
|
def test_camera_sources_proxy_unreachable_is_502(logged_in_client, mock_admin_api):
|
|
import requests
|
|
with patch("blueprints.admin.http_requests.get",
|
|
side_effect=requests.ConnectionError("rifiutata")):
|
|
resp = logged_in_client.get("/admin/api/camera/sources?url=http://acq:8765")
|
|
assert resp.status_code == 502
|
|
assert "rifiutata" in resp.get_json()["detail"]
|
|
|
|
|
|
@pytest.mark.parametrize("status,payload", [(200, {"altro": 1}), (200, {"sources": "x"}), (500, {"sources": []})])
|
|
def test_camera_sources_proxy_invalid_response_is_502(logged_in_client, mock_admin_api, status, payload):
|
|
with patch("blueprints.admin.http_requests.get",
|
|
return_value=_fake_sources_response(status, payload)):
|
|
resp = logged_in_client.get("/admin/api/camera/sources?url=http://acq:8765")
|
|
assert resp.status_code == 502
|
|
|
|
|
|
def test_camera_sources_proxy_non_admin_is_redirected(client, mock_admin_api):
|
|
with client.session_transaction() as sess:
|
|
sess["api_key"] = "test-key"
|
|
sess["user"] = {"id": 2, "username": "operator", "roles": ["MeasurementTec"],
|
|
"is_admin": False, "active": True}
|
|
sess["user_id"] = 2
|
|
with patch("blueprints.admin.http_requests.get") as get:
|
|
resp = client.get("/admin/api/camera/sources?url=http://acq:8765")
|
|
assert resp.status_code in (301, 302)
|
|
get.assert_not_called()
|
|
|
|
|
|
def test_station_form_has_the_camera_sources_picker(logged_in_client, mock_admin_api):
|
|
mock_admin_api.get.return_value = []
|
|
html = logged_in_client.get("/admin/stations").get_data(as_text=True)
|
|
assert "cameraSourcesList" in html
|
|
assert "loadCameraSources" in html
|
|
assert "/acquisizione" in html
|