vision: ricetta DXF riapribile, immagine salvata, rotte source e files, soglia segue la quota (Refs #105)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
This commit is contained in:
@@ -0,0 +1,226 @@
|
||||
"""#105: la ricetta «Crea ricetta da DXF» si puo' riaprire e modificare.
|
||||
|
||||
Il worker non gira nei test: si finge come in test_vision_dxf_compare.
|
||||
"""
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from sqlalchemy import select
|
||||
|
||||
from src.backend.config import settings
|
||||
from src.backend.models.orm.task import RecipeSubtask, RecipeTask
|
||||
from src.backend.services import vision_service
|
||||
from src.backend.tests.conftest import auth_headers
|
||||
from src.backend.tests.test_vision_dxf_compare import ( # noqa: F401 (fixture)
|
||||
redirect_uploads, validate_ok,
|
||||
)
|
||||
from src.backend.tests.test_vision_recipe_upload import an_image, make_task
|
||||
|
||||
DXF = b"0\nEOF"
|
||||
|
||||
|
||||
def _wizard_graph(band=0.3, pct=90.0, polarity="light") -> dict:
|
||||
"""Grafo come lo costruisce src/vision/costruttori.py::_graph."""
|
||||
return {
|
||||
"schema_version": 2, "name": "Confronto DXF",
|
||||
"meta": {"dxf": {"disegno": {"path": "disegno.dxf", "sha256": "x"}},
|
||||
"calibration": {"path": "calibrazione.json", "sha256": "y"}},
|
||||
"tools": [
|
||||
{"id": "pezzo", "name": "Pezzo", "kind": "LocateBlob",
|
||||
"values": {"polarity": polarity}, "links": {}},
|
||||
{"id": "dxf", "name": "Confronto DXF", "kind": "CompareDxf",
|
||||
"values": {"band_plus": band, "band_minus": band, "points_in_pct_nominal": pct},
|
||||
"links": {}},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def w2(monkeypatch):
|
||||
async def fake(image_bytes, dxf_bytes, **kwargs):
|
||||
graph = _wizard_graph(kwargs["band_mm"], kwargs["min_points_pct"], kwargs["polarity"])
|
||||
return {
|
||||
"graph": graph,
|
||||
"files": {
|
||||
"disegno.dxf": base64.b64encode(dxf_bytes).decode(),
|
||||
"calibrazione.json": base64.b64encode(
|
||||
json.dumps({"px_per_mm": kwargs["px_per_mm"]}).encode()).decode(),
|
||||
},
|
||||
"outputs": {"dxf.points_in_pct": 99.0}, "overlay": [], "outcome": "pass",
|
||||
}
|
||||
monkeypatch.setattr("src.backend.services.vision_service.dxf_compare_recipe", fake)
|
||||
|
||||
|
||||
def _bmp() -> bytes:
|
||||
buf = io.BytesIO()
|
||||
Image.new("L", (32, 32)).save(buf, format="BMP")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _save(client, user, task, *, image=None, ctype="image/png", **data):
|
||||
return client.post(
|
||||
f"/api/vision/tasks/{task.id}/recipe/dxf-compare",
|
||||
headers=auth_headers(user),
|
||||
files={"image": ("p", image if image is not None else an_image(), ctype),
|
||||
"dxf": ("d.dxf", DXF, "application/octet-stream")},
|
||||
data={"save": "true", "px_per_mm": "10", **{k: str(v) for k, v in data.items()}},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_keeps_the_reference_image_png(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
image = an_image()
|
||||
|
||||
resp = await _save(client, maker_user, task, image=image)
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
folder = settings.upload_path / resp.json()["recipe_dir"]
|
||||
assert (folder / "riferimento.png").read_bytes() == image
|
||||
assert (folder / "disegno.dxf").read_bytes() == DXF # il resto della cartella non cambia
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_keeps_the_reference_image_bmp(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
image = _bmp()
|
||||
|
||||
resp = await _save(client, maker_user, task, image=image, ctype="image/bmp")
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
folder = settings.upload_path / resp.json()["recipe_dir"]
|
||||
assert (folder / "riferimento.bmp").read_bytes() == image
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_source_returns_what_the_wizard_saved(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
await _save(client, maker_user, task, band_mm=0.3, min_points_pct=90, polarity="light")
|
||||
|
||||
resp = await client.get(
|
||||
f"/api/vision/tasks/{task.id}/recipe/dxf-compare/source", headers=auth_headers(maker_user),
|
||||
)
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json() == {
|
||||
"band_mm": 0.3, "min_points_pct": 90.0, "polarity": "light", "px_per_mm": 10.0,
|
||||
"dxf": "disegno.dxf", "image": "riferimento.png",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_source_404_without_recipe_or_with_a_hand_made_graph(client, db_session, maker_user):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
url = f"/api/vision/tasks/{task.id}/recipe/dxf-compare/source"
|
||||
|
||||
assert (await client.get(url, headers=auth_headers(maker_user))).status_code == 404
|
||||
|
||||
task.vision_json = {"schema_version": 2, "name": "a mano", "tools": [], "meta": {}}
|
||||
await db_session.commit()
|
||||
resp = await client.get(url, headers=auth_headers(maker_user))
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["detail"] == "nessuna ricetta da DXF su questo task"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_source_needs_maker(client, db_session, maker_user, measurement_tec_user):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
resp = await client.get(
|
||||
f"/api/vision/tasks/{task.id}/recipe/dxf-compare/source",
|
||||
headers=auth_headers(measurement_tec_user),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_source_of_an_old_recipe_has_no_image_and_no_unreadable_scale(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
|
||||
folder = tmp_path / "vision" / "recipes" / "1" / "abc"
|
||||
folder.mkdir(parents=True)
|
||||
(folder / "disegno.dxf").write_bytes(DXF)
|
||||
(folder / "calibrazione.json").write_text("{non json", encoding="utf-8")
|
||||
|
||||
source = vision_service.dxf_compare_source(_wizard_graph(), "vision/recipes/1/abc")
|
||||
|
||||
assert source["image"] is None and source["px_per_mm"] is None
|
||||
assert source["dxf"] == "disegno.dxf"
|
||||
assert vision_service.dxf_compare_source(None, None) is None
|
||||
# senza cartella i valori del grafo restano
|
||||
assert vision_service.dxf_compare_source(_wizard_graph(), None)["dxf"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_files_returns_the_bytes(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
image = _bmp()
|
||||
await _save(client, maker_user, task, image=image, ctype="image/bmp")
|
||||
|
||||
img = await client.get(
|
||||
f"/api/vision/tasks/{task.id}/recipe/files/riferimento.bmp", headers=auth_headers(maker_user),
|
||||
)
|
||||
dxf = await client.get(
|
||||
f"/api/vision/tasks/{task.id}/recipe/files/disegno.dxf", headers=auth_headers(maker_user),
|
||||
)
|
||||
|
||||
assert img.status_code == 200 and img.content == image
|
||||
assert img.headers["content-type"].startswith("image/")
|
||||
assert dxf.status_code == 200 and dxf.content == DXF
|
||||
assert dxf.headers["content-type"] # dal tipo dell'estensione, di ripiego octet-stream
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_files_404_on_unsafe_missing_or_no_recipe(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
base = f"/api/vision/tasks/{task.id}/recipe/files/"
|
||||
headers = auth_headers(maker_user)
|
||||
|
||||
assert (await client.get(base + "disegno.dxf", headers=headers)).status_code == 404 # nessuna ricetta
|
||||
|
||||
await _save(client, maker_user, task)
|
||||
assert (await client.get(base + "manca.bmp", headers=headers)).status_code == 404
|
||||
for unsafe in ("../x", "%2e%2e/x", "..%2f..%2fx", "sub/../../x"):
|
||||
assert (await client.get(base + unsafe, headers=headers)).status_code == 404, unsafe
|
||||
# la funzione non lascia uscire dalla cartella nemmeno con un assoluto
|
||||
assert vision_service.recipe_file_path("vision/recipes/1/abc", "../x") is None
|
||||
assert vision_service.recipe_file_path("vision/recipes/1/abc", "/etc/passwd") is None
|
||||
assert vision_service.recipe_file_path("../fuori", "x") is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resave_follows_the_new_threshold(client, db_session, maker_user, w2, validate_ok):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
task_id = task.id
|
||||
await _save(client, maker_user, task, min_points_pct=95)
|
||||
|
||||
resp = await _save(client, maker_user, task, min_points_pct=80)
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
db_session.expire_all()
|
||||
quotes = (await db_session.execute(
|
||||
select(RecipeSubtask).where(RecipeSubtask.vision_output == "dxf.points_in_pct")
|
||||
.where(RecipeSubtask.task_id == task_id))).scalars().all()
|
||||
assert len(quotes) == 1 and float(quotes[0].ltl) == 80.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ensure_points_quote_leaves_quote_with_measurements(
|
||||
client, db_session, maker_user, w2, validate_ok, monkeypatch,
|
||||
):
|
||||
task, _ = await make_task(db_session, maker_user)
|
||||
task_id = task.id
|
||||
await _save(client, maker_user, task, min_points_pct=95)
|
||||
|
||||
async def has_measurements(db, version_id):
|
||||
return True
|
||||
monkeypatch.setattr("src.backend.services.recipe_service.version_has_measurements", has_measurements)
|
||||
task = await db_session.get(RecipeTask, task_id)
|
||||
|
||||
changed = await vision_service.ensure_points_quote(db_session, task, 80)
|
||||
|
||||
assert changed is False
|
||||
quote = (await db_session.execute(
|
||||
select(RecipeSubtask).where(RecipeSubtask.vision_output == "dxf.points_in_pct")
|
||||
.where(RecipeSubtask.task_id == task_id))).scalar_one()
|
||||
assert float(quote.ltl) == 95.0
|
||||
Reference in New Issue
Block a user