Adriano
a5e8472e88
/camera/calibrate non salva piu' (calcolo_id in memoria, ultimi 8); nuove rotte GET/POST/DELETE /camera/calibrazioni e PUT /camera/calibrazione-attiva (409 se la risoluzione non combacia col frame della sorgente); DELETE /camera/calibration ora scollega. Test: rotte nuove e test esistenti aggiornati (save=true -> salva con nome). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""Test delle rotte dell'archivio unico di calibrazioni con nome (#91): scripts/camera_service.py.
|
||
|
||
Uso: uv run --python 3.13 --extra vision-worker --extra dev \
|
||
--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q
|
||
Sorgenti: camera:folder su cartelle temporanee (520x400 scacchiera piana, 640x480 densa).
|
||
Solo tmp_path: niente ProgramData, niente camera vera.
|
||
"""
|
||
import json
|
||
|
||
import cv2
|
||
import numpy as np
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from test_calibrazione import ( # fixture e helper della calibrazione (#83/#84)
|
||
CAMPI, CHIAVI_ATTIVA, CHIAVI_DENSA, CHIAVI_PLANARE, SORGENTE, _carica, _png,
|
||
carica_su, cartella, cartella_densa, client, client_denso, salva_con_nome,
|
||
una_scacchiera, una_scacchiera_densa,
|
||
)
|
||
|
||
__all__ = ["cartella", "cartella_densa", "client", "client_denso"] # le fixture importate restano usate
|
||
|
||
|
||
def _calcola_file(c, **extra):
|
||
"""Calcolo planare da un'immagine (520x400): ritorna la risposta JSON."""
|
||
r = c.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI, **extra},
|
||
files={"image": ("scacco.png", _png(una_scacchiera()), "image/png")})
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
# ---- POST /camera/calibrate
|
||
|
||
def test_calibrate_senza_source_con_immagine_ok(client):
|
||
r = client.post("/camera/calibrate", data=CAMPI,
|
||
files={"image": ("C:\\foto\\scacco.png", _png(una_scacchiera()), "image/png")})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["origine"] == "file:scacco.png" # solo il nome del file, non il percorso
|
||
|
||
|
||
def test_calibrate_senza_source_e_senza_immagine_400(client):
|
||
r = client.post("/camera/calibrate", data=CAMPI)
|
||
assert r.status_code == 400 and "source" in r.json()["detail"]
|
||
|
||
|
||
def test_calibrate_da_sorgente_ha_origine_camera_e_id(client):
|
||
r = client.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI})
|
||
assert r.status_code == 200, r.text
|
||
corpo = r.json()
|
||
assert corpo["origine"] == SORGENTE and corpo["calibrated_at"]
|
||
assert len(corpo["calcolo_id"]) == 32 and int(corpo["calcolo_id"], 16) >= 0 # esadecimale
|
||
assert corpo["calcolo_id"] != client.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI}).json()["calcolo_id"]
|
||
|
||
|
||
# ---- POST /camera/calibrazioni
|
||
|
||
def test_salva_201_con_metadati(client):
|
||
calcolo = _calcola_file(client)
|
||
r = salva_con_nome(client, calcolo, " Banco 1 ")
|
||
assert r.status_code == 201, r.text
|
||
meta = r.json()
|
||
assert meta["nome"] == "Banco 1" and meta["origine"] == "file:scacco.png"
|
||
assert (meta["image_width"], meta["image_height"]) == (520, 400)
|
||
assert meta["metodo"] == "planare" and meta["attiva_su"] == []
|
||
assert meta["px_per_mm"] == pytest.approx(20.0, rel=0.005)
|
||
# lo stesso calcolo si puo' salvare anche con un altro nome
|
||
assert salva_con_nome(client, calcolo, "Banco 2").status_code == 201
|
||
|
||
|
||
def test_salva_calcolo_sconosciuto_404(client):
|
||
r = client.post("/camera/calibrazioni", json={"calcolo_id": "0" * 32, "nome": "X"})
|
||
assert r.status_code == 404 and r.json()["detail"] == "calcolo non più disponibile: ricalcolare"
|
||
|
||
|
||
def test_salva_calcolo_scaduto_dopo_otto_calcoli_404(client):
|
||
primo = _calcola_file(client)
|
||
ultimo = primo
|
||
for _ in range(8): # il nono calcolo spinge fuori il primo
|
||
ultimo = _calcola_file(client)
|
||
assert salva_con_nome(client, primo, "Vecchio").status_code == 404
|
||
assert salva_con_nome(client, ultimo, "Nuovo").status_code == 201
|
||
|
||
|
||
@pytest.mark.parametrize("nome", ["", " ", "x" * 61, "a\x01b", None, 7])
|
||
def test_salva_nome_non_valido_400(client, nome):
|
||
r = client.post("/camera/calibrazioni", json={"calcolo_id": _calcola_file(client)["calcolo_id"], "nome": nome})
|
||
assert r.status_code == 400, r.text
|
||
assert client.get("/camera/calibrazioni").json()["calibrazioni"] == []
|
||
|
||
|
||
def test_salva_corpo_incompleto_400(client):
|
||
assert client.post("/camera/calibrazioni", json={"nome": "X"}).status_code == 400
|
||
calcolo = _calcola_file(client)
|
||
r = client.post("/camera/calibrazioni", json={"calcolo_id": calcolo["calcolo_id"], "nome": "X", "sovrascrivi": "si"})
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_salva_nome_gia_usato_409_e_sovrascrivi(client):
|
||
a = _calcola_file(client)
|
||
assert salva_con_nome(client, a, "Banco").status_code == 201
|
||
r = salva_con_nome(client, a, "BANCO") # senza distinguere maiuscole
|
||
assert r.status_code == 409 and r.json()["detail"] == "esiste già una calibrazione «Banco»"
|
||
assert len(client.get("/camera/calibrazioni").json()["calibrazioni"]) == 1
|
||
r = salva_con_nome(client, a, "BANCO", sovrascrivi=True)
|
||
assert r.status_code == 201 and r.json()["nome"] == "BANCO"
|
||
voci = client.get("/camera/calibrazioni").json()["calibrazioni"]
|
||
assert [v["nome"] for v in voci] == ["BANCO"]
|
||
|
||
|
||
# ---- GET /camera/calibrazioni
|
||
|
||
def test_elenco_vuoto_e_ordine(client):
|
||
assert client.get("/camera/calibrazioni").json() == {"calibrazioni": [], "attiva": None}
|
||
prima = _calcola_file(client)
|
||
salva_con_nome(client, prima, "prima")
|
||
seconda = _calcola_file(client)
|
||
seconda_ts = seconda["calibrated_at"]
|
||
salva_con_nome(client, seconda, "seconda")
|
||
voci = client.get("/camera/calibrazioni").json()["calibrazioni"]
|
||
assert {v["nome"] for v in voci} == {"prima", "seconda"}
|
||
assert [v["calibrated_at"] for v in voci] == sorted((v["calibrated_at"] for v in voci), reverse=True)
|
||
assert set(voci[0]) == {"nome", "origine", "image_width", "image_height", "px_per_mm", "mm_per_px",
|
||
"metodo", "square_mm", "calibrated_at", "attiva_su"}
|
||
assert seconda_ts >= prima["calibrated_at"]
|
||
|
||
|
||
def test_elenco_con_source_dice_l_attiva(client):
|
||
salva_con_nome(client, _calcola_file(client), "Banco")
|
||
salva_con_nome(client, _calcola_file(client), "Altra")
|
||
assert carica_su(client, SORGENTE, "banco").status_code == 200
|
||
r = client.get("/camera/calibrazioni", params={"source": SORGENTE}).json()
|
||
assert r["attiva"] == "Banco"
|
||
attiva_su = {v["nome"]: v["attiva_su"] for v in r["calibrazioni"]}
|
||
assert attiva_su == {"Banco": [SORGENTE], "Altra": []}
|
||
assert client.get("/camera/calibrazioni", params={"source": "camera:altra"}).json()["attiva"] is None
|
||
assert client.get("/camera/calibrazioni", params={"source": " "}).status_code == 400
|
||
|
||
|
||
# ---- DELETE /camera/calibrazioni
|
||
|
||
def test_elimina_204_404_e_toglie_dalle_sorgenti(client):
|
||
salva_con_nome(client, _calcola_file(client), "Banco")
|
||
carica_su(client, SORGENTE, "Banco")
|
||
assert client.delete("/camera/calibrazioni", params={"nome": "BANCO"}).status_code == 204
|
||
assert client.get("/camera/calibrazioni").json()["calibrazioni"] == []
|
||
assert client.get("/camera/calibration", params={"source": SORGENTE}).status_code == 404 # non piu' attiva
|
||
assert json.loads((client.config / "attive.json").read_text(encoding="utf-8")) == {}
|
||
assert client.delete("/camera/calibrazioni", params={"nome": "Banco"}).status_code == 404
|
||
assert client.delete("/camera/calibrazioni").status_code == 404 # senza nome
|
||
|
||
|
||
# ---- PUT /camera/calibrazione-attiva
|
||
|
||
def test_carica_risoluzione_uguale_200_come_get(client):
|
||
salva_con_nome(client, _calcola_file(client), "Banco") # 520x400 = la scacchiera in cartella
|
||
r = carica_su(client, SORGENTE, "banco")
|
||
assert r.status_code == 200, r.text
|
||
assert set(r.json()) == CHIAVI_PLANARE | CHIAVI_ATTIVA
|
||
assert r.json()["nome"] == "Banco" and r.json()["source"] == SORGENTE
|
||
assert client.get("/camera/calibration", params={"source": SORGENTE}).json() == r.json()
|
||
|
||
|
||
def test_carica_risoluzione_diversa_409_e_anteprima_continua(client_denso):
|
||
# calibrazione da file 520x400, sorgente 640x480
|
||
salva_con_nome(client_denso, _calcola_file(client_denso), "Piccola")
|
||
r = carica_su(client_denso, SORGENTE, "Piccola")
|
||
assert r.status_code == 409
|
||
assert r.json()["detail"] == ("la calibrazione «Piccola» è per immagini 520×400, "
|
||
"la sorgente dà 640×480: non si può caricare")
|
||
assert client_denso.get("/camera/calibration", params={"source": SORGENTE}).status_code == 404 # nulla attivato
|
||
# l'acquisizione continua: grab e anteprima rispondono, frame non raddrizzato
|
||
g = client_denso.post("/camera/grab", json={"source": SORGENTE})
|
||
assert g.status_code == 200 and g.headers["x-raddrizzato"] == "no"
|
||
a = client_denso.post("/camera/anteprima", json={"source": SORGENTE})
|
||
assert a.status_code == 200 and a.headers["x-raddrizzato"] == "no"
|
||
assert (a.headers["x-larghezza"], a.headers["x-altezza"]) == ("640", "480")
|
||
|
||
|
||
def test_carica_risoluzione_diversa_non_scalza_l_attiva_precedente(client_denso):
|
||
densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}).json()
|
||
salva_con_nome(client_denso, densa, "Giusta")
|
||
salva_con_nome(client_denso, _calcola_file(client_denso), "Sbagliata")
|
||
assert carica_su(client_denso, SORGENTE, "Giusta").status_code == 200
|
||
assert carica_su(client_denso, SORGENTE, "Sbagliata").status_code == 409
|
||
assert client_denso.get("/camera/calibration", params={"source": SORGENTE}).json()["nome"] == "Giusta"
|
||
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "si"
|
||
|
||
|
||
def test_carica_densa_su_sorgente_stessa_risoluzione_raddrizza(client_denso):
|
||
densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}).json()
|
||
assert salva_con_nome(client_denso, densa, "Densa").json()["metodo"] == "densa"
|
||
r = carica_su(client_denso, SORGENTE, "Densa")
|
||
assert r.status_code == 200 and set(r.json()) == CHIAVI_DENSA | CHIAVI_ATTIVA
|
||
assert client_denso.post("/camera/anteprima", json={"source": SORGENTE}).headers["x-raddrizzato"] == "si"
|
||
|
||
|
||
def test_carica_nome_sconosciuto_404(client):
|
||
r = carica_su(client, SORGENTE, "nessuna")
|
||
assert r.status_code == 404
|
||
assert client.put("/camera/calibrazione-attiva", json={"source": SORGENTE}).status_code == 404 # senza nome
|
||
assert client.put("/camera/calibrazione-attiva", json={"nome": "x"}).status_code == 400 # senza source
|
||
|
||
|
||
def test_carica_su_sorgente_inesistente_404(client):
|
||
salva_con_nome(client, _calcola_file(client), "Banco")
|
||
assert carica_su(client, "camera:non-esiste", "Banco").status_code == 404
|
||
|
||
|
||
def test_carica_senza_risoluzione_registrata_409(cartella, tmp_path):
|
||
_carica()
|
||
import calibrazione
|
||
config = tmp_path / "config_senza"
|
||
calibrazione.Archivio(config).salva_nome("Senza", {"px_per_mm": 10.0, "calibrated_at": "2026-01-01T00:00:00+00:00"})
|
||
with TestClient(_carica().create_app(cartella, config_dir=config)) as c:
|
||
r = carica_su(c, SORGENTE, "Senza")
|
||
assert r.status_code == 409
|
||
assert r.json()["detail"] == "la calibrazione «Senza» non ha la risoluzione registrata: non si può caricare"
|
||
|
||
|
||
# ---- contratto verso il backend e migrazione all'avvio
|
||
|
||
def test_get_calibration_contratto_per_il_backend(client):
|
||
# src/backend/services/vision_service.py legge questi campi
|
||
salva_con_nome(client, _calcola_file(client), "Banco")
|
||
carica_su(client, SORGENTE, "Banco")
|
||
dati = client.get("/camera/calibration", params={"source": SORGENTE}).json()
|
||
for campo in ("px_per_mm", "mm_per_px", "image_width", "image_height", "calibrated_at", "metodo"):
|
||
assert campo in dati
|
||
assert client.get("/camera/calibration", params={"source": "camera:altra"}).json() == {
|
||
"detail": "sorgente non calibrata"}
|
||
|
||
|
||
def test_avvio_migra_il_formato_vecchio(cartella, tmp_path):
|
||
_carica()
|
||
import calibrazione
|
||
config = tmp_path / "config_vecchia"
|
||
config.mkdir()
|
||
nudo = calibrazione.Archivio.__new__(calibrazione.Archivio) # per avere il nome file del formato vecchio
|
||
nudo.cartella = config
|
||
vecchio = nudo._file(SORGENTE)
|
||
vecchio.write_text(json.dumps({"source": SORGENTE, "px_per_mm": 20.0, "mm_per_px": 0.05,
|
||
"image_width": 520, "image_height": 400, "metodo": "planare",
|
||
"calibrated_at": "2026-10-01T10:00:00+00:00"}), encoding="utf-8")
|
||
with TestClient(_carica().create_app(cartella, config_dir=config)) as c:
|
||
dati = c.get("/camera/calibration", params={"source": SORGENTE}).json()
|
||
assert dati["nome"] == SORGENTE and dati["source"] == SORGENTE and dati["px_per_mm"] == 20.0
|
||
voci = c.get("/camera/calibrazioni", params={"source": SORGENTE}).json()
|
||
assert [v["nome"] for v in voci["calibrazioni"]] == [SORGENTE] and voci["attiva"] == SORGENTE
|
||
assert voci["calibrazioni"][0]["attiva_su"] == [SORGENTE]
|
||
assert not vecchio.exists()
|