Jesse.Markowitz/AIChatExporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
55b9ce12f641cb205f81fb54d84e5efaafad670a
AIChatExporter/tests/test_providers.py
T
JesseMarkowitz 710889b65f fix: read project attribution from the conversation's gizmo_id 
A conversation in a project absent from CHATGPT_PROJECT_IDS exported into
no-project/ even though its own payload names the project. Found while
investigating the media 403s: bpi-f3-case-options sits in
g-p-6a4edf5160848191927b05da20f49151, which is not among the 13 configured
ids, so it filed under no-project.2026.

Both existing sources are bounded by what the user configured. The detail
response is not: it carries gizmo_id. Use it as a third fallback, after the
listing annotation and the project map, and cache the result into the map.
Attribution now stays correct with no list to maintain, and moving a chat
into a new project stops silently misfiling it.

Only g-p- ids are treated as projects — a custom GPT is not a project and
must not become a folder.

CHATGPT_PROJECT_IDS still matters for the listing pass: conversations that
live only inside a project never appear in the default listing, so an
unconfigured project's chats can be missed entirely. Each one is now
reported once per run with the id to add, which turns "some chats are
missing" into a line to paste.

8 tests cover precedence, the g-p- guard, map caching and the once-per-run
report. 318 pass.
2026-08-17 12:49:04 -04:00

