"""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)