Adriano
6a8ad4dae4
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VNVrf2wZAmMAkgmwEuBB36
206 lines
8.7 KiB
Python
206 lines
8.7 KiB
Python
"""gitea:Adriano/TieMeasureFlow#53 #54 #55 - operator side of the check / note results.
|
|
|
|
The backend (task-results, in_loop, close_task_ids) lives in another unit: api_client
|
|
is mocked and the tests work against the contract written in the issues.
|
|
"""
|
|
import io
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tests.conftest import api_get_router
|
|
from tests.test_template_js_syntax import (
|
|
_INLINE_SCRIPT_RX, _check_alpine_attributes, _force_italian, _node_check,
|
|
)
|
|
|
|
FLASK_APP_DIR = Path(__file__).resolve().parent.parent
|
|
|
|
RECIPE = {
|
|
"id": 3, "code": "REC-1", "name": "Ricetta",
|
|
"measurement_interval_minutes": 30, "allow_manual_input": True,
|
|
"current_version": {"id": 5, "version_number": 1},
|
|
}
|
|
|
|
|
|
def _task(task_id, task_type, **extra):
|
|
task = {
|
|
"id": task_id, "recipe_id": 3, "version_id": 5, "task_type": task_type,
|
|
"title": "Task d'esempio", "directive": None, "description": None,
|
|
"file_path": None, "file_type": None, "annotations_json": None,
|
|
"subtasks": [], "fields": None,
|
|
}
|
|
task.update(extra)
|
|
return task
|
|
|
|
|
|
def _render(client, mock_api_client, monkeypatch, task):
|
|
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({
|
|
f"/api/tasks/{task['id']}": task,
|
|
"/api/recipes/3/tasks": [{"id": task["id"], "order_index": 0}],
|
|
"/api/recipes/3": RECIPE,
|
|
})
|
|
return client.get(f"/measure/execute/{task['id']}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page
|
|
# ---------------------------------------------------------------------------
|
|
def test_check_task_has_ok_nok_buttons(logged_in_client, mock_api_client, monkeypatch):
|
|
resp = _render(logged_in_client, mock_api_client, monkeypatch, _task(30, "check"))
|
|
html = resp.get_data(as_text=True)
|
|
assert resp.status_code == 200
|
|
assert "submit('ok')" in html and "submit('nok')" in html
|
|
assert "taskResultForm()" in html
|
|
assert "/measure/api/task-results" in html
|
|
assert "js/task-results.js" in html
|
|
|
|
|
|
def test_note_with_fields_shows_field_inputs(logged_in_client, mock_api_client, monkeypatch):
|
|
fields = [
|
|
{"key": "peso", "label": "Peso", "type": "number", "default": "1.5", "attachment": False},
|
|
{"key": "foto", "label": "Foto", "type": "text", "default": None, "attachment": True},
|
|
]
|
|
task = _task(31, "note", fields=fields)
|
|
html = _render(logged_in_client, mock_api_client, monkeypatch, task).get_data(as_text=True)
|
|
# the fields travel to the page in the task data, the inputs are drawn from them
|
|
assert '"key": "peso"' in html or '"key":"peso"' in html
|
|
assert 'x-model="f.values[fld.key]"' in html
|
|
assert "pickFile(f, fld.key, $event)" in html
|
|
assert "submit('done')" in html
|
|
|
|
|
|
def test_close_panel_and_modal_hook_present(logged_in_client, mock_api_client, monkeypatch):
|
|
html = _render(logged_in_client, mock_api_client, monkeypatch, _task(32, "check")).get_data(as_text=True)
|
|
assert "task-close-open" in html
|
|
assert "closeTasksRecorded()" in html
|
|
assert "close_task_ids" in html
|
|
assert "/measure/api/tasks/0" in html
|
|
|
|
|
|
@pytest.mark.parametrize("task", [
|
|
_task(30, "check"),
|
|
_task(31, "note", fields=[{"key": "n", "label": "L'etichetta", "type": "text",
|
|
"default": None, "attachment": True}]),
|
|
])
|
|
def test_inline_js_and_alpine_attributes_valid(logged_in_client, mock_api_client, monkeypatch, task):
|
|
_force_italian(logged_in_client)
|
|
html = _render(logged_in_client, mock_api_client, monkeypatch, task).get_data(as_text=True)
|
|
scripts = _INLINE_SCRIPT_RX.findall(html)
|
|
assert scripts
|
|
for i, body in enumerate(scripts):
|
|
_node_check(body, f"/measure/execute check-note script[{i}]")
|
|
_check_alpine_attributes(html, "/measure/execute check-note")
|
|
|
|
|
|
def test_plain_note_keeps_old_footer(logged_in_client, mock_api_client, monkeypatch):
|
|
"""A note without fields has no result panel: it is still the plain «Completato»."""
|
|
html = _render(logged_in_client, mock_api_client, monkeypatch, _task(33, "note")).get_data(as_text=True)
|
|
assert "goToNextTask()" in html
|
|
# the panel markup is in the page either way; hasResultPanel (false here) decides
|
|
assert "hasResultPanel" in html
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Proxies
|
|
# ---------------------------------------------------------------------------
|
|
def test_task_result_post_is_forwarded(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {"id": 9, "outcome": "nok"}
|
|
body = {
|
|
"task_id": 30, "production_run_id": 4, "outcome": "nok", "note": "graffio",
|
|
"values": None, "attachments": None, "ignored": "x",
|
|
}
|
|
resp = logged_in_client.post("/measure/api/task-results", json=body)
|
|
assert resp.status_code == 201
|
|
assert resp.get_json()["id"] == 9
|
|
args, kwargs = mock_api_client.post.call_args
|
|
assert args[0] == "/api/task-results"
|
|
assert kwargs["data"] == {
|
|
"task_id": 30, "production_run_id": 4, "outcome": "nok", "note": "graffio",
|
|
"values": None, "attachments": None,
|
|
}
|
|
|
|
|
|
def test_task_result_422_detail_is_passed_through(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {
|
|
"error": True, "status_code": 422, "detail": "Per un NOK serve la nota",
|
|
}
|
|
resp = logged_in_client.post(
|
|
"/measure/api/task-results", json={"task_id": 30, "outcome": "nok"})
|
|
assert resp.status_code == 422
|
|
assert resp.get_json()["detail"] == "Per un NOK serve la nota"
|
|
|
|
|
|
def test_task_result_requires_task_id(logged_in_client, mock_api_client):
|
|
resp = logged_in_client.post("/measure/api/task-results", json={"outcome": "ok"})
|
|
assert resp.status_code == 400
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
|
|
def test_task_results_list_forwards_filters(logged_in_client, mock_api_client):
|
|
mock_api_client.get.return_value = [{"id": 1, "task_id": 30}]
|
|
resp = logged_in_client.get("/measure/api/task-results?production_run_id=4&task_id=30")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == [{"id": 1, "task_id": 30}]
|
|
args, kwargs = mock_api_client.get.call_args
|
|
assert args[0] == "/api/task-results"
|
|
assert kwargs["params"] == {"production_run_id": "4", "task_id": "30"}
|
|
|
|
|
|
def test_attachment_upload_forwards_multipart(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {"file_path": "general/foto.png", "file_type": "image/png"}
|
|
resp = logged_in_client.post(
|
|
"/measure/api/attachments",
|
|
data={"file": (io.BytesIO(b"\x89PNG fake"), "foto.png", "image/png")},
|
|
content_type="multipart/form-data",
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.get_json()["file_path"] == "general/foto.png"
|
|
args, kwargs = mock_api_client.post.call_args
|
|
assert args[0] == "/api/task-results/attachments" # #55
|
|
name, content, mime = kwargs["files"]["file"]
|
|
assert name == "foto.png" and mime == "image/png" and content.startswith(b"\x89PNG")
|
|
|
|
|
|
def test_attachment_upload_requires_file(logged_in_client, mock_api_client):
|
|
resp = logged_in_client.post("/measure/api/attachments", data={})
|
|
assert resp.status_code == 400
|
|
mock_api_client.post.assert_not_called()
|
|
|
|
|
|
def test_attachment_upload_error_detail_is_passed_through(logged_in_client, mock_api_client):
|
|
mock_api_client.post.return_value = {"error": True, "status_code": 400, "detail": "File type not allowed"}
|
|
resp = logged_in_client.post(
|
|
"/measure/api/attachments",
|
|
data={"file": (io.BytesIO(b"x"), "a.exe", "application/octet-stream")},
|
|
content_type="multipart/form-data",
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()["detail"] == "File type not allowed"
|
|
|
|
|
|
def test_get_task_proxy(logged_in_client, mock_api_client):
|
|
mock_api_client.get.return_value = {"id": 40, "task_type": "note", "fields": []}
|
|
resp = logged_in_client.get("/measure/api/tasks/40")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["id"] == 40
|
|
assert mock_api_client.get.call_args[0][0] == "/api/tasks/40"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Component logic (plain node, stubbed fetch)
|
|
# ---------------------------------------------------------------------------
|
|
def test_task_results_component_js():
|
|
node = shutil.which("node")
|
|
if not node:
|
|
pytest.skip("node non installato: test JS di task-results.js saltato")
|
|
script = Path(__file__).resolve().parent / "js" / "test_task_results.js"
|
|
result = subprocess.run(
|
|
[node, str(script)], cwd=str(FLASK_APP_DIR),
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
assert result.returncode == 0, f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}"
|