diff --git a/app.py b/app.py index fcb25e5..634442b 100644 --- a/app.py +++ b/app.py @@ -1,5 +1,6 @@ """Flask app that serves the web GUI for browsing sessions.""" +import argparse import os import sys @@ -35,10 +36,8 @@ def index(): return app -def build_cli_parser(): +def build_cli_parser() -> argparse.ArgumentParser: """CLI argument parser for ``python app.py`` (stdlib only; safe to import in tests).""" - import argparse - parser = argparse.ArgumentParser(description="Claude Code Chat Browser") parser.add_argument("--port", type=int, default=5000) parser.add_argument("--host", default="127.0.0.1") diff --git a/models/errors.py b/models/errors.py index 6af19dd..40c3eb3 100644 --- a/models/errors.py +++ b/models/errors.py @@ -1,7 +1,16 @@ -"""HTTP error response shapes.""" +"""HTTP error response shapes and domain validation errors.""" from typing import TypedDict class ErrorResponse(TypedDict): error: str + + +class SessionValidationError(ValueError): + """Raised when parsed JSONL output does not match SessionDict contract.""" + + def __init__(self, path: str, detail: str) -> None: + self.path = path + self.detail = detail + super().__init__(f"Session validation failed at {path}: {detail}") diff --git a/tests/test_jsonl_validation.py b/tests/test_jsonl_validation.py new file mode 100644 index 0000000..c9899a2 --- /dev/null +++ b/tests/test_jsonl_validation.py @@ -0,0 +1,86 @@ +"""Runtime validation at the JSONL → SessionDict boundary.""" + +import os +import sys +from typing import Any + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from models.errors import SessionValidationError # noqa: E402 +from utils.jsonl_parser import parse_session # noqa: E402 +from utils.validation import validate_session_dict # noqa: E402 + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") + + +def _valid_payload(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "session_id": "abc123", + "title": "Test Session", + "messages": [{"role": "user", "text": "hello"}], + "metadata": {"session_id": "abc123"}, + } + base.update(overrides) + return base + + +class TestValidateSessionDict: + def test_missing_session_id(self): + payload = _valid_payload() + del payload["session_id"] + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict(payload) + assert exc_info.value.path == "session_id" + + def test_wrong_type_session_id(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict(_valid_payload(session_id=123)) + assert exc_info.value.path == "session_id" + + def test_null_session_id(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict(_valid_payload(session_id=None)) + assert exc_info.value.path == "session_id" + assert exc_info.value.detail == "must not be null" + + def test_null_role_in_message(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict( + _valid_payload(messages=[{"role": None, "text": "x"}]) + ) + assert exc_info.value.path == "messages[0].role" + + def test_missing_role_in_message(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict( + _valid_payload(messages=[{"text": "no role key"}]) + ) + assert exc_info.value.path == "messages[0].role" + + def test_metadata_not_dict(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict(_valid_payload(metadata="not-a-dict")) + assert exc_info.value.path == "metadata" + + def test_message_not_dict(self): + with pytest.raises(SessionValidationError) as exc_info: + validate_session_dict(_valid_payload(messages=["not-a-dict"])) + assert exc_info.value.path == "messages[0]" + + def test_valid_payload_returns_session_dict(self): + result = validate_session_dict(_valid_payload()) + assert result["session_id"] == "abc123" + assert result["messages"][0]["role"] == "user" + + +class TestParseSessionValidationRegression: + def test_session_minimal_fixture_unchanged(self): + path = os.path.join(FIXTURES, "session_minimal.jsonl") + session = parse_session(path) + assert session["session_id"] == "session_minimal" + assert session["title"] == "Hello from integration fixture" + assert len(session["messages"]) == 2 + assert session["messages"][0]["role"] == "user" + assert session["messages"][1]["role"] == "assistant" diff --git a/utils/jsonl_parser.py b/utils/jsonl_parser.py index 51ce8d9..14b71c3 100644 --- a/utils/jsonl_parser.py +++ b/utils/jsonl_parser.py @@ -4,9 +4,10 @@ import json import os from datetime import datetime -from typing import Any, cast +from typing import Any from models.session import MessageDict, QuickSessionInfoDict, SessionDict +from utils.validation import validate_session_dict def parse_session(filepath: str) -> SessionDict: @@ -120,14 +121,13 @@ def parse_session(filepath: str) -> SessionDict: title = _infer_title(messages) - return cast( - SessionDict, + return validate_session_dict( { "session_id": session_id, "title": title, "messages": messages, "metadata": metadata, - }, + } ) diff --git a/utils/validation.py b/utils/validation.py new file mode 100644 index 0000000..3d7dcf7 --- /dev/null +++ b/utils/validation.py @@ -0,0 +1,56 @@ +"""Runtime validation for TypedDict shapes at untrusted-data boundaries.""" + +from typing import Any, cast + +from models.errors import SessionValidationError +from models.session import SessionDict + +_ROOT_PATH = "(root)" + + +def _require_field( + container: dict[str, Any], + key: str, + expected_type: type[Any], + type_label: str, + *, + path: str | None = None, +) -> Any: + field_path = path or key + if key not in container: + raise SessionValidationError(field_path, "missing required field") + return _require_value(field_path, container[key], expected_type, type_label) + + +def _require_value( + path: str, + val: Any, + expected_type: type[Any], + type_label: str, +) -> Any: + if val is None: + raise SessionValidationError(path, "must not be null") + if not isinstance(val, expected_type): + raise SessionValidationError( + path, f"expected {type_label}, got {type(val).__name__}" + ) + return val + + +def validate_session_dict(data: dict[str, Any]) -> SessionDict: + """Validate a plain dict matches SessionDict before returning it.""" + # Runtime guard for dynamic callers; mypy already types the parameter as dict. + if not isinstance(data, dict): + raise SessionValidationError(_ROOT_PATH, "expected dict") + + _require_field(data, "session_id", str, "str") + _require_field(data, "title", str, "str") + messages = _require_field(data, "messages", list, "list") + _require_field(data, "metadata", dict, "dict") + + for index, message in enumerate(messages): + path = f"messages[{index}]" + msg_dict = _require_value(path, message, dict, "dict") + _require_field(msg_dict, "role", str, "str", path=f"{path}.role") + + return cast(SessionDict, data)