Artur Mukhamadiev
e65fbf4b67
FastAPI app, upstream Confluence/model adapters, authoritative history, rootless container lifecycle, artifact storage and downloads, fake peers under backend/dev, tests under tests/backend. Root pytest.ini deselects the live marker by default; requirements gain the backend dependencies.
141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
"""Unit tests for HistoryManager and WarningsManager."""
|
|
|
|
import pytest
|
|
from backend.history import HistoryManager, WarningsManager, rfc3339_utc
|
|
|
|
|
|
def test_warnings_aggregation_and_limit():
|
|
mgr = WarningsManager(max_warnings=3)
|
|
mgr.add_warning("code1", "msg1")
|
|
mgr.add_warning("code1", "msg1") # duplicate -> ignored
|
|
mgr.add_warning("code2", "msg2")
|
|
mgr.add_warning("code3", "msg3")
|
|
mgr.add_warning("code4", "msg4") # exceeds limit of 3
|
|
|
|
warnings = mgr.get_warnings()
|
|
assert len(warnings) == 3
|
|
assert warnings[0]["code"] == "code1"
|
|
assert warnings[1]["code"] == "code2"
|
|
assert warnings[2]["code"] == "code3"
|
|
|
|
|
|
def test_history_record_and_pages_accessed():
|
|
warnings = WarningsManager()
|
|
history = HistoryManager(warnings)
|
|
|
|
t1 = rfc3339_utc()
|
|
t2 = rfc3339_utc()
|
|
|
|
# 1. Search call -> should NOT appear in pages_accessed
|
|
history.record_call(
|
|
tool_call_id="call_1",
|
|
tool="confluence_search",
|
|
parameters={"query": "deploy"},
|
|
started_at=t1,
|
|
completed_at=t2,
|
|
status="success",
|
|
cache_hit=False,
|
|
result={"pages": [{"page_id": "847291", "title": "Deploy Guide"}]},
|
|
error=None,
|
|
)
|
|
|
|
# 2. View call success -> should appear in pages_accessed
|
|
history.record_call(
|
|
tool_call_id="call_2",
|
|
tool="confluence_view",
|
|
parameters={"page_id": "847291"},
|
|
started_at=t1,
|
|
completed_at=t2,
|
|
status="success",
|
|
cache_hit=False,
|
|
result={
|
|
"page_id": "847291",
|
|
"title": "Deploy Guide",
|
|
"space": "OPS",
|
|
"url": "https://approved.example.com/pages/viewpage.action?pageId=847291",
|
|
"markdown": "# Guide",
|
|
"truncated": False,
|
|
},
|
|
error=None,
|
|
)
|
|
|
|
# 3. Repeated view call (cache hit) -> should NOT duplicate page in pages_accessed
|
|
history.record_call(
|
|
tool_call_id="call_3",
|
|
tool="confluence_view",
|
|
parameters={"page_id": "847291"},
|
|
started_at=t2,
|
|
completed_at=t2,
|
|
status="success",
|
|
cache_hit=True,
|
|
result={
|
|
"page_id": "847291",
|
|
"title": "Deploy Guide",
|
|
"space": "OPS",
|
|
"url": "https://approved.example.com/pages/viewpage.action?pageId=847291",
|
|
"markdown": "# Guide",
|
|
"truncated": False,
|
|
},
|
|
error=None,
|
|
)
|
|
|
|
# 4. View call with error -> should NOT appear in pages_accessed
|
|
history.record_call(
|
|
tool_call_id="call_4",
|
|
tool="confluence_view",
|
|
parameters={"page_id": "999999"},
|
|
started_at=t2,
|
|
completed_at=t2,
|
|
status="error",
|
|
cache_hit=False,
|
|
result=None,
|
|
error={"code": "upstream_failed", "message": "Not found"},
|
|
)
|
|
|
|
# Verify history list
|
|
entries = history.get_tool_history()
|
|
assert len(entries) == 4
|
|
assert entries[0]["tool_call_id"] == "call_1"
|
|
assert entries[3]["error"] == {"code": "upstream_failed", "message": "Not found"}
|
|
assert entries[3]["result"] is None
|
|
|
|
# Verify pages_accessed
|
|
pages = history.get_pages_accessed()
|
|
assert len(pages) == 1
|
|
assert pages[0]["page_id"] == "847291"
|
|
assert pages[0]["title"] == "Deploy Guide"
|
|
assert pages[0]["space"] == "OPS"
|
|
assert pages[0]["url"] == "https://approved.example.com/pages/viewpage.action?pageId=847291"
|
|
assert pages[0]["accessed_at"] == t2
|
|
|
|
|
|
def test_malformed_url_in_page_accessed():
|
|
warnings = WarningsManager()
|
|
history = HistoryManager(warnings)
|
|
|
|
t = rfc3339_utc()
|
|
history.record_call(
|
|
tool_call_id="call_bad_url",
|
|
tool="confluence_view",
|
|
parameters={"page_id": "123"},
|
|
started_at=t,
|
|
completed_at=t,
|
|
status="success",
|
|
cache_hit=False,
|
|
result={
|
|
"page_id": "123",
|
|
"title": "Bad URL Page",
|
|
"space": "TEST",
|
|
"url": "javascript:alert(1)", # malformed / unsafe URL
|
|
"markdown": "content",
|
|
"truncated": False,
|
|
},
|
|
error=None,
|
|
)
|
|
|
|
pages = history.get_pages_accessed()
|
|
assert len(pages) == 1
|
|
assert pages[0]["url"] == "" # replaced with empty string
|
|
warns = warnings.get_warnings()
|
|
assert any(w["code"] == "unusable_page_url" for w in warns)
|