FFFF.F.FFF.FF.FFFFF.F [100%] =================================== FAILURES =================================== ________ test_f1_schema_and_hyrimoz_preset_are_additive_and_idempotent _________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_f1_schema_and_hyrimoz_pre0') def test_f1_schema_and_hyrimoz_preset_are_additive_and_idempotent(tmp_path: Path) -> None: > database, _, rx = _case(tmp_path) ^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f1.py:72: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_dashboard_v5_sprint7c_f1.py:26: in _case api = dispatch_api(database, "/api/v1/medications", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError _______ test_exact_unknown_status_historical_hyrimoz_capture_and_replay ________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_exact_unknown_status_hist0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7bc49e4c3990> def test_exact_unknown_status_historical_hyrimoz_capture_and_replay(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: > database, _, rx = _case(tmp_path) ^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f1.py:95: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_dashboard_v5_sprint7c_f1.py:26: in _case api = dispatch_api(database, "/api/v1/medications", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ________ test_preset_match_needs_no_deviation_but_actual_deviation_does ________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_preset_match_needs_no_dev0') def test_preset_match_needs_no_deviation_but_actual_deviation_does(tmp_path: Path) -> None: > database, _, rx = _case(tmp_path) ^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f1.py:167: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_dashboard_v5_sprint7c_f1.py:26: in _case api = dispatch_api(database, "/api/v1/medications", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError _____________ test_historical_and_real_plan_modes_are_fail_closed ______________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_historical_and_real_plan_0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7bc49daecbd0> def test_historical_and_real_plan_modes_are_fail_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: > database, _, rx = _case(tmp_path) ^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f1.py:183: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ tests/test_dashboard_v5_sprint7c_f1.py:26: in _case api = dispatch_api(database, "/api/v1/medications", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError __ test_copy_first_migration_preserves_every_legacy_value_and_proves_restore ___ tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_copy_first_migration_pres0') 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) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:81: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/migrate_sprint7c_f_medication_schema.py:296: in safe_prepare copy_changes, copy_after = _migrate_and_verify( scripts/health/migrate_sprint7c_f_medication_schema.py:233: in _migrate_and_verify assert_schema(connection) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError ______ test_worker_replay_uses_capture_action_log_and_stale_preview_fails ______ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_worker_replay_uses_captur0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7bc49d4cf550> 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:197: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError _____ test_new_reader_separates_statuses_and_old_reader_remains_compatible _____ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_new_reader_separates_stat0') 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:237: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ________ test_preview_revision_changes_when_bound_prescription_changes _________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_preview_revision_changes_0') 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:257: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ____ test_preview_binds_complete_payload_and_plan_can_only_be_consumed_once ____ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_preview_binds_complete_pa0') 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:283: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ________ test_default_doctor_report_uses_minimal_medication_projection _________ connection = params = {'from': '2026-01-01', 'to': '2026-08-21'} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_default_doctor_report_use0') 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", ) tests/test_dashboard_v5_sprint7c_f.py:314: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4682: in dispatch_api return _doctor_report( scripts/health/dashboard_v5/read_api.py:4224: in _doctor_report _record_medications( _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = params = {'from': '2026-01-01', 'to': '2026-08-21'} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError _________ test_public_tokens_are_keyed_and_invalid_filters_fail_closed _________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_public_tokens_are_keyed_a0') 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:348: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ______________ test_correction_dose_semantics_are_status_specific ______________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_correction_dose_semantics0') 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] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:388: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError ______ test_report_marks_correction_chain_and_next_plan_excludes_consumed ______ connection = params = {'from': '2026-06-01', 'to': '2026-08-21'} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_report_marks_correction_c0') 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") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:466: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4682: in dispatch_api return _doctor_report( scripts/health/dashboard_v5/read_api.py:4224: in _doctor_report _record_medications( _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = params = {'from': '2026-06-01', 'to': '2026-08-21'} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError __________ test_active_status_requires_revision_source_and_provenance __________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_active_status_requires_re0') 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:484: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError _ test_rebuild_preserves_legacy_schema_objects_and_replaces_stale_managed_triggers _ tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_rebuild_preserves_legacy_0') 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) tests/test_dashboard_v5_sprint7c_f.py:510: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError ____________ test_next_plan_uses_latest_effective_correction_state _____________ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: > assert_medication_schema(connection) scripts/health/dashboard_v5/read_api.py:3545: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = def assert_schema(connection: sqlite3.Connection) -> None: _require_base(connection) missing: dict[str, list[str]] = {} for table, expected in ( ("medikamente", PRESCRIPTION_COLUMNS), ("medication_administrations", EVENT_COLUMNS), ): actual = _columns(connection, table) absent = [name for name, _ in expected if name not in actual] if absent: missing[table] = absent marker = connection.execute( "SELECT schema_version FROM medication_schema_meta WHERE migration_name=?", (MIGRATION_NAME,), ).fetchone() if missing or marker is None or int(marker[0]) != SCHEMA_VERSION: raise RuntimeError("medication-history schema missing") key = connection.execute( "SELECT key FROM medication_public_identity_key WHERE singleton=1" ).fetchone() if key is None or not isinstance(key[0], bytes) or len(key[0]) != 32: raise RuntimeError("medication public identity key missing") indexes = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='index' AND name IN (?,?,?,?)", tuple(sorted(MANAGED_INDEX_NAMES)), ) } if "ux_medication_plan_effective_consumption" in indexes: raise RuntimeError("obsolete static plan-consumption index remains") for name, expected_digest in MANAGED_INDEX_SHA256.items(): if name not in indexes: raise RuntimeError(f"medication-history index contract missing: {name}") actual_digest = hashlib.sha256(_normalized_sql(indexes[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: raise RuntimeError(f"medication-history index contract stale: {name}") triggers = { str(row[0]): str(row[1] or "") for row in connection.execute( "SELECT name,sql FROM sqlite_master WHERE type='trigger' AND name LIKE 'trg_medication_%'" ) } required = set(MANAGED_TRIGGER_SHA256) if not required <= set(triggers): raise RuntimeError("medication-history trigger contract missing") for name, expected_digest in MANAGED_TRIGGER_SHA256.items(): actual_digest = hashlib.sha256(_normalized_sql(triggers[name]).encode("utf-8")).hexdigest() if actual_digest != expected_digest: > raise RuntimeError(f"medication-history trigger contract stale: {name}") E RuntimeError: medication-history trigger contract stale: trg_medication_event_validate_insert scripts/health/dashboard_v5/medication_schema.py:590: RuntimeError The above exception was the direct cause of the following exception: tmp_path = PosixPath('/tmp/pytest-of-agent/pytest-234/test_next_plan_uses_latest_eff0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7bc49d6f7090> 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", "") ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/test_dashboard_v5_sprint7c_f.py:607: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ scripts/health/dashboard_v5/read_api.py:4630: in dispatch_api return _record_medications(connection, parse_query(query, {"from", "to", "medication", "status", "source"})) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ connection = , params = {} def _record_medications( connection: sqlite3.Connection, params: dict[str, str] ) -> dict[str, Any]: start, end = parse_range(params) medication_filter = params.get("medication", "") if medication_filter and not re.fullmatch(r"medrx_[a-f0-9]{24}", medication_filter): raise APIError(400, "medication_filter_not_allowed") status_filter = params.get("status", "") raw_source_filter = params.get("source", "") source_filter = safe_metadata_text(raw_source_filter, 80, allow_empty=True) if raw_source_filter and source_filter is None: raise APIError(400, "source_filter_not_allowed") if status_filter and status_filter not in {"planned", "administered", "missed", "corrected", "unknown"}: raise APIError(400, "medication_status_not_allowed") empty: dict[str, Any] = { "contract": MEDICATION_HISTORY_CONTRACT, "prescriptions": [], "current_prescriptions": [], "other_prescriptions": [], "planned": [], "administered": [], "missed": [], "corrected": [], "unknown": [], "sources": [], "truncated": False, "truncated_sections": [], "next_cursor": None, } if not table_exists(connection, "medication_administrations") or not table_exists(connection, "medikamente"): return empty try: assert_medication_schema(connection) except RuntimeError as error: > raise APIError(503, "medication_schema_unavailable") from error E dashboard_v5.read_api.APIError: medication_schema_unavailable scripts/health/dashboard_v5/read_api.py:3547: APIError =========================== short test summary info ============================ FAILED tests/test_dashboard_v5_sprint7c_f1.py::test_f1_schema_and_hyrimoz_preset_are_additive_and_idempotent FAILED tests/test_dashboard_v5_sprint7c_f1.py::test_exact_unknown_status_historical_hyrimoz_capture_and_replay FAILED tests/test_dashboard_v5_sprint7c_f1.py::test_preset_match_needs_no_deviation_but_actual_deviation_does FAILED tests/test_dashboard_v5_sprint7c_f1.py::test_historical_and_real_plan_modes_are_fail_closed FAILED tests/test_dashboard_v5_sprint7c_f.py::test_copy_first_migration_preserves_every_legacy_value_and_proves_restore FAILED tests/test_dashboard_v5_sprint7c_f.py::test_worker_replay_uses_capture_action_log_and_stale_preview_fails FAILED tests/test_dashboard_v5_sprint7c_f.py::test_new_reader_separates_statuses_and_old_reader_remains_compatible FAILED tests/test_dashboard_v5_sprint7c_f.py::test_preview_revision_changes_when_bound_prescription_changes FAILED tests/test_dashboard_v5_sprint7c_f.py::test_preview_binds_complete_payload_and_plan_can_only_be_consumed_once FAILED tests/test_dashboard_v5_sprint7c_f.py::test_default_doctor_report_uses_minimal_medication_projection FAILED tests/test_dashboard_v5_sprint7c_f.py::test_public_tokens_are_keyed_and_invalid_filters_fail_closed FAILED tests/test_dashboard_v5_sprint7c_f.py::test_correction_dose_semantics_are_status_specific FAILED tests/test_dashboard_v5_sprint7c_f.py::test_report_marks_correction_chain_and_next_plan_excludes_consumed FAILED tests/test_dashboard_v5_sprint7c_f.py::test_active_status_requires_revision_source_and_provenance FAILED tests/test_dashboard_v5_sprint7c_f.py::test_rebuild_preserves_legacy_schema_objects_and_replaces_stale_managed_triggers FAILED tests/test_dashboard_v5_sprint7c_f.py::test_next_plan_uses_latest_effective_correction_state 16 failed, 5 passed in 37.56s __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/repos/HealthManager__HERMES_CWD_8d46a20096ed__