Jesse.Markowitz/AIChatExporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
AIChatExporter/tests/test_providers.py
T
JesseMarkowitzandClaude Opus 4.8 1e016ea652 canary: detect provider API schema drift; release v0.7.0 
The export reads ChatGPT's and Claude's undocumented internal web APIs,
which can change shape without notice; the worst failure for a backup tool
is a silent one (skipped/mis-parsed content with no error). Add a `canary`
command + BaseProvider.check_drift() (overridden by ChatGPT/Claude) that
fetches one listing page + one conversation per provider and asserts only
the normalizer's load-bearing fields — not the full response shape, which
churns harmlessly. The top silent risk it guards is a renamed retrieval-tool
author bypassing the hidden-content collapse. ERROR findings exit non-zero;
WARN findings are surfaced but non-fatal so a backup run is never blocked.

Also retires the in-app watch mode and headless StartOS direction from the
roadmap (tool stays a local, manually-run CLI) and updates docs.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-28 01:37:24 -04:00

1119 lines
44 KiB
Python
Raw Permalink Blame History

"""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()
# ---------------------------------------------------------------------------
# API-drift canary (§10): assert load-bearing fields, flag silent risks.
# ---------------------------------------------------------------------------
def _sev(findings):
return {f["severity"] for f in findings}
def _checks(findings, severity):
return {f["check"] for f in findings if f["severity"] == severity}
class TestChatGPTDriftCanary:
def _provider(self, page, detail):
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p.list_conversations = lambda offset=0, limit=5: page
p.get_conversation = lambda cid: detail
return p
@staticmethod
def _detail(mapping):
return {
"conversation_id": "c1", "title": "T",
"create_time": 1.0, "update_time": 2.0, "mapping": mapping,
}
@staticmethod
def _node(role="user", name=None, content_type="text", parts=("hello",)):
return {
"id": "n", "parent": None, "children": [],
"message": {
"author": {"role": role, "name": name},
"content": {"content_type": content_type, "parts": list(parts)},
"metadata": {},
},
}
def test_healthy_shape_is_ok_only(self):
from src.providers.base import DRIFT_OK
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail({"n1": self._node()}))
assert _sev(p.check_drift()) == {DRIFT_OK}
def test_missing_listing_id_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"title": "T"}], self._detail({"n1": self._node()}))
assert DRIFT_ERROR in _sev(p.check_drift())
def test_empty_mapping_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail({}))
findings = p.check_drift()
assert "detail" in _checks(findings, DRIFT_ERROR)
def test_unfamiliar_tool_author_warns_collapse(self):
from src.providers.base import DRIFT_WARN
mapping = {
"n1": self._node(), # a healthy text msg so parts check passes
"n2": self._node(role="tool", name="brand_new_retriever"),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "collapse" in _checks(p.check_drift(), DRIFT_WARN)
def test_new_content_type_warns(self):
from src.providers.base import DRIFT_WARN
mapping = {
"n1": self._node(),
"n2": self._node(content_type="hologram_v2", parts=()),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "content_type" in _checks(p.check_drift(), DRIFT_WARN)
def test_known_tool_author_not_flagged(self):
# file_search is a known collapse author — must NOT warn.
mapping = {
"n1": self._node(),
"n2": self._node(role="tool", name="file_search"),
}
p = self._provider([{"id": "c1", "title": "T", "update_time": "x"}],
self._detail(mapping))
assert "collapse" not in _checks(p.check_drift(), "warn")
class TestClaudeDriftCanary:
def _provider(self, page, detail):
from src.providers.claude import ClaudeProvider
p = ClaudeProvider.__new__(ClaudeProvider)
p.list_conversations = lambda offset=0, limit=5: page
p.get_conversation = lambda cid: detail
return p
@staticmethod
def _detail(msgs):
return {"uuid": "u1", "name": "N", "created_at": "x", "updated_at": "y",
"chat_messages": msgs}
def test_healthy_flat_shape_is_ok(self):
from src.providers.base import DRIFT_OK
msgs = [{"uuid": "m1", "sender": "human", "text": "hi",
"created_at": "x", "attachments": [], "files": []}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert _sev(p.check_drift()) == {DRIFT_OK}
def test_content_as_list_warns(self):
from src.providers.base import DRIFT_WARN
msgs = [{"uuid": "m1", "sender": "assistant",
"content": [{"type": "text", "text": "hi"}], "created_at": "x"}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert "content" in _checks(p.check_drift(), DRIFT_WARN)
def test_nonempty_attachments_warn(self):
from src.providers.base import DRIFT_WARN
msgs = [{"uuid": "m1", "sender": "human", "text": "hi",
"created_at": "x", "attachments": [{"file_name": "a.pdf"}]}]
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail(msgs))
assert "attachments" in _checks(p.check_drift(), DRIFT_WARN)
def test_empty_messages_is_error(self):
from src.providers.base import DRIFT_ERROR
p = self._provider([{"uuid": "u1", "name": "N", "updated_at": "z"}],
self._detail([]))
assert DRIFT_ERROR in _sev(p.check_drift())
Reference in New Issue View Git Blame Copy Permalink