"""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)