from __future__ import annotations # ruff: noqa: E402 import io import json import os import sqlite3 import sys from pathlib import Path import pytest from PIL import Image, ImageDraw, ImageFont from reportlab.pdfgen import canvas ROOT = Path(__file__).resolve().parents[1] HEALTH = ROOT / "scripts" / "health" sys.path.insert(0, str(HEALTH)) sys.path.insert(0, str(ROOT / "tests")) from dashboard_v5.document_review import ( MAX_DOCUMENT_BYTES, _external_tool_env, candidate_rows, quarantine_upload, validate_upload_file, ) from dashboard_v5.read_api import dispatch_api from dashboard_v5.sprint6i_a_schema import apply_schema, assert_schema from migrate_sprint6i_a_schema import safe_migrate from fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture import health_dashboard_action_worker as worker def metadata(**changes): base = {"document_date":"2026-07-18","document_type":"laboratory_report","institution":"Synthetic Clinic","personal_title":"Synthetic document","investigation_day":"2026-07-18","note":"synthetic only"} base.update(changes); return base def test_external_media_tools_do_not_inherit_python_package_injection(monkeypatch): monkeypatch.setenv("PYTHONPATH", "/tmp/synthetic-incompatible-packages") monkeypatch.setenv("PYTHONHOME", "/tmp/synthetic-python-home") environment = _external_tool_env() assert "PYTHONPATH" not in environment assert "PYTHONHOME" not in environment assert environment["LC_ALL"] == "C.UTF-8" def text_pdf(text: str) -> bytes: output = io.BytesIO(); pdf = canvas.Canvas(output, pagesize=(300,300)); pdf.drawString(24,250,text); pdf.save(); return output.getvalue() def scan_pdf(text: str) -> bytes: image=Image.new("RGB",(1200,500),"white");draw=ImageDraw.Draw(image) font=ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",54) draw.text((40,180),text,fill="black",font=font,stroke_width=1) png=io.BytesIO();image.save(png,format="PNG");png.seek(0) output=io.BytesIO();pdf=canvas.Canvas(output,pagesize=(600,250));pdf.drawInlineImage(Image.open(png),0,0,600,250);pdf.save();return output.getvalue() def configure_worker(monkeypatch, database: Path, quarantine: Path, storage: Path): monkeypatch.setattr(worker,"DASHBOARD_DB",database) monkeypatch.setattr(worker,"DOCUMENT_QUARANTINE",quarantine) monkeypatch.setattr(worker,"DOCUMENT_STORAGE",storage) monkeypatch.setattr(worker,"DASHBOARD_V5_FILE",None) def test_secure_quarantine_rejects_magic_symlink_traversal_and_oversize(tmp_path): quarantine=tmp_path/"q" valid=quarantine_upload(text_pdf("Synthetic PDF"),metadata(),quarantine) assert valid["mime"]=="application/pdf" and valid["token"].startswith("docq_") with pytest.raises(ValueError,match="unsupported_file_type"): quarantine_upload(b"active",metadata(),quarantine) fake=tmp_path/"fake.pdf";fake.write_bytes(b"\xff\xd8\xffnot-a-real-jpeg") with pytest.raises(ValueError): validate_upload_file(fake) outside=tmp_path/"outside.pdf";outside.write_bytes(text_pdf("outside"));link=tmp_path/"link.pdf";link.symlink_to(outside) with pytest.raises(OSError): validate_upload_file(link) large=tmp_path/"large.pdf" with large.open("wb") as handle: handle.write(b"%PDF-"); handle.truncate(MAX_DOCUMENT_BYTES+1) with pytest.raises(ValueError,match="invalid_file_size"): validate_upload_file(large) assert not (quarantine/"../outside").resolve().is_relative_to(quarantine.resolve()) def test_text_pdf_import_is_unreviewed_searchable_and_idempotent_duplicate(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) data=text_pdf("CRP 4.2 mg/l Referenz: 0-5 mg/l") first=quarantine_upload(data,metadata(),quarantine) first_payload=worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":first["token"],"sha256":first["sha256"],"metadata":metadata()}) first_id=worker.apply_document_import(first_payload) second=quarantine_upload(data,metadata(personal_title="Duplicate"),quarantine) second_id=worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":second["token"],"sha256":second["sha256"],"metadata":metadata(personal_title="Duplicate")})) assert first_id != second_id connection=sqlite3.connect(database);connection.row_factory=sqlite3.Row;assert_schema(connection) rows=connection.execute("SELECT extraction_status,content_status,search_status,duplicate_document_id FROM document_processing ORDER BY document_id DESC LIMIT 2").fetchall() assert rows[1]["extraction_status"]=="text_layer" and rows[1]["content_status"]=="pending" and rows[1]["search_status"]=="machine_searchable" assert rows[0]["duplicate_document_id"] is not None assert connection.execute("SELECT COUNT(*) FROM document_candidates WHERE candidate_type='laboratory_value'").fetchone()[0]>=1 assert connection.execute("SELECT COUNT(*) FROM dokumente d JOIN document_processing p ON p.document_id=d.id WHERE d.review_status='nicht_geprueft'").fetchone()[0] == 2 connection.close() def test_quarantine_swap_is_detected_before_import(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) upload=quarantine_upload(text_pdf("Original synthetic"),metadata(),quarantine) raw=quarantine/f"{upload['token']}.bin";raw.write_bytes(text_pdf("Swapped synthetic"));raw.chmod(0o600) action=worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":upload["token"],"sha256":upload["sha256"],"metadata":metadata()}) with pytest.raises(ValueError,match="quarantine_metadata_mismatch"): worker.apply_document_import(action) assert not list(storage.glob("*")) def test_scanned_pdf_uses_local_ocr_and_image_is_registered(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) scanned=quarantine_upload(scan_pdf("Synthetic OCR Report 2026"),metadata(document_type="doctor_report"),quarantine) worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":scanned["token"],"sha256":scanned["sha256"],"metadata":metadata(document_type="doctor_report")})) image=Image.new("RGB",(500,200),"white");ImageDraw.Draw(image).text((20,80),"Synthetic photo",fill="black");raw=io.BytesIO();image.save(raw,format="JPEG") photo=quarantine_upload(raw.getvalue(),metadata(document_type="photo"),quarantine) worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":photo["token"],"sha256":photo["sha256"],"metadata":metadata(document_type="photo")})) connection=sqlite3.connect(database) statuses=[row[0] for row in connection.execute("SELECT extraction_status FROM document_processing ORDER BY document_id DESC LIMIT 2")] connection.close() assert statuses==["ocr","ocr"] def test_review_queue_candidates_compare_and_machine_search_are_explicit(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) for value in ("CRP 4.2 mg/l Diagnose: Synthetic alpha","CRP 5.1 mg/l Diagnose: Synthetic alpha changed"): upload=quarantine_upload(text_pdf(value),metadata(document_type="doctor_report"),quarantine) worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":upload["token"],"sha256":upload["sha256"],"metadata":metadata(document_type="doctor_report")})) listing=dispatch_api(database,"/api/v1/documents","category=doctor_report&limit=10") latest=listing["documents"][0] review=dispatch_api(database,f"/api/v1/documents/{latest['id']}/review","") assert review["statuses"]["content"]=="pending" and review["pages"] and review["candidates"] queue=dispatch_api(database,"/api/v1/document-review-queue","") assert {item["code"] for item in queue["groups"]}>={"now_reviewable"} assert queue["total_decisions"]>=1 comparison=dispatch_api(database,f"/api/v1/documents/{latest['id']}/compare","") assert comparison["medical_evaluation"] is False and comparison["new"]+comparison["changed"]+comparison["identical"]>=1 assert dispatch_api(database,"/api/v1/search","q=Synthetic&include_machine=0")["groups"]["documents"]==[] machine=dispatch_api(database,"/api/v1/search","q=Synthetic&include_machine=1")["groups"]["documents"] assert machine and all(item["drill_down_target"]=="document_machine_extraction" for item in machine) def test_candidate_decision_only_changes_staging_via_worker(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) upload=quarantine_upload(text_pdf("Befunddatum 2026-07-18\nCRP 4.2 mg/l"),metadata(),quarantine) intake=worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":upload["token"],"sha256":upload["sha256"],"metadata":metadata()})) connection=sqlite3.connect(database);candidate=connection.execute("SELECT id FROM document_candidates WHERE status='open' LIMIT 1").fetchone()[0];labs_before=connection.execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0];connection.close() for operation,target,action in (("original_review","","d"*32),("page_review","page_1","e"*32)): source_payload={"version":1,"action":"document_review","operation":operation,"document_id":intake,"target_id":target,"value":"","unit":"","decision":"","metadata":{}} worker.apply_document_review(worker.validate_document_action(source_payload),action) projected=next(item for item in dispatch_api(database,"/api/v1/lab-review","status=all")["candidates"] if item["id"]==candidate) payload={"version":1,"action":"document_review","operation":"candidate_decision","document_id":intake,"target_id":candidate,"value":"4.2","unit":"mg/l","decision":"confirmed","metadata":{},"expected_candidate_revision":projected["revision_number"],"preview_revision":projected["revision"]} normalized=worker.validate_document_action(payload);worker.apply_document_review(normalized,"a"*32) connection=sqlite3.connect(database) assert connection.execute("SELECT status FROM document_candidates WHERE id=?",(candidate,)).fetchone()[0]=="confirmed" assert connection.execute("SELECT COUNT(*) FROM laborwerte").fetchone()[0]==labs_before assert connection.execute("SELECT review_status FROM dokumente d JOIN document_processing p ON p.document_id=d.id WHERE p.intake_id=?",(intake,)).fetchone()[0]=="nicht_geprueft" versions_before=connection.execute("SELECT COUNT(*) FROM document_text_versions").fetchone()[0] connection.close() correction={"version":1,"action":"document_review","operation":"text_correction","document_id":intake,"target_id":"page_1","value":"CRP 4.3 mg/l","unit":"","decision":"","metadata":{}} worker.apply_document_review(worker.validate_document_action(correction),"b"*32) connection=sqlite3.connect(database) assert connection.execute("SELECT COUNT(*) FROM document_text_versions").fetchone()[0]==versions_before+1 assert connection.execute("SELECT original_text FROM document_text_versions ORDER BY version LIMIT 1").fetchone()[0] != connection.execute("SELECT original_text FROM document_text_versions ORDER BY version DESC LIMIT 1").fetchone()[0] connection.close() incomplete={"version":1,"action":"document_review","operation":"content_review","document_id":intake,"target_id":"","value":"","unit":"","decision":"","metadata":{}} with pytest.raises(RuntimeError,match="incomplete"): worker.apply_document_review(worker.validate_document_action(incomplete),"c"*32) for operation,target,action_id in (("original_review","","1"*32),("page_review","page_1","2"*32)): action={"version":1,"action":"document_review","operation":operation,"document_id":intake,"target_id":target,"value":"","unit":"","decision":"","metadata":{}} worker.apply_document_review(worker.validate_document_action(action),action_id) connection=sqlite3.connect(database);open_candidates=[row[0] for row in connection.execute("SELECT id FROM document_candidates WHERE status IN ('open','conflicting')")];connection.close() projected_by_id={item["id"]:item for item in dispatch_api(database,"/api/v1/lab-review","status=all")["candidates"]} for index,candidate_id in enumerate(open_candidates): projected=projected_by_id[candidate_id] action={"version":1,"action":"document_review","operation":"candidate_decision","document_id":intake,"target_id":candidate_id,"value":"rejected","unit":"","decision":"rejected","metadata":{},"expected_candidate_revision":projected["revision_number"],"preview_revision":projected["revision"]} worker.apply_document_review(worker.validate_document_action(action),f"{index+10:032x}") worker.apply_document_review(worker.validate_document_action(incomplete),"f"*32) connection=sqlite3.connect(database) assert connection.execute("SELECT review_status FROM dokumente d JOIN document_processing p ON p.document_id=d.id WHERE p.intake_id=?",(intake,)).fetchone()[0]=="geprueft" connection.close() def test_identical_repeated_page_does_not_create_new_medical_candidate(tmp_path,monkeypatch): database=tmp_path/"health.db";build_dashboard_v5_fixture(database) quarantine=tmp_path/"quarantine";storage=tmp_path/"documents";configure_worker(monkeypatch,database,quarantine,storage) ids=[] for text in ("CRP 4.2 mg/l","CRP 4.2 mg/l "): upload=quarantine_upload(text_pdf(text),metadata(),quarantine) ids.append(worker.apply_document_import(worker.validate_document_action({"version":1,"action":"document_import","quarantine_token":upload["token"],"sha256":upload["sha256"],"metadata":metadata()}))) connection=sqlite3.connect(database) second_id=connection.execute("SELECT document_id FROM document_processing WHERE intake_id=?",(ids[1],)).fetchone()[0] assert connection.execute("SELECT repetition_status FROM document_pages WHERE document_id=?",(second_id,)).fetchone()[0]=="identical" assert connection.execute("SELECT COUNT(*) FROM document_candidates WHERE document_id=? AND candidate_type='laboratory_value'",(second_id,)).fetchone()[0]==0 connection.close() def test_candidate_contract_includes_unit_page_context_engine_and_confidence(): rows=candidate_rows(["CRP 4.2 mg/l Referenz: 0-5 mg/l"],"text_layer",{"document_date":"2026-07-18","institution":"Synthetic Clinic"}) lab=next(item for item in rows if item["candidate_type"]=="laboratory_value") assert lab["unit"].lower()=="mg/l" and lab["page_number"]==1 and lab["section_number"]==1 assert lab["context_text"] and lab["engine"]=="text_layer" and 0<=lab["confidence"]<=1 def test_additive_copy_first_migration_is_idempotent_with_restore_proof(tmp_path): database=tmp_path/"health.db";backup=tmp_path/"backup"/"health-pre-6ia.db";build_dashboard_v5_fixture(database) result=safe_migrate(database,backup) assert result=={"copy_migration":"ok","production_migration":"ok","idempotency":"ok","integrity":"ok","foreign_keys":"ok","restore_test":"ok"} assert backup.exists() and (backup.stat().st_mode & 0o777)==0o600 connection=sqlite3.connect(database);assert_schema(connection);connection.close()