"""Tests TDD — Extraction « dossier patient » (brique 3). Deux couches testées : 1. ``vwb_db.persist_extracted_dossier`` : depuis une grille OCR (List[List[cell]]), crée ExtractionJob → ExtractedTable → ExtractedField et commit. Testé sur SQLite mémoire via un app-context Flask jetable (PAS la vraie DB VWB — isolation). 2. ``replay_engine._handle_extract_dossier_action`` : lit last_screenshot, appelle ``extract_grid_from_image`` (mocké), applique la gate qualité (complete / needs_review), persiste via vwb_db et n'échoue JAMAIS le replay (grille vide → needs_review, sans lever). ⚠️ Canal extraction = données patient EN CLAIR (volontaire) : on vérifie que les valeurs sont persistées telles quelles, sans tokenisation. """ import pytest from flask import Flask # vwb_db ajoute visual_workflow_builder/backend au sys.path à l'import → # doit précéder l'import de db.models (couplage worker→DB VWB mutualisé). import agent_v0.server_v1.vwb_db as vwb_db import agent_v0.server_v1.replay_engine as replay_engine from db.models import db, ExtractionJob, ExtractedTable, ExtractedField # --------------------------------------------------------------------------- # Fixtures : app Flask jetable sur SQLite mémoire (isolation totale) # --------------------------------------------------------------------------- @pytest.fixture def mem_app(): """App Flask minimale liée à une DB SQLite en mémoire.""" app = Flask("test_extract_dossier") app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db.init_app(app) with app.app_context(): db.create_all() yield app def _grid_2x2(): """Grille connue 2×2 (confiances hautes).""" return [ [ {"text": "Nom", "bbox": [[0, 0], [1, 0], [1, 1], [0, 1]], "confidence": 0.95, "row": 0, "col": 0}, {"text": "MOREL", "bbox": [[2, 0], [3, 0], [3, 1], [2, 1]], "confidence": 0.92, "row": 0, "col": 1}, ], [ {"text": "IPP", "bbox": [[0, 2], [1, 2], [1, 3], [0, 3]], "confidence": 0.90, "row": 1, "col": 0}, {"text": "25123456", "bbox": [[2, 2], [3, 2], [3, 3], [2, 3]], "confidence": 0.88, "row": 1, "col": 1}, ], ] # --------------------------------------------------------------------------- # 1) persist_extracted_dossier # --------------------------------------------------------------------------- @pytest.mark.unit def test_persist_extracted_dossier_creates_job_table_fields(mem_app): job_id = vwb_db.persist_extracted_dossier( _grid_2x2(), patient_ref="MOREL Catherine", source_session_id="sess-42", screenshot_ref="/captures/last.png", screen_bbox={"x": 0, "y": 0, "width": 800, "height": 600}, status="complete", ) assert isinstance(job_id, str) and job_id job = db.session.get(ExtractionJob, job_id) assert job is not None assert job.status == "complete" assert job.patient_ref == "MOREL Catherine" # EN CLAIR, non tokenisé assert job.source_session_id == "sess-42" tables = ExtractedTable.query.filter_by(job_id=job_id).all() assert len(tables) == 1 assert tables[0].screenshot_ref == "/captures/last.png" assert tables[0].screen_bbox == {"x": 0, "y": 0, "width": 800, "height": 600} fields = ExtractedField.query.filter_by(table_id=tables[0].id).all() assert len(fields) == 4 # 2×2 cellules values = {(f.row, f.col): f.value for f in fields} assert values[(0, 1)] == "MOREL" # valeur patient EN CLAIR conservée assert values[(1, 1)] == "25123456" confs = {(f.row, f.col): f.confidence for f in fields} assert confs[(0, 0)] == pytest.approx(0.95) @pytest.mark.unit def test_persist_extracted_dossier_empty_grid_still_creates_job(mem_app): """Grille vide → Job + Table sans Field (statut transmis tel quel).""" job_id = vwb_db.persist_extracted_dossier( [], patient_ref=None, source_session_id="sess-empty", screenshot_ref="/captures/empty.png", screen_bbox=None, status="needs_review", ) job = db.session.get(ExtractionJob, job_id) assert job is not None and job.status == "needs_review" tables = ExtractedTable.query.filter_by(job_id=job_id).all() assert len(tables) == 1 assert ExtractedField.query.filter_by(table_id=tables[0].id).count() == 0 # --------------------------------------------------------------------------- # 2) _handle_extract_dossier_action # --------------------------------------------------------------------------- @pytest.mark.unit def test_handle_extract_dossier_complete(mem_app, monkeypatch, tmp_path): # screenshot bidon sur disque (le mock OCR ignore le contenu) shot = tmp_path / "shot.png" shot.write_bytes(b"\x89PNG") # extract_grid_from_image mocké → grille 2×2 de confiance haute monkeypatch.setattr( "core.llm.extract_grid_from_image", lambda *a, **k: _grid_2x2(), ) # vwb_app_context pointé sur l'app mémoire de la fixture monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context()) monkeypatch.setattr(replay_engine, "vwb_db", vwb_db, raising=False) replay_state = { "last_screenshot": str(shot), "variables": {}, "replay_id": "rep-1", } action = { "type": "extract_dossier", "parameters": { "output_var": "dossier_id", "patient_ref": "MOREL Catherine", "expected_cols": 2, "min_confidence": 0.5, }, } ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-42") assert ok is True job_id = replay_state["variables"]["dossier_id"] assert isinstance(job_id, str) and job_id with mem_app.app_context(): job = db.session.get(ExtractionJob, job_id) assert job is not None assert job.status == "complete" # gate OK : non vide, conf ok, 2 cols @pytest.mark.unit def test_handle_extract_dossier_low_confidence_needs_review(mem_app, monkeypatch, tmp_path): shot = tmp_path / "shot.png" shot.write_bytes(b"\x89PNG") low_grid = [ [{"text": "x", "bbox": [], "confidence": 0.10, "row": 0, "col": 0}], ] monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: low_grid) monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context()) replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-2"} action = {"type": "extract_dossier", "parameters": {"min_confidence": 0.5}} ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-low") assert ok is False # gate a basculé en needs_review job_id = replay_state["variables"]["extracted_dossier"] with mem_app.app_context(): assert db.session.get(ExtractionJob, job_id).status == "needs_review" @pytest.mark.unit def test_handle_extract_dossier_empty_grid_no_raise(mem_app, monkeypatch, tmp_path): shot = tmp_path / "shot.png" shot.write_bytes(b"\x89PNG") monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: []) monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context()) replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-3"} action = {"type": "extract_dossier", "parameters": {}} # Ne lève jamais ; grille vide → needs_review ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-empty") assert ok is False job_id = replay_state["variables"]["extracted_dossier"] with mem_app.app_context(): assert db.session.get(ExtractionJob, job_id).status == "needs_review" @pytest.mark.unit def test_handle_extract_dossier_persist_failure_no_raise(mem_app, monkeypatch, tmp_path): """Si la persistance lève, le handler log et n'échoue PAS le replay.""" shot = tmp_path / "shot.png" shot.write_bytes(b"\x89PNG") monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: _grid_2x2()) monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context()) def _boom(*a, **k): raise RuntimeError("DB down") monkeypatch.setattr(vwb_db, "persist_extracted_dossier", _boom) replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-4"} action = {"type": "extract_dossier", "parameters": {}} ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-boom") assert ok is False # jamais de raise @pytest.mark.unit def test_extract_dossier_declared_in_action_type_sets(): assert "extract_dossier" in replay_engine._ALLOWED_ACTION_TYPES assert "extract_dossier" in replay_engine._SERVER_SIDE_ACTION_TYPES