TieMeasureFlow/src/vision_worker/tests/test_dxf_compare.py

"""POST /recipes/dxf-compare del worker (#83): forma della risposta ed errori."""
import pytest

# Come test_worker.py: serve l'albero vision (Python >=3.13), altrimenti skipped.
pytest.importorskip(
    "visionsuite.task",
    reason="the vision tree requires Python >=3.13; see test_worker.py",
)

import base64
import io
import time

import numpy as np
from httpx import ASGITransport, AsyncClient
from PIL import Image

from src.vision.tests.test_costruttori import PX_PER_MM, SAMPLE_DXF, a_scene
from src.vision_worker.main import app


def _png(array: np.ndarray) -> bytes:
    buffer = io.BytesIO()
    Image.fromarray(array).save(buffer, format="PNG")
    return buffer.getvalue()


async def _post(image: bytes, dxf: bytes, **data):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://worker") as client:
        return await client.post(
            "/recipes/dxf-compare",
            files={"image": ("riferimento.png", image, "image/png"),
                   "dxf": ("disegno.dxf", dxf, "application/octet-stream")},
            data={"px_per_mm": str(PX_PER_MM), "band_mm": "2.0", **data},
        )


@pytest.mark.asyncio
async def test_dxf_compare_200_con_le_chiavi_del_contratto():
    inizio = time.monotonic()
    response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes())
    durata = time.monotonic() - inizio

    assert response.status_code == 200, response.text
    body = response.json()
    assert {"graph", "files", "outputs", "overlay", "outcome"} <= set(body)
    assert set(body["files"]) == {"disegno.dxf", "calibrazione.json"}
    assert base64.b64decode(body["files"]["disegno.dxf"]) == SAMPLE_DXF.read_bytes()
    assert set(body["outputs"]) == {
        "dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max"}
    assert body["outputs"]["dxf.points_in_pct"] > 95.0
    assert body["graph"]["schema_version"] == 2
    print(f"\nW2 sulla scena di prova: {durata:.2f}s")  # riportato nel rapporto


@pytest.mark.asyncio
async def test_dxf_compare_immagine_senza_pezzo_e_422():
    bianca = np.full((600, 800), 255, np.uint8)
    response = await _post(_png(bianca), SAMPLE_DXF.read_bytes())
    assert response.status_code == 422
    assert "pezzo" in response.json()["detail"]


@pytest.mark.asyncio
async def test_dxf_compare_dxf_illeggibile_e_422():
    response = await _post(_png(a_scene(0.0)), b"non e' un dxf")
    assert response.status_code == 422
    assert "DXF" in response.json()["detail"]


@pytest.mark.asyncio
async def test_dxf_compare_immagine_non_decodificabile_e_422():
    response = await _post(b"non e' un'immagine", SAMPLE_DXF.read_bytes())
    assert response.status_code == 422
    assert "not decodable" in response.json()["detail"].lower()


@pytest.mark.asyncio
async def test_dxf_compare_parametro_fuori_range_e_422():
    response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes(), polarity="grigio")
    assert response.status_code == 422


# --- #101: POST /dxf/preview ---------------------------------------------------

def _rect_dxf(width: float, height: float, y0: float = 3.0) -> bytes:
    """DXF minimo con ezdxf: un rettangolo chiuso, y in alto rispetto a 0."""
    import ezdxf
    doc = ezdxf.new()
    doc.modelspace().add_lwpolyline(
        [(1, y0), (1 + width, y0), (1 + width, y0 + height), (1, y0 + height)], close=True,
    )
    buffer = io.StringIO()
    doc.write(buffer)
    return buffer.getvalue().encode("utf-8")


def _post_preview(dxf: bytes):
    # TestClient sincrono: non serve pytest-asyncio (assente nel venv del worker)
    from fastapi.testclient import TestClient
    return TestClient(app).post(
        "/dxf/preview", files={"dxf": ("d.dxf", dxf, "application/octet-stream")},
    )


def test_dxf_preview_rettangolo_10x5():
    resp = _post_preview(_rect_dxf(10, 5))
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["larghezza_mm"] == 10 and body["altezza_mm"] == 5
    assert body["entita"] == 1 and body["punti"] == sum(len(c) for c in body["contorni"])
    # y ribaltata: il rettangolo stava a y 3..8 nel DXF, ora a -8..-3 (asse verso il basso)
    assert body["bbox"] == {"xmin": 1, "ymin": -8, "xmax": 11, "ymax": -3}
    ring = body["contorni"][0]
    assert ring[0] == ring[-1]  # contorno chiuso: primo == ultimo
    assert all(-8 <= y <= -3 for _, y in ring)


@pytest.mark.parametrize("content", [b"", b"questo non e' un dxf"])
def test_dxf_preview_illeggibile_e_422(content):
    resp = _post_preview(content)
    assert resp.status_code == 422
    assert resp.json()["detail"].startswith("DXF")


def test_dxf_preview_senza_geometria_e_422():
    import ezdxf
    doc = ezdxf.new()
    buffer = io.StringIO()
    doc.write(buffer)
    resp = _post_preview(buffer.getvalue().encode("utf-8"))
    assert resp.status_code == 422
    assert "senza geometria" in resp.json()["detail"]


def test_dxf_preview_decima_oltre_il_tetto(monkeypatch):
    from src.vision_worker import main
    monkeypatch.setattr(main, "_PREVIEW_MAX_POINTS", 50)
    import ezdxf
    doc = ezdxf.new()
    doc.modelspace().add_circle((0, 0), 20)
    buffer = io.StringIO()
    doc.write(buffer)
    full = main._dxf_preview(buffer.getvalue().encode("utf-8"))
    assert full["punti"] <= 50 + 1 and full["larghezza_mm"] == pytest.approx(40, abs=0.01)  # dimensioni dai punti completi