"""Unit tests for the Confluence REST utilities and storage-format conversion.""" from __future__ import annotations from typing import Any import pytest from agent.utils import confluence # --- storage-format conversion --------------------------------------------- def test_text_to_storage_wraps_blocks_in_paragraphs() -> None: storage = confluence.text_to_storage("first block\n\nsecond block") assert storage == "

first block

second block

" def test_text_to_storage_escapes_html() -> None: storage = confluence.text_to_storage("a < b & c > d") assert storage == "

a < b & c > d

" def test_text_to_storage_empty_is_empty() -> None: assert confluence.text_to_storage("") == "" def test_storage_to_text_strips_tags() -> None: assert confluence.storage_to_text("

hello world

") == "hello world" def test_storage_to_text_handles_none_and_empty() -> None: assert confluence.storage_to_text("") == "" def test_storage_to_text_unescapes_entities() -> None: assert confluence.storage_to_text("

a < b & c

") == "a < b & c" # --- Confluence REST utilities (mocked transport) --------------------------- @pytest.fixture def _confluence_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(confluence, "CONFLUENCE_BASE_URL", "https://seahaven.atlassian.net") monkeypatch.setattr(confluence, "CONFLUENCE_EMAIL", "bot@seahavenind.com") monkeypatch.setattr(confluence, "CONFLUENCE_API_TOKEN", "token") def _mock_request( monkeypatch: pytest.MonkeyPatch, responses: dict[str, Any] | list[dict[str, Any]] ) -> list[dict[str, Any]]: calls: list[dict[str, Any]] = [] queue = responses if isinstance(responses, list) else None async def fake_request(method: str, path: str, *, json=None, params=None): calls.append({"method": method, "path": path, "json": json, "params": params}) if queue is not None: return queue[len(calls) - 1] return responses monkeypatch.setattr(confluence, "_request", fake_request) return calls async def test_get_page_normalizes_fields( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: raw = { "id": "123", "title": "Architecture Map", "body": {"storage": {"value": "

System overview

"}}, "version": {"number": 3}, "space": {"key": "IT"}, "_links": {"webui": "/spaces/IT/pages/123/Architecture+Map"}, } _mock_request(monkeypatch, raw) result = await confluence.get_page("123") page = result["page"] assert page["id"] == "123" assert page["title"] == "Architecture Map" assert page["body"] == "System overview" assert page["version"] == 3 assert page["space_key"] == "IT" assert page["url"] == ( "https://seahaven.atlassian.net/wiki/spaces/IT/pages/123/Architecture+Map" ) async def test_create_page_builds_payload( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: calls = _mock_request( monkeypatch, {"id": "456", "title": "New Page", "_links": {"webui": "/spaces/IT/pages/456/New+Page"}}, ) result = await confluence.create_page("IT", "New Page", "hello world", parent_id="100") assert result["success"] is True assert result["page"]["id"] == "456" sent = calls[0]["json"] assert sent["type"] == "page" assert sent["space"] == {"key": "IT"} assert sent["title"] == "New Page" assert sent["body"]["storage"]["value"] == "

hello world

" assert sent["body"]["storage"]["representation"] == "storage" assert sent["ancestors"] == [{"id": "100"}] async def test_update_page_reads_current_then_increments_version( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: get_response = { "id": "123", "title": "Old Title", "body": {"storage": {"value": "

old

"}}, "version": {"number": 5}, "space": {"key": "IT"}, "_links": {"webui": "/spaces/IT/pages/123/Old+Title"}, } put_response = { "id": "123", "title": "Old Title", "_links": {"webui": "/spaces/IT/pages/123/Old+Title"}, } calls = _mock_request(monkeypatch, [get_response, put_response]) result = await confluence.update_page("123", body="new body") assert result["success"] is True assert calls[0]["method"] == "GET" assert calls[1]["method"] == "PUT" assert calls[1]["path"] == "/content/123" sent = calls[1]["json"] assert sent["version"]["number"] == 6 assert sent["title"] == "Old Title" assert sent["body"]["storage"]["value"] == "

new body

" async def test_add_comment_builds_container( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: calls = _mock_request(monkeypatch, {"id": "999"}) result = await confluence.add_comment("123", "great work") assert result["success"] is True sent = calls[0]["json"] assert sent["type"] == "comment" assert sent["container"] == {"id": "123", "type": "page"} assert sent["body"]["storage"]["value"] == "

great work

" async def test_search_normalizes_results( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: _mock_request( monkeypatch, { "results": [ { "id": "123", "title": "Architecture Map", "type": "page", "_links": {"webui": "/spaces/IT/pages/123/Architecture+Map"}, } ] }, ) result = await confluence.search('space = "IT"') assert result["results"][0]["id"] == "123" assert result["results"][0]["title"] == "Architecture Map" assert result["results"][0]["type"] == "page" async def test_request_without_env_returns_error(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(confluence, "CONFLUENCE_BASE_URL", "") monkeypatch.setattr(confluence, "CONFLUENCE_API_TOKEN", "") result = await confluence._request("GET", "/content/123") assert "error" in result async def test_get_page_propagates_error(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(confluence, "CONFLUENCE_BASE_URL", "") monkeypatch.setattr(confluence, "CONFLUENCE_API_TOKEN", "") result = await confluence.get_page("123") assert "error" in result async def test_get_page_encodes_page_id( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: calls = _mock_request(monkeypatch, {"id": "x", "title": "t"}) await confluence.get_page("../../admin/foo") assert calls[0]["path"] == "/content/..%2F..%2Fadmin%2Ffoo" async def test_update_page_encodes_page_id( _confluence_env: None, monkeypatch: pytest.MonkeyPatch ) -> None: get_resp = { "id": "1", "title": "T", "body": {"storage": {"value": ""}}, "version": {"number": 1}, "space": {"key": "IT"}, "_links": {"webui": "/x"}, } calls = _mock_request( monkeypatch, [get_resp, {"id": "1", "title": "T", "_links": {"webui": "/x"}}] ) await confluence.update_page("1?status=trashed", body="new") # Both the internal get_page and the PUT must encode the id (no raw query). assert calls[0]["path"] == "/content/1%3Fstatus%3Dtrashed" assert calls[1]["method"] == "PUT" assert calls[1]["path"] == "/content/1%3Fstatus%3Dtrashed"