Adriano
6080e6aebc
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
227 lines
9.1 KiB
Python
227 lines
9.1 KiB
Python
"""#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
|