from __future__ import annotations import importlib.util import sqlite3 import sys from datetime import date from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts" / "health")) from dashboard_v5 import read_api # noqa: E402 from dashboard_v5.read_api import dispatch_api # noqa: E402 fixture_spec = importlib.util.spec_from_file_location("dashboard_v5_fixture_b2", ROOT / "tests/fixtures/dashboard_v5_fixture.py") assert fixture_spec and fixture_spec.loader fixture_module = importlib.util.module_from_spec(fixture_spec) fixture_spec.loader.exec_module(fixture_module) def load_module(name: str, path: Path): spec = importlib.util.spec_from_file_location(name, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module @pytest.fixture() def fixture_db(tmp_path: Path) -> Path: db = tmp_path / "fixture.db" fixture_module.build_dashboard_v5_fixture(db, anchor_date=date(2026, 6, 15)) return db def test_patient_forms_validate_strict_shapes_without_direct_database_write(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: dashboard = tmp_path / "v5.html" dashboard.write_text("") monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_V5_FILE", str(dashboard)) monkeypatch.delenv("HEALTH_DASHBOARD_DB", raising=False) server = load_module("health_dashboard_server_b2_contract", ROOT / "scripts/health/health_dashboard_server.py") monkeypatch.setattr(server, "local_today", lambda: date(2026, 6, 15)) symptom = {"csrf_token":["x"],"return_to":["v5"],"occurred_at":["2026-06-15T08:30"],"symptom_type":["headache"],"severity":["2"],"onset_at":[""],"duration_minutes":["45"],"label":[""],"notes":["synthetic note"]} payload, target = server.validate_patient_action(symptom, server.SYMPTOM_EVENT_ROUTE) assert target == "v5" and payload["action"] == "symptom_event" and payload["symptom_type"] == "headache" bad = {**symptom, "notes": ["https://invalid.example"]} with pytest.raises(ValueError): server.validate_patient_action(bad, server.SYMPTOM_EVENT_ROUTE) def test_worker_applies_additive_patient_events_without_changing_daily_21_score(fixture_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("HEALTH_DASHBOARD_DB", str(fixture_db)) monkeypatch.delenv("HEALTH_DASHBOARD_V5_FILE", raising=False) worker = load_module("health_dashboard_worker_b2", ROOT / "scripts/health/health_dashboard_action_worker.py") migration = load_module("health_dashboard_migration_b3", ROOT / "scripts/health/migrate_patient_action_schema.py") with sqlite3.connect(fixture_db) as migration_connection: assert migration.apply_migration(migration_connection) with sqlite3.connect(fixture_db) as connection: before = connection.execute("SELECT count(*) FROM symptom_log WHERE kontext='daily_quick_score'").fetchone()[0] medication = connection.execute("SELECT medication_name FROM medication_administrations WHERE trim(medication_name)<>'' LIMIT 1").fetchone()[0] actions = [ {"version":1,"action":"symptom_event","occurred_at":"2026-06-15T08:30","symptom_type":"headache","severity":2,"onset_at":"","duration_minutes":45,"label":"","notes":"synthetic"}, {"version":1,"action":"medication_event","occurred_at":"2026-06-15T09:00","event_type":"administered","medication_name":medication,"dose":"1","unit":"unit","route":"oral","notes":"synthetic"}, {"version":1,"action":"general_event","occurred_at":"2026-06-15T10:00","category":"stress","label":"Synthetic event","intensity":1,"notes":"synthetic"}, ] for action in actions: worker.apply_patient_action(worker.validate_patient_payload(action)) with sqlite3.connect(fixture_db) as connection: assert connection.execute("SELECT count(*) FROM symptom_log WHERE kontext='daily_quick_score'").fetchone()[0] == before assert connection.execute("SELECT count(*) FROM symptom_log WHERE kontext='additional_symptom' AND symptom='headache'").fetchone()[0] == 1 assert connection.execute("SELECT count(*) FROM medication_administrations WHERE occurred_at='2026-06-15T09:00'").fetchone()[0] == 1 assert connection.execute("SELECT count(*) FROM health_events WHERE occurred_at='2026-06-15T10:00'").fetchone()[0] == 1 def test_doctor_summary_is_neutral_and_contains_new_bounded_sections(fixture_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(read_api, "local_today", lambda: date(2026, 6, 15)) report = dispatch_api(fixture_db, "/api/v1/doctor-report", "from=2026-06-01&to=2026-06-15") assert set(report["selected_sections"]) == set(read_api.REPORT_SECTIONS) assert "provenance" not in report assert {"nutrition", "observations", "timeline", "temporal_observations", "additional_symptoms"} <= set(report) assert "Kausalitäts" in report["medical_statement"] assert all("keine Kausalitätsaussage" in item["statement"] for item in report["temporal_observations"]) assert len(report["observations"]) <= 6 def test_events_endpoint_exposes_compact_filterable_domains(fixture_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(read_api, "local_today", lambda: date(2026, 6, 15)) payload = dispatch_api(fixture_db, "/api/v1/events", "from=2026-06-01&to=2026-06-15") assert set(payload["types"]) == set(read_api.ALLOWED_EVENT_TYPES) assert set(read_api.ALLOWED_EVENT_TYPES) >= {"symptom_day", "medication_administered", "nutrition_day", "laboratory", "document", "appointment"} assert all(set(item) <= {"date", "end_date", "type", "category", "label"} for item in payload["events"])