"""Test delle rotte di acquisizione di scripts/camera_service.py (#87). Uso: uv run --python 3.13 --extra vision-worker --extra dev \ --with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q Copre: POST /camera/anteprima, GET/PUT /camera/parametri, POST /camera/parametri/salva, caricamento dei settaggi alla riapertura, alias GET /acquisizione. Sorgente regolabile finta (iniettata su open_source) e FolderCamera su PNG temporanei. """ import importlib.util import json from pathlib import Path import cv2 import numpy as np import pytest from fastapi.testclient import TestClient from visionsuite.camera.camera_interface import CameraStateError MODULO = Path(__file__).resolve().parents[1] / "camera_service.py" SORGENTE = "camera:finta" def _carica(): spec = importlib.util.spec_from_file_location("camera_service_acq", MODULO) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod class Finta: """Camera regolabile finta: stesso vocabolario di GenicamSource.""" DOMINI = { "esposizione": {"tipo": "numero", "min": 10.0, "max": 100000.0, "passo": 1.0}, "esposizione_auto": {"tipo": "scelta", "scelte": ["Off", "Once", "Continuous"]}, "guadagno": {"tipo": "numero", "min": 0.0, "max": 24.0}, "guadagno_auto": {"tipo": "scelta", "scelte": ["Off", "Once", "Continuous"]}, } def __init__(self, supportati=None, rifiuta=()): self.valori = {"esposizione": 5000.0, "esposizione_auto": "Continuous", "guadagno": 1.0, "guadagno_auto": "Off"} self.supportati = set(supportati if supportati is not None else self.DOMINI) self.rifiuta = set(rifiuta) self.scritture = [] self.caricato = None self.chiusa = False # metà bianca (255) e metà a 100: saturati 50%, media 177,5 frame = np.full((1000, 2000, 3), 100, np.uint8) frame[:, :1000] = 255 self.frame = frame def supporta(self, nome): return nome in self.supportati def dominio(self, nome): return {"capacita": nome, "valore": self.valori[nome], **self.DOMINI[nome]} def leggi(self, nome): return self.valori[nome] def scrivi(self, nome, valore, assesta=0): if nome in self.rifiuta: raise CameraStateError(f"non posso impostare {nome}: nodo in sola lettura") self.scritture.append((nome, valore, assesta)) self.valori[nome] = valore def salva_settaggi(self, percorso): Path(percorso).write_text(json.dumps(self.valori), encoding="utf-8") return dict(self.valori) def carica_settaggi(self, percorso, forza=False): self.caricato = Path(percorso) self.valori.update(json.loads(self.caricato.read_text(encoding="utf-8"))) return [] def grab(self, timeout=0): return self.frame.copy() def close(self): self.chiusa = True @pytest.fixture def cartella(tmp_path): img = tmp_path / "foto" img.mkdir() for nome in ("a.png", "b.png"): cv2.imwrite(str(img / nome), np.full((48, 64, 3), 128, np.uint8)) return img def _client(cartella, config, monkeypatch, finta): mod = _carica() monkeypatch.setattr(mod, "open_source", lambda *a, **k: finta) c = TestClient(mod.create_app(cartella, config)) c.mod = mod return c @pytest.fixture def finta(): return Finta() @pytest.fixture def client(cartella, tmp_path, monkeypatch, finta): with _client(cartella, tmp_path / "cfg", monkeypatch, finta) as c: c.finta = finta c.config = tmp_path / "cfg" yield c @pytest.fixture def client_cartella(cartella, tmp_path): mod = _carica() with TestClient(mod.create_app(cartella, tmp_path / "cfg")) as c: yield c # ---- anteprima def test_anteprima_jpeg_ridotta_e_header(client): r = client.post("/camera/anteprima", json={"source": SORGENTE, "lato_max": 500}) assert r.status_code == 200 assert r.headers["content-type"] == "image/jpeg" img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_COLOR) assert img.shape[:2] == (250, 500) # lato lungo 2000 -> 500 assert r.headers["x-raddrizzato"] == "no" assert (r.headers["x-larghezza"], r.headers["x-altezza"]) == ("2000", "1000") # frame pieno assert r.headers["x-saturati-pct"] == "50.00" assert r.headers["x-media"] == "177.5" assert float(r.headers["x-nitidezza"]) > 0 # bordo netto fra le due metà def test_anteprima_mai_ingrandita(client_cartella): r = client_cartella.post("/camera/anteprima", json={"source": "camera:folder"}) assert r.status_code == 200 img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_COLOR) assert img.shape[:2] == (48, 64) # lato_max 1280 > 64: resta com'è assert r.headers["x-nitidezza"] == "0.0" # grigio uniforme assert r.headers["x-saturati-pct"] == "0.00" assert r.headers["x-media"] == "128.0" def test_anteprima_parametri_errati_400(client): for corpo in ({}, {"source": SORGENTE, "lato_max": 100}, {"source": SORGENTE, "lato_max": 5000}, {"source": SORGENTE, "timeout_ms": 0}, {"source": SORGENTE, "grezzo": "si"}): assert client.post("/camera/anteprima", json=corpo).status_code == 400, corpo def test_anteprima_sorgente_inesistente_404(client_cartella): assert client_cartella.post("/camera/anteprima", json={"source": "camera:non-esiste"}).status_code == 404 def test_anteprima_errore_camera_503_e_timeout_504(client): from visionsuite.camera.camera_interface import CameraGrabError def rotta(timeout=0): raise CameraGrabError("Camera is not open") client.finta.grab = rotta r = client.post("/camera/anteprima", json={"source": SORGENTE}) assert r.status_code == 503 and r.json()["detail"] == "La camera non è aperta" def lenta(timeout=0): raise TimeoutError("scaduto") client.finta.grab = lenta assert client.post("/camera/anteprima", json={"source": SORGENTE}).status_code == 504 # ---- parametri def test_parametri_regolabile(client): r = client.get("/camera/parametri", params={"source": SORGENTE}) assert r.status_code == 200 j = r.json() assert j["source"] == SORGENTE and j["regolabile"] is True assert list(j["parametri"]) == ["esposizione", "esposizione_auto", "guadagno", "guadagno_auto"] assert j["parametri"]["esposizione"]["valore"] == 5000.0 assert j["parametri"]["esposizione_auto"]["scelte"] == ["Off", "Once", "Continuous"] def test_parametri_solo_quelli_supportati(cartella, tmp_path, monkeypatch): with _client(cartella, tmp_path / "cfg", monkeypatch, Finta(supportati=["esposizione"])) as c: j = c.get("/camera/parametri", params={"source": SORGENTE}).json() assert j["regolabile"] is True and list(j["parametri"]) == ["esposizione"] def test_parametri_sorgente_non_regolabile(client_cartella): j = client_cartella.get("/camera/parametri", params={"source": "camera:folder"}).json() assert j == {"source": "camera:folder", "regolabile": False, "parametri": {}} def test_parametri_senza_source_400(client): assert client.get("/camera/parametri").status_code == 400 def test_put_scrive_prima_gli_auto(client): r = client.put("/camera/parametri", json={"source": SORGENTE, "valori": { "esposizione": 8000, "guadagno": 3.5, "esposizione_auto": "Off", "guadagno_auto": "Off"}}) assert r.status_code == 200 assert client.finta.scritture == [ ("esposizione_auto", "Off", 3), ("guadagno_auto", "Off", 3), ("esposizione", 8000, 3), ("guadagno", 3.5, 3)] j = r.json() # valori riletti dalla camera assert j["parametri"]["esposizione"]["valore"] == 8000 assert j["parametri"]["esposizione_auto"]["valore"] == "Off" def test_put_nome_ignoto_400(client): r = client.put("/camera/parametri", json={"source": SORGENTE, "valori": {"frame_rate": 10}}) assert r.status_code == 400 assert client.finta.scritture == [] for corpo in ({"source": SORGENTE}, {"source": SORGENTE, "valori": {}}, {"source": SORGENTE, "valori": {"guadagno": True}}): assert client.put("/camera/parametri", json=corpo).status_code == 400, corpo def test_put_non_regolabile_409(client_cartella): r = client_cartella.put("/camera/parametri", json={"source": "camera:folder", "valori": {"esposizione": 1000}}) assert r.status_code == 409 assert r.json()["detail"] == "questa sorgente non ha parametri regolabili" def test_put_rifiuto_della_camera_422(cartella, tmp_path, monkeypatch): with _client(cartella, tmp_path / "cfg", monkeypatch, Finta(rifiuta=["guadagno"])) as c: r = c.put("/camera/parametri", json={"source": SORGENTE, "valori": {"guadagno": 5}}) assert r.status_code == 422 assert "guadagno" in r.json()["detail"] # ---- salvataggio e riapertura def test_salva_crea_il_file(client): r = client.post("/camera/parametri/salva", json={"source": SORGENTE}) assert r.status_code == 200 j = r.json() assert j["source"] == SORGENTE and j["salvati"] is True atteso = client.config / "settaggi" / j["file"] assert atteso.is_file() assert j["file"].startswith("camera_finta-") and j["file"].endswith(".json") assert json.loads(atteso.read_text(encoding="utf-8"))["esposizione"] == 5000.0 def test_salva_non_regolabile_409(client_cartella): r = client_cartella.post("/camera/parametri/salva", json={"source": "camera:folder"}) assert r.status_code == 409 def test_riapertura_carica_i_settaggi(cartella, tmp_path, monkeypatch): config = tmp_path / "cfg" prima = Finta() with _client(cartella, config, monkeypatch, prima) as c: c.put("/camera/parametri", json={"source": SORGENTE, "valori": {"esposizione": 7777}}) file = c.post("/camera/parametri/salva", json={"source": SORGENTE}).json()["file"] seconda = Finta() # servizio riavviato: camera nuova, valori di fabbrica with _client(cartella, config, monkeypatch, seconda) as c: assert seconda.caricato is None # niente apertura finché nessuno la chiede c.get("/camera/parametri", params={"source": SORGENTE}) assert seconda.caricato == config / "settaggi" / file assert seconda.valori["esposizione"] == 7777 def test_settaggi_rotti_non_bloccano_l_apertura(cartella, tmp_path, monkeypatch): class Rotta(Finta): def carica_settaggi(self, percorso, forza=False): raise ValueError("file illeggibile") config = tmp_path / "cfg" with _client(cartella, config, monkeypatch, Finta()) as c: file = c.post("/camera/parametri/salva", json={"source": SORGENTE}).json()["file"] assert (config / "settaggi" / file).is_file() with _client(cartella, config, monkeypatch, Rotta()) as c: assert c.get("/camera/parametri", params={"source": SORGENTE}).status_code == 200 # ---- pagina def test_acquisizione_alias_della_pagina(client): a = client.get("/acquisizione") c = client.get("/calibrazione") assert a.status_code == 200 and a.text == c.text