from __future__ import annotations import json import sqlite3 import sys from datetime import date from pathlib import Path import pytest 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.data_provider import medication_data from dashboard_v5.capture_contract import validate_capture_payload from dashboard_v5.medication_contract import action_context_revision, resolve_action_preview from dashboard_v5.medication_schema import apply_schema, assert_schema from dashboard_v5.read_api import APIError, _next_planned_medications, dispatch_api from fixtures.dashboard_v5_fixture import build_dashboard_v5_fixture from migrate_sprint7c_f_medication_schema import safe_prepare import health_dashboard_action_worker as worker def _legacy_db(path: Path) -> None: connection = sqlite3.connect(path) connection.execute("PRAGMA foreign_keys=ON") connection.executescript( """ CREATE TABLE dokumente(id INTEGER PRIMARY KEY); CREATE TABLE medikamente( id INTEGER PRIMARY KEY AUTOINCREMENT,dokument_id INTEGER,medikament_name TEXT NOT NULL, dosierung TEXT,anwendungsform TEXT,erhaltungsform TEXT,ermittlung_datum TEXT, FOREIGN KEY(dokument_id) REFERENCES dokumente(id)); CREATE TABLE medication_administrations( id INTEGER PRIMARY KEY AUTOINCREMENT,datum TEXT NOT NULL,medication_name TEXT NOT NULL, dose TEXT,route TEXT,event_type TEXT,scheduled_next_date TEXT,notes TEXT, source TEXT DEFAULT 'manual',created_at TEXT DEFAULT CURRENT_TIMESTAMP,occurred_at TEXT, UNIQUE(datum,medication_name,event_type)); CREATE TABLE capture_action_log( idempotency_key TEXT PRIMARY KEY,request_version INTEGER NOT NULL,action_hash TEXT NOT NULL, entry_id TEXT NOT NULL,processed_at TEXT NOT NULL); """ ) connection.executemany( "INSERT INTO medikamente(medikament_name,dosierung,anwendungsform) VALUES(?,?,?)", [(f"Legacy {index}", "free 40 mg/0.8 ml" if index == 1 else None, "subkutan") for index in range(1, 8)], ) connection.executemany( """INSERT INTO medication_administrations( datum,medication_name,dose,route,event_type,scheduled_next_date,notes,source,occurred_at) VALUES(?,?,?,?,?,?,?,?,?)""", [ (f"2026-0{index}-01", f"Legacy {index}", "free 40 mg/0.8 ml" if index <= 3 else None, "subkutan", "administered" if index <= 4 else "documented_status", f"2026-0{index+1}-01", "legacy note", "manual", f"2026-0{index}-01T10:00") for index in range(1, 6) ], ) connection.commit() connection.close() def _table_rows(path: Path, table: str) -> list[tuple]: connection = sqlite3.connect(path) try: return connection.execute(f'SELECT * FROM "{table}" ORDER BY id').fetchall() finally: connection.close() def test_copy_first_migration_preserves_every_legacy_value_and_proves_restore(tmp_path: Path) -> None: source = tmp_path / "legacy.db" backup = tmp_path / "backup.db" migrated = tmp_path / "migrated.db" restored = tmp_path / "restored.db" _legacy_db(source) before_master = _table_rows(source, "medikamente") before_events = _table_rows(source, "medication_administrations") report = safe_prepare(source, backup, migrated, restored) assert report["migration_name"] == "sprint7c_f1_medication_capture_hotfix" assert report["schema_version"] == 8 assert report["dry_run"] == report["idempotency"] == report["restore_proof_status"] == "ok" assert report["medication_master_rows"] == 7 assert report["medication_event_rows"] == 5 assert backup.exists() and restored.exists() and migrated.exists() assert _table_rows(restored, "medikamente") == before_master assert _table_rows(restored, "medication_administrations") == before_events connection = sqlite3.connect(migrated) connection.row_factory = sqlite3.Row assert_schema(connection) rows = connection.execute("SELECT dose,planned_dose_value,planned_dose_unit,actual_dose_value,actual_dose_unit,business_revision FROM medication_administrations ORDER BY id").fetchall() assert rows[0]["dose"] == "free 40 mg/0.8 ml" assert all(all(row[key] is None for key in ("planned_dose_value", "planned_dose_unit", "actual_dose_value", "actual_dose_unit", "business_revision")) for row in rows) assert connection.execute("PRAGMA integrity_check").fetchone()[0] == "ok" assert connection.execute("PRAGMA foreign_key_check").fetchall() == [] connection.close() def test_structured_corrections_reject_orphans_cycles_missing_reason_and_mutation(tmp_path: Path) -> None: database = tmp_path / "health.db" _legacy_db(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row apply_schema(connection) connection.commit() revision = "a" * 64 medication_id = connection.execute("SELECT id FROM medikamente ORDER BY id LIMIT 1").fetchone()[0] with pytest.raises(sqlite3.IntegrityError): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status,correction_reason,business_revision) VALUES('2026-06-01','Legacy 1','corrected',?,999,'administered','reason',?)""", (medication_id, revision)) origin = connection.execute("SELECT id FROM medication_administrations WHERE medication_name='Legacy 1'").fetchone()[0] with pytest.raises(sqlite3.IntegrityError): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status,correction_reason,business_revision) VALUES('2026-06-02','Legacy 1','corrected',?,?,'administered','',?)""", (medication_id, origin, revision)) connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status,correction_reason,business_revision) VALUES('2026-06-02','Legacy 1','corrected',?,?,'administered','documented correction',?)""", (medication_id, origin, revision)) correction = connection.execute("SELECT id FROM medication_administrations WHERE event_type='corrected'").fetchone()[0] connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status,correction_reason,business_revision) VALUES('2026-06-03','Legacy 1','corrected',?,?,'missed','self-correction cycle two',?)""", (medication_id, correction, "b" * 64)) with pytest.raises(sqlite3.IntegrityError): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status,correction_reason,business_revision) VALUES('2026-06-04','Legacy 1','corrected',?,?,'planned','parallel correction',?)""", (medication_id, origin, "c" * 64)) with pytest.raises(sqlite3.IntegrityError): connection.execute("UPDATE medication_administrations SET dose='rewrite' WHERE id=?", (origin,)) with pytest.raises(sqlite3.IntegrityError): connection.execute("UPDATE medication_administrations SET notes='overwrite' WHERE id=?", (correction,)) with pytest.raises(sqlite3.IntegrityError): connection.execute("DELETE FROM medication_administrations WHERE id=?", (correction,)) connection.close() def _structured_payload(data: dict, key: str = "1" * 32) -> dict: return { "version": 1, "action": "capture_entry", "capture_type": "medication", "request_version": 1, "idempotency_key": key, "occurred_at": "2026-06-15T10:00", "ended_at": None, "data": data, "attachments": [], "corrects_entry_id": None, "withdraws_entry_id": None, } def _action_data(prescription: dict, *, status: str = "administered", preview: str | None = None) -> dict: return { "contract": "health.medication_action.v1", "status": status, "medication_ref": prescription["id"], "planned_event_ref": "", "name": prescription["name"], "planned_dose_value": "", "planned_dose_unit": "", "actual_dose_value": "10", "actual_dose_unit": "mg", "route_original": "oral", "route_normalized": "oral", "injection_region": "", "injection_side": "", "injection_detail": "", "lot_number": "LOT-SYNTHETIC", "correction_target_ref": "", "corrected_target_status": "", "correction_reason": "", "note": "synthetic only", "preview_revision": preview or prescription["preview_revision"], "plan_value_confirmed": False, "deviation_confirmed": True, "duplicate_confirmed": False, } def _bind_preview(database: Path, payload: dict) -> dict: payload = validate_capture_payload(payload) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row try: payload["data"]["preview_revision"] = resolve_action_preview(connection, payload)[0] finally: connection.close() return payload def test_worker_replay_uses_capture_action_log_and_stale_preview_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) medication = dispatch_api(database, "/api/v1/medications", "") prescription = next(item for item in medication["prescriptions"] if item["name"] == "SYNTHETIC_ADMINISTERED_MEDICATION") monkeypatch.setattr(worker, "DASHBOARD_DB", database) monkeypatch.setattr(worker, "DASHBOARD_V5_FILE", None) monkeypatch.setattr(worker, "CAPTURE_QUARANTINE", tmp_path / "quarantine") monkeypatch.setattr(worker, "CAPTURE_MEDIA", tmp_path / "media") payload = _bind_preview(database, _structured_payload(_action_data(prescription))) first = worker.apply_capture_action(payload) second = worker.apply_capture_action(payload) assert first == second connection = sqlite3.connect(database) assert connection.execute("SELECT COUNT(*) FROM capture_action_log WHERE idempotency_key=?", (payload["idempotency_key"],)).fetchone()[0] == 1 assert connection.execute("SELECT COUNT(*) FROM medication_administrations WHERE business_revision IS NOT NULL").fetchone()[0] == 1 connection.close() tampered = json.loads(json.dumps(payload)) tampered["idempotency_key"] = "c" * 32 tampered["data"]["actual_dose_value"] = "99" with pytest.raises(RuntimeError, match="stale medication preview revision"): worker.apply_capture_action(tampered) stale = _structured_payload(_action_data(prescription, preview="f" * 64), key="2" * 32) with pytest.raises(RuntimeError, match="stale medication preview revision"): worker.apply_capture_action(stale) def test_new_reader_separates_statuses_and_old_reader_remains_compatible(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row prescription = connection.execute("SELECT * FROM medikamente WHERE medikament_name='SYNTHETIC_ADMINISTERED_MEDICATION'").fetchone() revision = "c" * 64 connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,actual_dose_value,actual_dose_unit,route_original, route_normalized,source,occurred_at,business_revision) VALUES('2026-06-14','SYNTHETIC_ADMINISTERED_MEDICATION','missed',?,'','','oral','oral','fixture','2026-06-14T10:00',?)""", (prescription["id"], revision)) connection.commit() old_summary, old_events = medication_data(connection, "2026-06-15") connection.close() assert old_summary["last_administered"] is not None assert old_events history = dispatch_api(database, "/api/v1/medications", "") assert history["contract"] == "health.medication_history.v1" assert history["current_prescriptions"] assert history["administered"] and history["planned"] and history["missed"] and history["unknown"] assert history["unknown"][0]["status"] == "unknown" def test_preview_revision_changes_when_bound_prescription_changes(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row row = connection.execute("SELECT * FROM medikamente WHERE medikament_name='SYNTHETIC_ADMINISTERED_MEDICATION'").fetchone() before = action_context_revision(row) connection.execute("UPDATE medikamente SET prescription_status='paused' WHERE id=?", (row["id"],)) changed = connection.execute("SELECT * FROM medikamente WHERE id=?", (row["id"],)).fetchone() after = action_context_revision(changed) connection.close() assert before != after medication = dispatch_api(database, "/api/v1/medications", "") unknown = next(item for item in medication["other_prescriptions"] if item["status"] == "unknown") with pytest.raises(RuntimeError, match="not explicitly active"): _bind_preview(database, _structured_payload(_action_data(unknown), key="d" * 32)) def test_same_day_structured_events_and_linear_correction_cycles_are_supported(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row medication = connection.execute("SELECT id,medikament_name FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1").fetchone() for occurred_at, revision in (("2026-06-15T08:00", "d" * 64), ("2026-06-15T20:00", "e" * 64)): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,actual_dose_value,actual_dose_unit, route_normalized,source,occurred_at,business_revision) VALUES('2026-06-15',?,'administered',?,'10','mg','oral','fixture',?,?)""", (medication["medikament_name"], medication["id"], occurred_at, revision)) connection.commit() assert connection.execute("SELECT COUNT(*) FROM medication_administrations WHERE medication_id=? AND datum='2026-06-15' AND business_revision IS NOT NULL", (medication["id"],)).fetchone()[0] == 2 connection.close() def test_preview_binds_complete_payload_and_plan_can_only_be_consumed_once(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) history = dispatch_api(database, "/api/v1/medications", "") planned = history["planned"][0] prescription = next(item for item in history["prescriptions"] if item["id"] == planned["prescription_id"]) payload = _structured_payload(_action_data(prescription)) payload["data"]["planned_event_ref"] = planned["id"] payload["data"]["planned_dose_value"] = planned["planned_dose_value"] or "" payload["data"]["planned_dose_unit"] = planned["planned_dose_unit"] or "" payload = _bind_preview(database, payload) original_revision = payload["data"]["preview_revision"] changed = json.loads(json.dumps(payload)) changed["data"]["actual_dose_value"] = "11" connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row try: assert resolve_action_preview(connection, changed)[0] != original_revision finally: connection.close() worker.DASHBOARD_DB = database worker.DASHBOARD_V5_FILE = None worker.CAPTURE_QUARANTINE = tmp_path / "quarantine" worker.CAPTURE_MEDIA = tmp_path / "media" worker.apply_capture_action(payload) replay = json.loads(json.dumps(payload)) replay["idempotency_key"] = "9" * 32 with pytest.raises(RuntimeError, match="already consumed"): _bind_preview(database, replay) def test_default_doctor_report_uses_minimal_medication_projection(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) report = dispatch_api( database, "/api/v1/doctor-report", "from=2026-01-01&to=2026-08-21§ions=medications", ) serialized = json.dumps(report["medications"], sort_keys=True) for forbidden in ( "note", "lot_number", "injection_region", "injection_side", "injection_detail", "route_original", "status_provenance", "preview_revision", "prescription_id", "corrects_event_id", ): assert forbidden not in serialized assert report["medications"]["unknown"] assert report["completeness"]["medications"]["status"] == "documented" def test_migration_rejects_alias_paths_and_keeps_failed_artifacts_private(tmp_path: Path) -> None: source = tmp_path / "legacy.db" _legacy_db(source) with pytest.raises(RuntimeError, match="already exists|pairwise distinct"): safe_prepare(source, tmp_path / "same.db", tmp_path / "same.db", tmp_path / "restore.db") assert not (tmp_path / "same.db").exists() corrupt = tmp_path / "corrupt.db" corrupt.write_bytes(b"not sqlite") with pytest.raises((RuntimeError, sqlite3.DatabaseError)): safe_prepare(corrupt, tmp_path / "bad-backup.db", tmp_path / "bad-migrated.db", tmp_path / "bad-restore.db") assert not (tmp_path / "bad-backup.db").exists() assert not (tmp_path / "bad-migrated.db").exists() assert not (tmp_path / "bad-restore.db").exists() def test_public_tokens_are_keyed_and_invalid_filters_fail_closed(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) payload = dispatch_api(database, "/api/v1/medications", "") serialized = json.dumps(payload, sort_keys=True) assert '"business_revision"' not in serialized assert all( isinstance(item["id"], str) and item["id"].startswith("medrx_") and len(item["id"]) == 30 for item in payload["prescriptions"] ) for bucket in ("planned", "administered", "missed", "corrected", "unknown"): for event in payload[bucket]: assert isinstance(event["id"], str) and event["id"].startswith("medevt_") and len(event["id"]) == 31 assert event["prescription_id"] is None or ( event["prescription_id"].startswith("medrx_") and len(event["prescription_id"]) == 30 ) for relation in ("planned_event_id", "corrects_event_id"): assert event[relation] is None or ( event[relation].startswith("medevt_") and len(event[relation]) == 31 ) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row row = connection.execute("SELECT * FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1").fetchone() connection.close() public = next(item for item in payload["current_prescriptions"] if item["name"] == row["medikament_name"]) assert public["preview_revision"] != action_context_revision(row) for candidate_id in range(1, 101): candidate = dict(row) candidate["id"] = candidate_id assert public["preview_revision"] != action_context_revision(candidate) with pytest.raises(APIError) as invalid: dispatch_api(database, "/api/v1/medications", "source=https%3A%2F%2Fevil.example") assert invalid.value.status == 400 script = (ROOT / "scripts/health/assets/health-assets/dashboard-v5-record.js").read_text() assert "url.searchParams.set('medication'" not in script assert "tab === 'medications' && key === 'medication'" in script assert "fetch('/api/v1/capture/csrf'" in script assert "csrf_token:frozenCsrf" in script def test_correction_dose_semantics_are_status_specific(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) prescription = dispatch_api(database, "/api/v1/medications", "")["current_prescriptions"][0] target = "medevt_" + "a" * 24 def correction(target_status: str) -> dict: data = _action_data(prescription, status="corrected") data.update({ "correction_target_ref": target, "corrected_target_status": target_status, "correction_reason": "synthetic correction", "planned_dose_value": "", "planned_dose_unit": "", "actual_dose_value": "", "actual_dose_unit": "", }) return data administered = correction("administered") administered.update({"actual_dose_value": "10", "actual_dose_unit": "mg"}) validate_capture_payload(_structured_payload(administered, key="3" * 32)) administered["actual_dose_value"] = "" with pytest.raises(ValueError, match="administered correction"): validate_capture_payload(_structured_payload(administered, key="4" * 32)) planned = correction("planned") planned.update({"planned_dose_value": "10", "planned_dose_unit": "mg"}) validate_capture_payload(_structured_payload(planned, key="5" * 32)) planned["actual_dose_value"] = "10" planned["actual_dose_unit"] = "mg" with pytest.raises(ValueError, match="planned correction"): validate_capture_payload(_structured_payload(planned, key="6" * 32)) missed = correction("missed") validate_capture_payload(_structured_payload(missed, key="7" * 32)) missed["planned_dose_value"] = "10" missed["planned_dose_unit"] = "mg" with pytest.raises(ValueError, match="cannot assert dose"): validate_capture_payload(_structured_payload(missed, key="8" * 32)) def test_report_marks_correction_chain_and_next_plan_excludes_consumed(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row prescription = connection.execute("SELECT * FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1").fetchone() medication_id = int(prescription["id"]) name = str(prescription["medikament_name"]) original = connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,actual_dose_value,actual_dose_unit, route_original,route_normalized,source,occurred_at,business_revision) VALUES('2026-06-20',?,'administered',?,'10','mg','oral','oral','fixture','2026-06-20T10:00',?)""", (name, medication_id, "1" * 64)).lastrowid first = connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,route_original,route_normalized,source,occurred_at,business_revision) VALUES('2026-06-21',?,'corrected',?,?, 'missed','synthetic','', 'unknown','fixture','2026-06-21T10:00',?)""", (name, medication_id, original, "2" * 64)).lastrowid connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,actual_dose_value,actual_dose_unit,route_original,route_normalized,source,occurred_at,business_revision) VALUES('2026-06-22',?,'corrected',?,?, 'administered','synthetic','10','mg','oral','oral','fixture','2026-06-22T10:00',?)""", (name, medication_id, first, "3" * 64)) plan = connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_dose_value,planned_dose_unit, route_original,route_normalized,source,occurred_at,business_revision) VALUES('2026-09-01',?,'planned',?,'10','mg','oral','oral','fixture','2026-09-01T09:00',?)""", (name, medication_id, "4" * 64)).lastrowid connection.commit() assert any(item["date"] == "2026-09-01" for item in _next_planned_medications(connection, date(2026, 8, 21))) connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_event_id,route_original,route_normalized, source,occurred_at,business_revision) VALUES('2026-09-01',?,'missed',?,?,'','unknown','fixture','2026-09-01T10:00',?)""", (name, medication_id, plan, "5" * 64)) connection.commit() assert not any(item["date"] == "2026-09-01" for item in _next_planned_medications(connection, date(2026, 8, 21))) connection.close() report = dispatch_api(database, "/api/v1/doctor-report", "from=2026-06-01&to=2026-08-21§ions=medications") effective = [item for item in report["medications"]["administered"] if item["date"] == "2026-06-22"] assert len(effective) == 1 assert effective[0]["is_correction"] is True assert effective[0]["is_latest_effective"] is True assert report["medications"]["corrected"] == [] assert not any(item["date"] == "2026-06-20" for item in report["medications"]["administered"]) def test_active_status_requires_revision_source_and_provenance(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) row = connection.execute("SELECT id FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1").fetchone() connection.execute("UPDATE medikamente SET business_revision=NULL,prescription_status_provenance=NULL WHERE id=?", row) connection.commit() connection.close() history = dispatch_api(database, "/api/v1/medications", "") affected = next(item for item in history["prescriptions"] if item["status"] == "unknown") assert affected not in history["current_prescriptions"] def test_rebuild_preserves_legacy_schema_objects_and_replaces_stale_managed_triggers(tmp_path: Path) -> None: database = tmp_path / "legacy.db" _legacy_db(database) connection = sqlite3.connect(database) connection.execute("CREATE INDEX ix_legacy_medication_source ON medication_administrations(source)") connection.execute("""CREATE TRIGGER trg_legacy_medication_note AFTER INSERT ON medication_administrations BEGIN UPDATE medication_administrations SET notes=COALESCE(NEW.notes,'') WHERE id=NEW.id; END""") apply_schema(connection) names = {row[0] for row in connection.execute( "SELECT name FROM sqlite_master WHERE type IN ('index','trigger')" )} assert "ix_legacy_medication_source" in names assert "trg_legacy_medication_note" in names connection.execute("DROP TRIGGER trg_medication_event_validate_insert") connection.execute("""CREATE TRIGGER trg_medication_event_validate_insert BEFORE INSERT ON medication_administrations WHEN 0 BEGIN SELECT 1; END""") with pytest.raises(RuntimeError, match="trigger contract stale"): assert_schema(connection) apply_schema(connection) apply_schema(connection) assert_schema(connection) with pytest.raises(sqlite3.IntegrityError, match="invalid medication relation"): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,business_revision) VALUES('2026-08-21','invalid','planned',999,?)""", ("a" * 64,)) connection.close() def test_apply_schema_rolls_back_managed_ddl_failure_without_committing_caller(tmp_path: Path) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) medication_id, name = connection.execute( "SELECT id,medikament_name FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1" ).fetchone() origin = connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,actual_dose_value,actual_dose_unit, route_normalized,source,occurred_at,business_revision) VALUES('2026-08-20',?,'administered',?,'1','mg','oral','fixture','2026-08-20T10:00',?)""", (name, medication_id, "d" * 64)).lastrowid connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-21',?,'corrected',?,?, 'missed','first','unknown','fixture','2026-08-21T10:00',?)""", (name, medication_id, origin, "e" * 64)) connection.commit() connection.execute("DROP INDEX ux_medication_correction_origin") connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'corrected',?,?, 'planned','parallel','unknown','fixture','2026-08-22T10:00',?)""", (name, medication_id, origin, "f" * 64)) before = connection.execute("""SELECT type,name,sql FROM sqlite_master WHERE (type='trigger' AND name LIKE 'trg_medication_%') OR name IN ('ix_medication_event_medication_time','ix_medication_event_plan','ux_medication_plan_effective_consumption') ORDER BY type,name""").fetchall() assert connection.in_transaction with pytest.raises(sqlite3.IntegrityError): apply_schema(connection) assert connection.in_transaction after = connection.execute("""SELECT type,name,sql FROM sqlite_master WHERE (type='trigger' AND name LIKE 'trg_medication_%') OR name IN ('ix_medication_event_medication_time','ix_medication_event_plan','ux_medication_plan_effective_consumption') ORDER BY type,name""").fetchall() assert after == before assert connection.execute( "SELECT COUNT(*) FROM medication_administrations WHERE corrects_event_id=?", (origin,) ).fetchone()[0] == 2 connection.rollback() connection.close() def test_next_plan_uses_latest_effective_correction_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: database = tmp_path / "health.db" build_dashboard_v5_fixture(database) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row prescription = connection.execute("SELECT id,medikament_name FROM medikamente WHERE prescription_status='active' ORDER BY id LIMIT 1").fetchone() medication_id, name = int(prescription["id"]), str(prescription["medikament_name"]) def plan(day: str, revision: str) -> int: return int(connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_dose_value,planned_dose_unit, route_normalized,source,occurred_at,business_revision) VALUES(?,?,'planned',?,'10','mg','oral','fixture',?||'T09:00',?)""", (day, name, medication_id, day, revision)).lastrowid) corrected_plan = plan("2026-09-02", "6" * 64) connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'corrected',?,?, 'missed','synthetic','unknown','fixture','2026-08-22T09:00',?)""", (name, medication_id, corrected_plan, "7" * 64)) restored_plan = plan("2026-09-03", "8" * 64) consumed = int(connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_event_id,route_normalized, source,occurred_at,business_revision) VALUES('2026-08-22',?,'missed',?,?,'unknown','fixture','2026-08-22T10:00',?)""", (name, medication_id, restored_plan, "9" * 64)).lastrowid) restored_latest = int(connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,planned_dose_value,planned_dose_unit,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'corrected',?,?, 'planned','synthetic','10','mg','oral','fixture','2026-08-22T11:00',?)""", (name, medication_id, consumed, "a" * 64)).lastrowid) connection.commit() with pytest.raises(sqlite3.IntegrityError, match="already effectively consumed"): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_event_id,actual_dose_value, actual_dose_unit,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'administered',?,?,'10','mg','oral','fixture','2026-08-22T09:30',?)""", (name, medication_id, corrected_plan, "d" * 64)) dates = {item["date"] for item in _next_planned_medications(connection, date(2026, 8, 21), limit=20)} assert "2026-09-02" not in dates assert "2026-09-03" in dates connection.close() public = dispatch_api(database, "/api/v1/medications", "") prescription_public = next(item for item in public["current_prescriptions"] if item["name"] == name) restored_public = next(item for item in public["planned"] if item["date"] == "2026-09-03") monkeypatch.setattr(worker, "DASHBOARD_DB", database) monkeypatch.setattr(worker, "DASHBOARD_V5_FILE", None) monkeypatch.setattr(worker, "CAPTURE_QUARANTINE", tmp_path / "quarantine") monkeypatch.setattr(worker, "CAPTURE_MEDIA", tmp_path / "media") action = _action_data(prescription_public) action["planned_event_ref"] = restored_public["id"] payload = _bind_preview(database, _structured_payload(action, key="b" * 32)) worker.apply_capture_action(payload) connection = sqlite3.connect(database) assert connection.execute( "SELECT COUNT(*) FROM medication_administrations WHERE planned_event_id=?", (restored_plan,), ).fetchone()[0] == 2 with pytest.raises(sqlite3.IntegrityError, match="already effectively consumed"): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,corrects_event_id,corrected_target_status, correction_reason,actual_dose_value,actual_dose_unit,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'corrected',?,?, 'administered','reactivate old chain','10','mg','oral','fixture','2026-08-22T11:30',?)""", (name, medication_id, restored_latest, "c" * 64)) with pytest.raises(sqlite3.IntegrityError, match="already effectively consumed"): connection.execute("""INSERT INTO medication_administrations( datum,medication_name,event_type,medication_id,planned_event_id,actual_dose_value, actual_dose_unit,route_normalized,source,occurred_at,business_revision) VALUES('2026-08-22',?,'administered',?,?,'10','mg','oral','fixture','2026-08-22T12:00',?)""", (name, medication_id, restored_plan, "c" * 64)) connection.close()