(self, tmp_path)
| 1570 | |
| 1571 | @pytest.mark.asyncio |
| 1572 | async def test_long_doc_marks_doc_message(self, tmp_path): |
| 1573 | wiki = tmp_path / "wiki" |
| 1574 | (wiki / "summaries").mkdir(parents=True) |
| 1575 | (wiki / "concepts").mkdir(parents=True) |
| 1576 | (wiki / "index.md").write_text( |
| 1577 | "# Index\n\n## Documents\n\n## Concepts\n", |
| 1578 | encoding="utf-8", |
| 1579 | ) |
| 1580 | sp = wiki / "summaries" / "big.md" |
| 1581 | sp.write_text("PageIndex tree summary.", encoding="utf-8") |
| 1582 | (tmp_path / ".openkb").mkdir() |
| 1583 | |
| 1584 | captured: list[list[dict]] = [] |
| 1585 | plan_response = json.dumps({"create": [], "update": [], "related": []}) |
| 1586 | |
| 1587 | def sync_side_effect(*args, **kwargs): |
| 1588 | captured.append(kwargs["messages"]) |
| 1589 | mock_resp = MagicMock() |
| 1590 | mock_resp.choices = [MagicMock()] |
| 1591 | # First call: overview (plain text); second: plan (JSON). |
| 1592 | mock_resp.choices[0].message.content = ( |
| 1593 | "Overview text" if len(captured) == 1 else plan_response |
| 1594 | ) |
| 1595 | mock_resp.usage = MagicMock(prompt_tokens=1, completion_tokens=1) |
| 1596 | mock_resp.usage.prompt_tokens_details = None |
| 1597 | return mock_resp |
| 1598 | |
| 1599 | with patch("openkb.agent.compiler.litellm") as mock_litellm: |
| 1600 | mock_litellm.completion = MagicMock(side_effect=sync_side_effect) |
| 1601 | mock_litellm.acompletion = AsyncMock() |
| 1602 | await compile_long_doc( |
| 1603 | "big", |
| 1604 | sp, |
| 1605 | "doc-id-1", |
| 1606 | tmp_path, |
| 1607 | "anthropic/claude-sonnet-4-5", |
| 1608 | ) |
| 1609 | |
| 1610 | overview_call = captured[0] |
| 1611 | assert overview_call[1]["role"] == "user" |
| 1612 | assert self._has_cache_breakpoint(overview_call[1]) |
| 1613 | |
| 1614 | |
| 1615 | class TestCompileLongDoc: |
nothing calls this directly
no test coverage detected