"""Unit: schema boundary caps for the saved-chat surface (phase 83, task 01). The first schema-boundary suite: every A1 cap in ``app/schemas.py`` is pinned at BOTH boundaries — a value exactly at the cap validates and one past it raises a pydantic ``ValidationError`` naming the offending field (the 422-shaped boundary response, house style, phase 56 precedent). Plus the regression pin: a realistic ``bor.chat.v1`` payload validates cleanly and round-trips ``model_dump()`` (the stored-shape contract — the caps added value bounds only, no key/shape change). """ from __future__ import annotations import pytest from pydantic import ValidationError from app.schemas import ( ChatMessage, SavedChatCreate, SavedChatUpdate, SourceRef, ToolCall, ) # --- boundary constants (phase 83, A1 — mirror sources in app/schemas.py) --- TEXT_CAP = 32_000 # mirrors HistoryTurn.text / .thinking SOURCES_CAP = 20 SUGGESTIONS_CAP = 50 CHIP_CAP = 200 TOOLS_CAP = 50 SOURCE_CAP = 120 # documents.source String(120) PATH_CAP = 1000 # documents.path String(1000) TITLE_CAP = 500 # documents.title String(500) NAME_CAP = 100 # ToolCall.name ARGUMENT_CAP = 2000 # ToolCall.argument MESSAGES_CAP = 200 # SavedChatCreate/Update.messages def _source_ref() -> dict: return {"source": "Homelab", "path": "kubernetes.md", "title": "K8s"} def _tool_call() -> dict: return {"name": "read", "argument": "Homelab/kubernetes.md"} def _user_message(text: str = "How did I install k3s?") -> ChatMessage: return ChatMessage.model_validate({"who": "user", "text": text}) def _failed_loc(exc: ValidationError, *loc: object) -> None: """Assert the first ``ValidationError`` names exactly the given field path (``loc`` tuple, e.g. ``("text",)`` or ``("messages", 0, "text")``) — the 422 must point at the offender.""" assert exc.errors()[0]["loc"] == tuple(loc), exc.errors() # --------------------------------------------------------------------------- # ChatMessage.text / .thinking # --------------------------------------------------------------------------- def test_chat_message_text_at_cap_validates() -> None: msg = _user_message("a" * TEXT_CAP) assert len(msg.text) == TEXT_CAP def test_chat_message_text_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: _user_message("a" * (TEXT_CAP + 1)) _failed_loc(exc.value, "text") def test_chat_message_thinking_at_cap_validates() -> None: msg = ChatMessage.model_validate({"who": "brain", "text": "ok", "thinking": "t" * TEXT_CAP}) assert len(msg.thinking or "") == TEXT_CAP def test_chat_message_thinking_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "thinking": "t" * (TEXT_CAP + 1)} ) _failed_loc(exc.value, "thinking") def test_chat_message_thinking_none_still_valid() -> None: assert _user_message().thinking is None assert ChatMessage.model_validate( {"who": "brain", "text": "ok", "thinking": None} ).thinking is None # --------------------------------------------------------------------------- # ChatMessage.sources / .suggestions / .tools # --------------------------------------------------------------------------- def test_chat_message_sources_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "sources": [_source_ref() for _ in range(SOURCES_CAP)]} ) assert len(msg.sources or []) == SOURCES_CAP def test_chat_message_sources_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( { "who": "brain", "text": "ok", "sources": [_source_ref() for _ in range(SOURCES_CAP + 1)], } ) _failed_loc(exc.value, "sources") def test_chat_message_suggestions_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": [f"chip {i}" for i in range(SUGGESTIONS_CAP)]} ) assert len(msg.suggestions or []) == SUGGESTIONS_CAP def test_chat_message_suggestions_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( { "who": "brain", "text": "ok", "suggestions": [f"chip {i}" for i in range(SUGGESTIONS_CAP + 1)], } ) _failed_loc(exc.value, "suggestions") def test_chat_message_suggestion_chip_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": ["c" * CHIP_CAP]} ) assert msg.suggestions is not None assert len(msg.suggestions[0]) == CHIP_CAP def test_chat_message_suggestion_chip_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "suggestions": ["c" * (CHIP_CAP + 1)]} ) # the offender is the ITEM (the list itself is well under max_items). _failed_loc(exc.value, "suggestions", 0) def test_chat_message_tools_at_cap_validates() -> None: msg = ChatMessage.model_validate( {"who": "brain", "text": "ok", "tools": [_tool_call() for _ in range(TOOLS_CAP)]} ) assert len(msg.tools or []) == TOOLS_CAP def test_chat_message_tools_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ChatMessage.model_validate( {"who": "brain", "text": "ok", "tools": [_tool_call() for _ in range(TOOLS_CAP + 1)]} ) _failed_loc(exc.value, "tools") # --------------------------------------------------------------------------- # SourceRef (documents column-length mirrors) # --------------------------------------------------------------------------- def test_source_ref_source_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "s" * SOURCE_CAP, "path": "a.md", "title": "A"} ) assert len(ref.source) == SOURCE_CAP def test_source_ref_source_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate({"source": "s" * (SOURCE_CAP + 1), "path": "a.md", "title": "A"}) _failed_loc(exc.value, "source") def test_source_ref_path_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "Homelab", "path": "p" * PATH_CAP, "title": "A"} ) assert len(ref.path) == PATH_CAP def test_source_ref_path_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate( {"source": "Homelab", "path": "p" * (PATH_CAP + 1), "title": "A"} ) _failed_loc(exc.value, "path") def test_source_ref_title_at_cap_validates() -> None: ref = SourceRef.model_validate( {"source": "Homelab", "path": "a.md", "title": "t" * TITLE_CAP} ) assert len(ref.title) == TITLE_CAP def test_source_ref_title_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SourceRef.model_validate( {"source": "Homelab", "path": "a.md", "title": "t" * (TITLE_CAP + 1)} ) _failed_loc(exc.value, "title") # --------------------------------------------------------------------------- # ToolCall # --------------------------------------------------------------------------- def test_tool_call_name_at_cap_validates() -> None: call = ToolCall.model_validate({"name": "n" * NAME_CAP, "argument": None}) assert len(call.name) == NAME_CAP def test_tool_call_name_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ToolCall.model_validate({"name": "n" * (NAME_CAP + 1), "argument": None}) _failed_loc(exc.value, "name") def test_tool_call_argument_at_cap_validates() -> None: call = ToolCall.model_validate({"name": "read", "argument": "a" * ARGUMENT_CAP}) assert len(call.argument or "") == ARGUMENT_CAP def test_tool_call_argument_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: ToolCall.model_validate({"name": "read", "argument": "a" * (ARGUMENT_CAP + 1)}) _failed_loc(exc.value, "argument") def test_tool_call_argument_none_still_valid() -> None: assert ToolCall.model_validate({"name": "ls", "argument": None}).argument is None assert ToolCall.model_validate({"name": "ls"}).argument is None # --------------------------------------------------------------------------- # SavedChatCreate / SavedChatUpdate .messages # --------------------------------------------------------------------------- def _minimal_messages(n: int) -> list[dict]: return [{"who": "user" if i % 2 == 0 else "brain", "text": f"m{i}"} for i in range(n)] def test_saved_chat_create_messages_at_cap_validates() -> None: payload = SavedChatCreate.model_validate( {"title": "big", "messages": _minimal_messages(MESSAGES_CAP)} ) assert len(payload.messages) == MESSAGES_CAP def test_saved_chat_create_messages_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SavedChatCreate.model_validate( {"title": "bigger", "messages": _minimal_messages(MESSAGES_CAP + 1)} ) _failed_loc(exc.value, "messages") def test_saved_chat_update_messages_at_cap_validates() -> None: payload = SavedChatUpdate.model_validate({"messages": _minimal_messages(MESSAGES_CAP)}) assert len(payload.messages) == MESSAGES_CAP def test_saved_chat_update_messages_one_over_cap_rejects() -> None: with pytest.raises(ValidationError) as exc: SavedChatUpdate.model_validate({"messages": _minimal_messages(MESSAGES_CAP + 1)}) _failed_loc(exc.value, "messages") # --------------------------------------------------------------------------- # JSON-shape regression: same KEYS as before, value bounds only # --------------------------------------------------------------------------- def test_extra_keys_still_forbidden() -> None: """``extra="forbid"`` is untouched — a stray key is still a 422-shaped rejection (the accepted/rejected KEYS did not change with phase 83).""" with pytest.raises(ValidationError) as exc: ChatMessage.model_validate({"who": "user", "text": "hi", "html": "x"}) _failed_loc(exc.value, "html") def test_minimal_message_still_validates() -> None: """Optional keys may be ABSENT exactly as pre-phase-83.""" msg = ChatMessage.model_validate({"who": "user", "text": "hi"}) assert (msg.sources, msg.deflected, msg.suggestions, msg.thinking, msg.tools, msg.stopped) == ( None, None, None, None, None, None, ) # --------------------------------------------------------------------------- # Realistic-payload round-trip (the stored-shape contract) # --------------------------------------------------------------------------- def test_realistic_bor_chat_v1_payload_round_trips() -> None: """A full ``bor.chat.v1``-shaped record (4–8 messages mixing user/brain, one brain message with ``thinking`` + ``tools`` + ``sources``, one with ``suggestions`` + ``stopped``) validates cleanly and ``model_dump()`` of the messages equals the input dict — ``None``-keys preserved (the phase-50/51 byte-identical stored shape; the caps added value bounds only, no key/shape change).""" messages: list[dict] = [ { "who": "user", "text": "How did I install k3s on the new node?", "sources": None, "deflected": None, "suggestions": None, "thinking": None, "tools": None, "stopped": None, }, { "who": "brain", "text": "Your k3s cluster runs on three nodes — here's how it happened.", "sources": [ {"source": "Homelab", "path": "kubernetes.md", "title": "Kubernetes Cluster"}, {"source": "Deployments", "path": "k3s-install.md", "title": "k3s Install Notes"}, ], "deflected": False, "suggestions": None, "thinking": "The kubernetes doc covers the cluster layout…", "tools": [ {"name": "read", "argument": "Homelab/kubernetes.md"}, {"name": "ls", "argument": None}, ], "stopped": None, }, { "who": "user", "text": "And what ports does Traefik expose?", "sources": None, "deflected": None, "suggestions": None, "thinking": None, "tools": None, "stopped": None, }, { "who": "brain", "text": "Traefik exposes 80/443 on every node.", "sources": None, "deflected": None, "suggestions": ["What is the Traefik dashboard password?"], "thinking": None, "tools": None, "stopped": True, # the owner stopped the generation mid-answer }, ] payload = SavedChatCreate.model_validate({"title": "k3s install", "messages": messages}) dumped = payload.model_dump() # the stored shape is byte-identical: every message dict — None-keys # and all — round-trips unchanged (and the top-level record keeps its # ``share`` default). assert dumped["messages"] == messages assert dumped["share"] is False def test_realistic_payload_round_trips_through_update_model() -> None: """Same record re-Saved through ``SavedChatUpdate`` (the re-Save upsert surface) — identical stored shape.""" msg = { "who": "brain", "text": "answer", "sources": [_source_ref()], "deflected": None, "suggestions": ["follow-up?"], "thinking": "scratchpad", "tools": [_tool_call()], "stopped": None, } payload = SavedChatUpdate.model_validate({"messages": [msg]}) assert payload.model_dump()["messages"] == [msg]