"""Unit tests for src/providers/ using fixture files.""" import json import logging from pathlib import Path import pytest from src.blocks import ( BLOCK_TYPE_FILE_PLACEHOLDER, BLOCK_TYPE_HIDDEN_CONTEXT_MARKER, BLOCK_TYPE_IMAGE_PLACEHOLDER, BLOCK_TYPE_TEXT, BLOCK_TYPE_THINKING, BLOCK_TYPE_TOOL_RESULT, BLOCK_TYPE_TOOL_USE, BLOCK_TYPE_UNKNOWN, render_blocks_to_markdown, ) from src.loss_report import LossReport FIXTURES = Path(__file__).parent / "fixtures" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _block_types(message: dict) -> list[str]: return [b.get("type") for b in (message.get("blocks") or [])] def _first_block(message: dict, block_type: str) -> dict | None: for b in message.get("blocks") or []: if b.get("type") == block_type: return b return None # --------------------------------------------------------------------------- # ChatGPT # --------------------------------------------------------------------------- class TestChatGPTNormalization: """ChatGPT normalize_conversation block-extraction behavior.""" def _get_provider(self): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) import requests p._session = requests.Session() p._org_id = None p._project_ids = [] p._project_map = {} p._project_name_cache = {} return p def test_normalizes_conversation(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assert result["id"] == "chatgpt-conv-001" assert result["title"] == "Python Async Tutorial" assert result["provider"] == "chatgpt" assert result["project"] is None assert result["created_at"] != "" assert result["updated_at"] != "" assert isinstance(result["messages"], list) def test_normalizes_without_project(self): raw = json.loads((FIXTURES / "chatgpt_no_project.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assert result["project"] is None assert result["id"] == "chatgpt-conv-002" def test_normalizes_with_project_from_map(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() p._project_map["chatgpt-conv-001"] = "My Research Project" result = p.normalize_conversation(raw) assert result["project"] == "My Research Project" def test_text_message_emits_text_block(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) user_msgs = [m for m in result["messages"] if m["role"] == "user"] # The "How does async/await..." message async_msgs = [ m for m in user_msgs if any( "async" in (b.get("text") or "").lower() for b in (m.get("blocks") or []) ) ] assert async_msgs, "expected a user message about async/await" assert _block_types(async_msgs[0]) == [BLOCK_TYPE_TEXT] def test_code_block_preserved_with_language(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assistant_msgs = [m for m in result["messages"] if m["role"] == "assistant"] # The first assistant message is the async/await answer with a python fence text_block = _first_block(assistant_msgs[0], BLOCK_TYPE_TEXT) assert text_block is not None assert "```python" in text_block["text"] def test_multimodal_voice_user_message(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) # node-mm-user: audio_transcription "What is the capital of France?" # + real_time_user_audio_video_asset_pointer wrapping a sediment:// URL capital_msgs = [ m for m in result["messages"] if any( "capital of france" in (b.get("text") or "").lower() for b in (m.get("blocks") or []) ) ] assert capital_msgs, "expected the audio_transcription text to surface" types = _block_types(capital_msgs[0]) assert BLOCK_TYPE_TEXT in types assert BLOCK_TYPE_FILE_PLACEHOLDER in types file_block = _first_block(capital_msgs[0], BLOCK_TYPE_FILE_PLACEHOLDER) assert file_block["ref"].startswith("sediment://") assert file_block["mime"] == "audio/wav" assert file_block["size_bytes"] == 50000 assert file_block["duration_seconds"] == pytest.approx(2.5) def test_multimodal_voice_reverse_order_preserved(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) # node-mm-user-rev has parts in REVERSE order: asset first, transcription second. rev_msgs = [ m for m in result["messages"] if any( "tell me more" in (b.get("text") or "").lower() for b in (m.get("blocks") or []) ) ] assert rev_msgs, "expected the reverse-order voice message" types = _block_types(rev_msgs[0]) # Order preserved: file_placeholder before text assert types == [BLOCK_TYPE_FILE_PLACEHOLDER, BLOCK_TYPE_TEXT] def test_image_only_user_message_renders(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) image_msgs = [ m for m in result["messages"] if any(b.get("type") == BLOCK_TYPE_IMAGE_PLACEHOLDER for b in (m.get("blocks") or [])) ] assert image_msgs, "image-only user message should now render" def test_user_editable_context_emits_blocks(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() p._hidden_content = "full" result = p.normalize_conversation(raw) # The user_editable_context message has user_profile + user_instructions. # It should now appear (was silently dropped pre-v0.4.0). uec_msgs = [ m for m in result["messages"] if any( "Custom Instructions" in (b.get("text") or "") for b in (m.get("blocks") or []) ) ] assert uec_msgs, "user_editable_context should be visible in output" # Hidden context marker should be prepended. assert uec_msgs[0]["blocks"][0]["type"] == BLOCK_TYPE_HIDDEN_CONTEXT_MARKER def test_user_editable_context_uses_safe_fence(self): """The user_instructions value contains embedded triple-backticks; the rendered Markdown must use a fence longer than 3 backticks so embedded fences are inert. """ from src.blocks import render_blocks_to_markdown raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() p._hidden_content = "full" result = p.normalize_conversation(raw) uec_msgs = [ m for m in result["messages"] if any( "Custom Instructions" in (b.get("text") or "") for b in (m.get("blocks") or []) ) ] assert uec_msgs rendered = render_blocks_to_markdown(uec_msgs[0]["blocks"]) # Content has ``` inside, so the wrap fence must be at least 4 backticks. assert "````" in rendered, "expected a 4+ backtick safe-fence wrap" def test_message_roles_are_valid(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) for msg in result["messages"]: assert msg["role"] in ("user", "assistant", "system", "tool") def test_message_count_matches(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assert result["message_count"] == len(result["messages"]) def test_loss_report_records_messages(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() report = LossReport() result = p.normalize_conversation(raw, report) assert report.messages_rendered == len(result["messages"]) assert report.conversations == 1 class TestChatGPTUnknownContent: """Unrecognised content types should produce visible unknown blocks + WARNING + tally.""" def _get_provider(self): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) import requests p._session = requests.Session() p._org_id = None p._project_ids = [] p._project_map = {} p._project_name_cache = {} return p def _make_unknown_conv(self): return { "id": "test-unknown", "title": "Test", "create_time": 1700000000.0, "update_time": 1700000001.0, "mapping": { "root": {"id": "root", "message": None, "parent": None, "children": ["msg1"]}, "msg1": { "id": "msg1", "message": { "id": "msg1", "author": {"role": "user"}, "content": { "content_type": "future_unknown_type_xyz", "some_field": "value", }, }, "parent": "root", "children": [], }, }, } def test_unknown_content_type_produces_unknown_block(self): p = self._get_provider() result = p.normalize_conversation(self._make_unknown_conv()) assert any( b.get("type") == BLOCK_TYPE_UNKNOWN for m in result["messages"] for b in (m.get("blocks") or []) ) def test_unknown_content_type_logs_warning(self, caplog): p = self._get_provider() with caplog.at_level(logging.WARNING): p.normalize_conversation(self._make_unknown_conv()) assert any("future_unknown_type_xyz" in r.message for r in caplog.records) def test_unknown_content_type_increments_loss_report(self): p = self._get_provider() report = LossReport() p.normalize_conversation(self._make_unknown_conv(), report) assert report.unknown_blocks["future_unknown_type_xyz"] == 1 class TestChatGPTHiddenContentPolicy: """EXPORTER_HIDDEN_CONTENT: collapse retrieval dumps and hidden context. file_search dumps re-inject attached-file text on every tool run (measured 86% of content bytes on real data) and are NOT hidden-flagged โ€” author.name is the discriminator. Hidden-flagged messages (Custom Instructions) are the secondary case. """ def _get_provider(self, policy=None): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) import requests p._session = requests.Session() p._org_id = None p._project_ids = [] p._project_map = {} p._project_name_cache = {} if policy is not None: p._hidden_content = policy return p def _make_conv(self): return { "id": "test-collapse", "title": "T", "create_time": 1700000000.0, "update_time": 1700000001.0, "mapping": { "root": {"id": "root", "message": None, "parent": None, "children": ["dump"]}, "dump": { "id": "dump", "parent": "root", "children": ["answer"], "message": { "id": "dump", "author": {"role": "tool", "name": "file_search"}, "content": { "content_type": "multimodal_text", "parts": ["Full text of an attached file " * 100], }, }, }, "answer": { "id": "answer", "parent": "dump", "children": ["uec"], "message": { "id": "answer", "author": {"role": "assistant"}, "content": {"content_type": "text", "parts": ["The answer."]}, }, }, "uec": { "id": "uec", "parent": "answer", "children": [], "message": { "id": "uec", "author": {"role": "user"}, "content": { "content_type": "user_editable_context", "user_profile": "Preferred name: Jesse", "user_instructions": "Be concise.", }, "metadata": {"is_visually_hidden_from_conversation": True}, }, }, }, } def test_default_policy_is_placeholder(self, monkeypatch): from src.blocks import BLOCK_TYPE_COLLAPSED monkeypatch.delenv("EXPORTER_HIDDEN_CONTENT", raising=False) # No _hidden_content attribute โ†’ env fallback โ†’ placeholder default. result = self._get_provider().normalize_conversation(self._make_conv()) collapsed = [ b for m in result["messages"] for b in m["blocks"] if b.get("type") == BLOCK_TYPE_COLLAPSED ] assert len(collapsed) == 2 def test_placeholder_collapses_file_search_dump(self): from src.blocks import BLOCK_TYPE_COLLAPSED, COLLAPSED_KIND_TOOL_DUMP result = self._get_provider("placeholder").normalize_conversation(self._make_conv()) tool_msgs = [m for m in result["messages"] if m["role"] == "tool"] assert len(tool_msgs) == 1 (block,) = tool_msgs[0]["blocks"] assert block["type"] == BLOCK_TYPE_COLLAPSED assert block["kind"] == COLLAPSED_KIND_TOOL_DUMP assert block["origin"] == "file_search" assert block["size_bytes"] > 1000 def test_placeholder_collapses_hidden_flagged_message(self): from src.blocks import BLOCK_TYPE_COLLAPSED, COLLAPSED_KIND_HIDDEN_CONTEXT result = self._get_provider("placeholder").normalize_conversation(self._make_conv()) user_msgs = [m for m in result["messages"] if m["role"] == "user"] assert len(user_msgs) == 1 (block,) = user_msgs[0]["blocks"] assert block["type"] == BLOCK_TYPE_COLLAPSED assert block["kind"] == COLLAPSED_KIND_HIDDEN_CONTEXT assert block["origin"] == "user_editable_context" def test_placeholder_keeps_normal_messages(self): result = self._get_provider("placeholder").normalize_conversation(self._make_conv()) assistant_msgs = [m for m in result["messages"] if m["role"] == "assistant"] assert assistant_msgs[0]["blocks"][0]["type"] == BLOCK_TYPE_TEXT assert assistant_msgs[0]["blocks"][0]["text"] == "The answer." def test_non_retrieval_tool_authors_not_collapsed(self): """web.run / browser / python tool messages are real content โ€” keep.""" from src.blocks import BLOCK_TYPE_COLLAPSED conv = self._make_conv() conv["mapping"]["dump"]["message"]["author"]["name"] = "web.run" conv["mapping"]["dump"]["message"]["content"] = { "content_type": "text", "parts": ["Search results summary."], } result = self._get_provider("placeholder").normalize_conversation(conv) tool_msgs = [m for m in result["messages"] if m["role"] == "tool"] assert tool_msgs[0]["blocks"][0]["type"] != BLOCK_TYPE_COLLAPSED def test_omit_drops_messages_entirely(self): result = self._get_provider("omit").normalize_conversation(self._make_conv()) assert [m["role"] for m in result["messages"]] == ["assistant"] def test_full_preserves_v040_behavior(self): result = self._get_provider("full").normalize_conversation(self._make_conv()) tool_msgs = [m for m in result["messages"] if m["role"] == "tool"] assert tool_msgs[0]["blocks"][0]["type"] == BLOCK_TYPE_TEXT user_msgs = [m for m in result["messages"] if m["role"] == "user"] assert user_msgs[0]["blocks"][0]["type"] == BLOCK_TYPE_HIDDEN_CONTEXT_MARKER def test_collapsed_tallied_in_loss_report(self): report = LossReport() self._get_provider("placeholder").normalize_conversation(self._make_conv(), report) assert report.collapsed["file_search"] == 1 assert report.collapsed["user_editable_context"] == 1 assert report.collapsed_bytes > 1000 summary = report.format_summary() assert "collapsed by policy: 2" in summary assert "EXPORTER_HIDDEN_CONTENT=full to keep" in summary def test_collapsed_block_renders_one_line_with_size(self): from src.blocks import BLOCK_TYPE_COLLAPSED result = self._get_provider("placeholder").normalize_conversation(self._make_conv()) tool_msgs = [m for m in result["messages"] if m["role"] == "tool"] rendered = render_blocks_to_markdown(tool_msgs[0]["blocks"]) assert rendered.startswith("> ๐Ÿ”ง **Tool output** โ€” `file_search` (") assert "KB" in rendered assert "omitted (EXPORTER_HIDDEN_CONTENT=full to keep)" in rendered assert "\n" not in rendered def test_invalid_env_value_falls_back_to_placeholder(self, monkeypatch, caplog): from src.providers.chatgpt import resolve_hidden_content_policy monkeypatch.setenv("EXPORTER_HIDDEN_CONTENT", "bogus") with caplog.at_level(logging.WARNING): assert resolve_hidden_content_policy() == "placeholder" assert any("EXPORTER_HIDDEN_CONTENT" in r.message for r in caplog.records) class TestRequestPacing: """REQUEST_DELAY politeness pacing between consecutive API requests.""" def _get_provider(self, delay): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) p._request_delay = delay p._last_request_at = None return p def test_first_request_not_delayed(self, monkeypatch): sleeps = [] monkeypatch.setattr("src.providers.base.time.sleep", lambda s: sleeps.append(s)) p = self._get_provider(2.0) p._pace() assert sleeps == [] def test_consecutive_requests_paced_with_jitter_window(self, monkeypatch): sleeps = [] monkeypatch.setattr("src.providers.base.time.sleep", lambda s: sleeps.append(s)) monkeypatch.setattr("src.providers.base.random.uniform", lambda a, b: 1.0) clock = {"now": 100.0} monkeypatch.setattr("src.providers.base.time.monotonic", lambda: clock["now"]) p = self._get_provider(2.0) p._pace() clock["now"] = 100.5 # only 0.5s elapsed since last request p._pace() assert sleeps == [pytest.approx(1.5)] def test_no_sleep_when_enough_time_elapsed(self, monkeypatch): sleeps = [] monkeypatch.setattr("src.providers.base.time.sleep", lambda s: sleeps.append(s)) monkeypatch.setattr("src.providers.base.random.uniform", lambda a, b: 1.0) clock = {"now": 100.0} monkeypatch.setattr("src.providers.base.time.monotonic", lambda: clock["now"]) p = self._get_provider(1.0) p._pace() clock["now"] = 105.0 p._pace() assert sleeps == [] def test_zero_delay_disables_pacing(self, monkeypatch): sleeps = [] monkeypatch.setattr("src.providers.base.time.sleep", lambda s: sleeps.append(s)) p = self._get_provider(0) p._pace() p._pace() assert sleeps == [] def test_pace_noop_without_init_attrs(self): """Providers built via __new__ in tests have no pacing attrs โ€” must not crash.""" from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) p._pace() # no exception def test_resolve_request_delay_default_and_overrides(self, monkeypatch, caplog): from src.providers.base import resolve_request_delay, DEFAULT_REQUEST_DELAY monkeypatch.delenv("REQUEST_DELAY", raising=False) assert resolve_request_delay() == DEFAULT_REQUEST_DELAY monkeypatch.setenv("REQUEST_DELAY", "2.5") assert resolve_request_delay() == 2.5 monkeypatch.setenv("REQUEST_DELAY", "0") assert resolve_request_delay() == 0.0 monkeypatch.setenv("REQUEST_DELAY", "abc") with caplog.at_level(logging.WARNING): assert resolve_request_delay() == DEFAULT_REQUEST_DELAY assert any("REQUEST_DELAY" in r.message for r in caplog.records) # --------------------------------------------------------------------------- # Claude # --------------------------------------------------------------------------- class TestClaudeNormalization: """Claude normalize_conversation block-extraction behavior.""" def _get_provider(self): from src.providers.claude import ClaudeProvider import requests p = ClaudeProvider.__new__(ClaudeProvider) p._session = requests.Session() p._org_id = None return p def test_normalizes_with_project(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assert result["id"] == "claude-conv-001" assert result["title"] == "StartOS Service Packaging" assert result["provider"] == "claude" assert result["project"] == "StarTOS Packaging" assert result["created_at"] == "2024-06-10T14:32:00.000Z" def test_normalizes_without_project(self): raw = json.loads((FIXTURES / "claude_no_project.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assert result["project"] is None def test_string_content_emits_text_block(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) thanks_msgs = [ m for m in result["messages"] if any( "thank you" in (b.get("text") or "").lower() for b in (m.get("blocks") or []) ) ] assert thanks_msgs def test_list_content_emits_blocks_in_order(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assistant_msgs = [m for m in result["messages"] if m["role"] == "assistant"] # msg-002 has text + tool_use, in that order. assert assistant_msgs types = _block_types(assistant_msgs[0]) assert BLOCK_TYPE_TEXT in types assert BLOCK_TYPE_TOOL_USE in types # Order preserved assert types.index(BLOCK_TYPE_TEXT) < types.index(BLOCK_TYPE_TOOL_USE) def test_tool_use_block_fields(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) assistant_msgs = [m for m in result["messages"] if m["role"] == "assistant"] tool_block = _first_block(assistant_msgs[0], BLOCK_TYPE_TOOL_USE) assert tool_block["name"] == "search" assert tool_block["input"] == {"query": "startOS docs"} assert tool_block["tool_id"] == "tool-001" def test_image_block_emits_image_placeholder(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) msg004 = [ m for m in result["messages"] if any(b.get("type") == BLOCK_TYPE_IMAGE_PLACEHOLDER for b in (m.get("blocks") or [])) ] assert msg004 img = _first_block(msg004[0], BLOCK_TYPE_IMAGE_PLACEHOLDER) assert img["ref"] == "claude-image-uuid-1" def test_unknown_block_type_records_loss(self): from src.blocks import BLOCK_TYPE_UNKNOWN as _UNK raw = { "uuid": "test-unknown", "name": "T", "chat_messages": [ { "uuid": "m1", "sender": "human", "content": [{"type": "future_block_xyz", "data": "..."}], } ], } p = self._get_provider() report = LossReport() result = p.normalize_conversation(raw, report) assert any( b.get("type") == _UNK for m in result["messages"] for b in (m.get("blocks") or []) ) assert report.unknown_blocks["future_block_xyz"] == 1 def test_thinking_block(self): raw = { "uuid": "thinking-test", "name": "T", "chat_messages": [ { "uuid": "m1", "sender": "assistant", "content": [ {"type": "thinking", "thinking": "Let me reason about this."}, {"type": "text", "text": "Here's the answer."}, ], } ], } p = self._get_provider() result = p.normalize_conversation(raw) types = _block_types(result["messages"][0]) assert BLOCK_TYPE_THINKING in types assert BLOCK_TYPE_TEXT in types def test_tool_result_with_nested_text_blocks(self): raw = { "uuid": "tool-result-test", "name": "T", "chat_messages": [ { "uuid": "m1", "sender": "assistant", "content": [ { "type": "tool_result", "tool_use_id": "tool-001", "content": [ {"type": "text", "text": "search hit 1"}, {"type": "text", "text": "search hit 2"}, ], "is_error": False, } ], } ], } p = self._get_provider() result = p.normalize_conversation(raw) tool_result = _first_block(result["messages"][0], BLOCK_TYPE_TOOL_RESULT) assert tool_result is not None assert "search hit 1" in tool_result["output"] assert "search hit 2" in tool_result["output"] assert tool_result["is_error"] is False def test_human_sender_maps_to_user(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) roles = {m["role"] for m in result["messages"]} assert "user" in roles assert "human" not in roles def test_loss_report_messages_recorded(self): raw = json.loads((FIXTURES / "claude_conversation.json").read_text()) p = self._get_provider() report = LossReport() result = p.normalize_conversation(raw, report) assert report.messages_rendered == len(result["messages"]) # --------------------------------------------------------------------------- # v0.4.1 โ€” execution_output, system_error, tether_browsing_display, conv_id # --------------------------------------------------------------------------- class TestChatGPTToolOutputs: """v0.4.1 ChatGPT tool-role content_types map onto tool_result blocks.""" def _get_provider(self): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) import requests p._session = requests.Session() p._org_id = None p._project_ids = [] p._project_map = {} p._project_name_cache = {} return p def test_execution_output_emits_tool_result_with_metadata(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) exec_msgs = [ m for m in result["messages"] if any( b.get("type") == BLOCK_TYPE_TOOL_RESULT and b.get("tool_name") == "container.exec" for b in (m.get("blocks") or []) ) ] assert exec_msgs, "expected execution_output to render as tool_result" block = next( b for b in exec_msgs[0]["blocks"] if b.get("type") == BLOCK_TYPE_TOOL_RESULT ) assert block["output"].startswith("Hello from container.exec") assert block["is_error"] is False assert block["summary"] == "Reading skill documentation" def test_execution_output_message_role_is_tool(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) tool_msgs = [m for m in result["messages"] if m["role"] == "tool"] assert tool_msgs, "tool-role messages must pass through (filter lifted in v0.4.0)" def test_empty_execution_output_skipped(self, caplog): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() with caplog.at_level(logging.DEBUG, logger="src.providers.chatgpt"): result = p.normalize_conversation(raw) # The empty execution_output (author.name="python") must NOT appear. python_msgs = [ m for m in result["messages"] if any( b.get("type") == BLOCK_TYPE_TOOL_RESULT and b.get("tool_name") == "python" for b in (m.get("blocks") or []) ) ] assert not python_msgs, "empty execution_output should be skipped" assert any("Skipping empty execution_output" in r.message for r in caplog.records) def test_system_error_emits_error_tool_result(self): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() result = p.normalize_conversation(raw) web_err = [ m for m in result["messages"] if any( b.get("type") == BLOCK_TYPE_TOOL_RESULT and b.get("tool_name") == "web" and b.get("is_error") is True for b in (m.get("blocks") or []) ) ] assert web_err, "system_error should render as tool_result with is_error=True" block = next(b for b in web_err[0]["blocks"] if b.get("tool_name") == "web") assert "503" in block["output"] def test_tether_browsing_display_spinner_skipped(self, caplog): raw = json.loads((FIXTURES / "chatgpt_conversation.json").read_text()) p = self._get_provider() with caplog.at_level(logging.DEBUG, logger="src.providers.chatgpt"): result = p.normalize_conversation(raw) spinner_msgs = [ m for m in result["messages"] if any( b.get("type") == BLOCK_TYPE_TOOL_RESULT and b.get("tool_name") == "file_search" for b in (m.get("blocks") or []) ) ] assert not spinner_msgs, "spinner tether_browsing_display should be skipped" assert any("tether_browsing_display spinner" in r.message for r in caplog.records) def test_tether_browsing_display_populated_renders_defensively(self): """Defensive case (never observed in real data) โ€” populated browse renders.""" conv = { "id": "test-tether", "title": "T", "create_time": 1700000000.0, "update_time": 1700000001.0, "mapping": { "root": {"id": "root", "message": None, "parent": None, "children": ["m1"]}, "m1": { "id": "m1", "parent": "root", "children": [], "message": { "id": "m1", "author": {"role": "tool", "name": "browser"}, "content": { "content_type": "tether_browsing_display", "result": "Found 3 results about kubernetes ingress.", "summary": "ingress search", "assets": None, "tether_id": None, }, }, }, }, } p = self._get_provider() result = p.normalize_conversation(conv) assert any( b.get("type") == BLOCK_TYPE_TOOL_RESULT and b.get("tool_name") == "browser" for m in result["messages"] for b in (m.get("blocks") or []) ) class TestChatGPTConvIdFallback: """v0.4.1: live ChatGPT detail responses use conversation_id, not id.""" def _get_provider(self): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) import requests p._session = requests.Session() p._org_id = None p._project_ids = [] p._project_map = {} p._project_name_cache = {} return p def test_falls_back_to_conversation_id(self): raw = { "conversation_id": "live-chatgpt-uuid", "title": "T", "create_time": 1700000000.0, "update_time": 1700000001.0, "mapping": { "root": {"id": "root", "message": None, "parent": None, "children": []}, }, } p = self._get_provider() result = p.normalize_conversation(raw) assert result["id"] == "live-chatgpt-uuid" def test_id_takes_precedence_when_both_present(self): raw = { "id": "from-id", "conversation_id": "from-conversation-id", "title": "T", "create_time": 1700000000.0, "update_time": 1700000001.0, "mapping": { "root": {"id": "root", "message": None, "parent": None, "children": []}, }, } p = self._get_provider() result = p.normalize_conversation(raw) assert result["id"] == "from-id" # --------------------------------------------------------------------------- # ChatGPT session-token health (ยง9): /api/auth/session `error` is the signal, # not `expires`/`accessToken`. Verified live 2026-06-28 โ€” a dead token returns # HTTP 200 with error=RefreshAccessTokenError and a stale accessToken. # --------------------------------------------------------------------------- class _FakeResp: def __init__(self, payload, status=200): self._payload = payload self.status_code = status def raise_for_status(self): if self.status_code >= 400: raise RuntimeError(f"HTTP {self.status_code}") def json(self): return self._payload class _FakeSession: """Minimal stand-in for the provider's HTTP session.""" def __init__(self, payload=None, status=200, raises=None): self._payload = payload self._status = status self._raises = raises def get(self, *args, **kwargs): if self._raises is not None: raise self._raises return _FakeResp(self._payload, self._status) class TestChatGPTSessionHealth: def _provider(self, session): from src.providers.chatgpt import ChatGPTProvider p = ChatGPTProvider.__new__(ChatGPTProvider) p._session = session return p def test_fetch_access_token_returns_token_when_healthy(self): p = self._provider(_FakeSession({"accessToken": "tok-123", "error": None})) assert p._fetch_access_token() == "tok-123" def test_fetch_access_token_fails_fast_on_refresh_error(self): from src.providers.base import ProviderError # Dead token: HTTP 200, error set, stale accessToken still present. p = self._provider( _FakeSession({"accessToken": "stale", "error": "RefreshAccessTokenError"}) ) with pytest.raises(ProviderError) as exc: p._fetch_access_token() assert "expired" in str(exc.value.original).lower() def test_fetch_access_token_fails_when_no_token(self): from src.providers.base import ProviderError p = self._provider(_FakeSession({"error": None})) with pytest.raises(ProviderError): p._fetch_access_token() def test_session_health_ok_when_no_error(self): p = self._provider(_FakeSession({"accessToken": "tok", "error": None})) ok, detail = p.session_health() assert ok is True assert detail == "Session active" def test_session_health_reports_expired_on_error(self): p = self._provider( _FakeSession({"accessToken": "stale", "error": "RefreshAccessTokenError"}) ) ok, detail = p.session_health() assert ok is False assert "RefreshAccessTokenError" in detail assert "refresh" in detail.lower() def test_session_health_does_not_use_rolling_expires(self): # A far-future `expires` must NOT make a dead token look healthy. p = self._provider( _FakeSession( { "accessToken": "stale", "error": "RefreshAccessTokenError", "expires": "2026-09-26T05:07:31.108Z", } ) ) ok, _ = p.session_health() assert ok is False def test_session_health_graceful_on_network_error(self): p = self._provider(_FakeSession(raises=RuntimeError("boom"))) ok, detail = p.session_health() assert ok is False assert "unreachable" in detail.lower()