1382 lines
54 KiB
Python
Raw 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())
# ---------------------------------------------------------------------------
# 4xx diagnostics: curl_cffi renders raise_for_status() as
# "HTTP Error {code}: {reason}", and HTTP/2 has no reason phrase — so a bare
# 403 logged as "HTTP Error 403:" says nothing. The body carries the cause.
# ---------------------------------------------------------------------------
class TestErrorBodyDiagnostics:
class _Resp:
ok = False
status_code = 403
reason = ""
headers: dict = {}
def __init__(self, payload=None, text=""):
self._payload = payload
self.text = text
def json(self):
if self._payload is None:
raise ValueError("not json")
return self._payload
def _provider(self, response):
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p._request_delay = 0
p._last_request_at = None
p._session = type("S", (), {"request": lambda *a, **k: response})()
return p
def test_detail_field_surfaces_in_error(self):
from src.providers.base import ProviderError
resp = self._Resp(payload={"detail": "File not accessible to this account"})
with pytest.raises(ProviderError) as exc:
self._provider(resp)._make_request("GET", "https://x/files/f1/download")
message = str(exc.value.original)
assert "403" in message
assert "File not accessible to this account" in message
def test_non_json_body_excerpted(self):
from src.providers.base import ProviderError
resp = self._Resp(text="<html>Forbidden</html>")
with pytest.raises(ProviderError) as exc:
self._provider(resp)._make_request("GET", "https://x/files/f1/download")
assert "Forbidden" in str(exc.value.original)
def test_empty_body_says_so_rather_than_nothing(self):
from src.providers.base import ProviderError
resp = self._Resp(text="")
with pytest.raises(ProviderError) as exc:
self._provider(resp)._make_request("GET", "https://x/files/f1/download")
assert "empty response body" in str(exc.value.original)
def test_secrets_in_error_body_are_redacted(self):
from src.providers.base import _describe_error_body
resp = self._Resp(payload={"error": {"message": "no", "access_token": "sk-abc"}})
described = _describe_error_body(resp)
assert "sk-abc" not in described
assert "[REDACTED]" in described
# ---------------------------------------------------------------------------
# Deleted assets: ChatGPT's download endpoint answers a missing upload with
# 403 Forbidden, not 404. Measured 2026-08-17 over 18 such assets — every one
# returned 404 "File not found" on /files/{id}, while assets that downloaded
# fine returned 200 on both in the same session.
# ---------------------------------------------------------------------------
class TestDeletedAssetReporting:
class _Resp:
def __init__(self, status, payload=None):
self.status_code = status
self.ok = 200 <= status < 400
self.reason = ""
self.headers: dict = {}
self.text = ""
self._payload = payload or {}
def json(self):
return self._payload
def _provider(self, responses):
"""responses: dict of url-substring → _Resp, consumed by substring match."""
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p._request_delay = 0
p._last_request_at = None
calls: list[str] = []
def request(method, url, **kwargs):
calls.append(url)
for fragment, resp in responses.items():
if url.endswith(fragment):
return resp
raise AssertionError(f"unexpected URL: {url}")
p._session = type("S", (), {"request": staticmethod(request)})()
p._calls = calls
return p
def test_403_confirmed_missing_is_reported_as_gone(self):
from src.providers.base import ProviderError
p = self._provider({
"/download": self._Resp(403, {"detail": "Forbidden"}),
"/files/file_x": self._Resp(404, {"detail": "File not found"}),
})
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
message = str(exc.value.original)
assert "no longer exists" in message
assert "not recoverable" in message
assert any(u.endswith("/files/file_x") for u in p._calls), "probe not sent"
def test_403_on_an_asset_that_still_exists_stays_a_403(self):
"""Don't call a live asset deleted — that would hide a real problem."""
from src.providers.base import ProviderError
p = self._provider({
"/download": self._Resp(403, {"detail": "Forbidden"}),
"/files/file_x": self._Resp(200, {"id": "file_x"}),
})
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
message = str(exc.value.original)
assert "403" in message
assert "no longer exists" not in message
def test_probe_failure_does_not_mask_the_original_403(self):
"""A probe that errors must leave the 403 intact, not swallow it."""
from src.providers.base import ProviderError
p = self._provider({"/download": self._Resp(403, {"detail": "Forbidden"})})
# No entry for /files/file_x — the fake session raises, standing in
# for a network error on the probe.
with pytest.raises(ProviderError) as exc:
p.download_asset("sediment://file_x")
assert "403" in str(exc.value.original)
assert "no longer exists" not in str(exc.value.original)
def test_no_probe_on_the_happy_path(self):
"""The probe costs a request — it must not fire on a good download."""
class _Bytes:
status_code = 200
content = b"data"
headers = {"content-type": "image/png"}
p = self._provider({
"/download": self._Resp(200, {"download_url": "https://cdn/x", "file_name": "a.png"}),
"https://cdn/x": _Bytes(),
})
content, mime, name = p.download_asset("sediment://file_x")
assert (content, mime, name) == (b"data", "image/png", "a.png")
assert not any(u.endswith("/files/file_x") for u in p._calls), "probed needlessly"
def test_classified_as_expired_not_forbidden(self):
"""The run summary must not call a deleted upload a permissions error."""
from src.media import _classify_failure
from src.providers.base import ProviderError
err = ProviderError(
"chatgpt",
"download_asset(file_x)",
RuntimeError(
"Asset no longer exists — HTTP 404 'File not found' on /files/{id}. "
"The upload was deleted or expired server-side; it is not "
"recoverable from ChatGPT."
),
)
assert _classify_failure(err) == "expired-or-missing"
# ---------------------------------------------------------------------------
# Project attribution: a conversation names its own project via gizmo_id, so
# it should not depend on the user having listed that project in config.
# ---------------------------------------------------------------------------
class TestProjectAttribution:
def _provider(self, *, project_ids=None, gizmo_name="My Project"):
from src.providers.chatgpt import ChatGPTProvider
p = ChatGPTProvider.__new__(ChatGPTProvider)
p._project_map = {}
p._project_name_cache = {}
p._project_ids = project_ids or []
p._hidden_content = "placeholder"
p._fetch_project_name = lambda gid: gizmo_name
return p
def _raw(self, **extra):
raw = {
"conversation_id": "conv-1",
"title": "T",
"create_time": 1_700_000_000,
"update_time": 1_700_000_000,
"mapping": {},
}
raw.update(extra)
return raw
def test_gizmo_id_supplies_project_when_unconfigured(self):
p = self._provider(gizmo_name="Tech Questions")
out = p.normalize_conversation(self._raw(gizmo_id="g-p-abc123"))
assert out["project"] == "Tech Questions"
def test_explicit_annotation_still_wins(self):
p = self._provider(gizmo_name="From Gizmo")
out = p.normalize_conversation(
self._raw(gizmo_id="g-p-abc123", _project_name="From Listing")
)
assert out["project"] == "From Listing"
def test_project_map_beats_gizmo_lookup(self):
p = self._provider(gizmo_name="From Gizmo")
p._project_map["conv-1"] = "From Map"
out = p.normalize_conversation(self._raw(gizmo_id="g-p-abc123"))
assert out["project"] == "From Map"
def test_custom_gpt_is_not_treated_as_a_project(self):
"""Only g-p- ids are projects; a custom GPT must not become a folder."""
p = self._provider()
out = p.normalize_conversation(self._raw(gizmo_id="g-xyz789"))
assert out["project"] is None
def test_no_gizmo_id_stays_unprojected(self):
p = self._provider()
assert p.normalize_conversation(self._raw())["project"] is None
def test_resolved_project_is_cached_into_the_map(self):
p = self._provider(gizmo_name="Cached")
p.normalize_conversation(self._raw(gizmo_id="g-p-abc123"))
assert p._project_map["conv-1"] == "Cached"
def test_unconfigured_project_is_reported_once(self, caplog):
p = self._provider(project_ids=["g-p-known"], gizmo_name="Surprise")
with caplog.at_level(logging.INFO):
p.normalize_conversation(self._raw(gizmo_id="g-p-surprise"))
p.normalize_conversation(
self._raw(conversation_id="conv-2", gizmo_id="g-p-surprise")
)
hits = [r for r in caplog.records if "not in CHATGPT_PROJECT_IDS" in r.message]
assert len(hits) == 1
assert "Surprise" in hits[0].message
def test_configured_project_is_not_reported(self, caplog):
p = self._provider(project_ids=["g-p-known"], gizmo_name="Known")
with caplog.at_level(logging.INFO):
p.normalize_conversation(self._raw(gizmo_id="g-p-known"))
assert not [r for r in caplog.records if "not in CHATGPT_PROJECT_IDS" in r.message]
Reference in New Issue View Git Blame Copy Permalink