"""Unit-Tests für die Confluence-Quellstrategie. Testet ConfluenceSource mit gemockten HTTP-Responses: - Einzelne Seiten abrufen - Seitenbäume traversieren - Space-Seiten abrufen - Inkrementelle Updates via Versions-Nummer - Fehlerbehandlung (API-Fehler, Parse-Fehler) - Frontmatter-Anreicherung (URL, Space, Titel, Last-Modified) - HTML-zu-Markdown-Konvertierung Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7 """ from __future__ import annotations import json import urllib.error from datetime import date from pathlib import Path from unittest.mock import MagicMock, patch import pytest from monorepo.knowledge.sources.base import SourceConfig from monorepo.knowledge.sources.confluence import ( ConfluenceClient, ConfluencePage, ConfluenceSource, _html_to_markdown, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def confluence_source() -> ConfluenceSource: """Erstellt eine ConfluenceSource-Instanz.""" return ConfluenceSource() @pytest.fixture def source_config() -> SourceConfig: """Erstellt eine gültige SourceConfig für Confluence.""" return SourceConfig( type="confluence", name="Test Confluence", params={ "base_url": "https://confluence.example.com", "token": "test-api-token-123", "spaces": ["ACV2"], }, ) @pytest.fixture def page_config() -> SourceConfig: """SourceConfig mit einzelnen Seiten-IDs.""" return SourceConfig( type="confluence", name="Page Source", params={ "base_url": "https://confluence.example.com", "api_key": "my-api-key", "pages": ["12345", "67890"], }, ) @pytest.fixture def tree_config() -> SourceConfig: """SourceConfig mit Seitenbäumen.""" return SourceConfig( type="confluence", name="Tree Source", params={ "base_url": "https://confluence.example.com", "token": "token-123", "tree_roots": ["99999"], }, ) def _make_page_response( page_id: str = "12345", title: str = "Sprint Planning W24", space_key: str = "ACV2", version: int = 3, last_modified: str = "2025-01-15T10:30:00.000+01:00", content_html: str = "

Inhalt der Seite

", ) -> dict: """Erzeugt eine typische Confluence-API-Antwort für eine Seite.""" return { "id": page_id, "title": title, "space": {"key": space_key}, "version": {"number": version, "when": last_modified}, "body": {"storage": {"value": content_html}}, "_links": {"webui": f"/display/{space_key}/{title.replace(' ', '+')}"}, "children": { "attachment": { "results": [ { "title": "diagram.png", "_links": {"download": "/download/attachments/12345/diagram.png"}, } ] } }, } def _make_child_response(child_ids: list[str]) -> dict: """Erzeugt eine Confluence-API-Antwort für Kind-Seiten.""" return { "results": [{"id": pid} for pid in child_ids], "size": len(child_ids), } def _make_space_response(page_ids: list[str], total: int | None = None) -> dict: """Erzeugt eine Confluence-API-Antwort für Space-Seiten.""" return { "results": [{"id": pid} for pid in page_ids], "size": total if total is not None else len(page_ids), } # --------------------------------------------------------------------------- # Tests: HTML → Markdown Konvertierung # --------------------------------------------------------------------------- class TestHtmlToMarkdown: """Tests für die HTML-zu-Markdown-Konvertierung.""" def test_empty_html(self) -> None: assert _html_to_markdown("") == "" def test_paragraph(self) -> None: result = _html_to_markdown("

Einfacher Text

") assert "Einfacher Text" in result def test_headings(self) -> None: result = _html_to_markdown("

Titel

Untertitel

") assert "# Titel" in result assert "## Untertitel" in result def test_bold_and_italic(self) -> None: result = _html_to_markdown("fett und kursiv") assert "**fett**" in result assert "*kursiv*" in result def test_links(self) -> None: result = _html_to_markdown('Link') assert "[Link](https://example.com)" in result def test_unordered_list(self) -> None: result = _html_to_markdown("") assert "- Eins" in result assert "- Zwei" in result def test_code_inline(self) -> None: result = _html_to_markdown("Nutze git pull bitte") assert "`git pull`" in result def test_line_breaks(self) -> None: result = _html_to_markdown("Zeile 1
Zeile 2") assert "Zeile 1\nZeile 2" in result def test_html_entities_decoded(self) -> None: result = _html_to_markdown("

A & B < C

") assert "A & B < C" in result def test_confluence_macros_removed(self) -> None: html = ( '' "Info text" "" ) result = _html_to_markdown(html) assert "ac:" not in result # --------------------------------------------------------------------------- # Tests: ConfluenceSource Interface # --------------------------------------------------------------------------- class TestConfluenceSourceInterface: """Tests für das SourceStrategy-Interface.""" def test_supports_incremental(self, confluence_source: ConfluenceSource) -> None: assert confluence_source.supports_incremental() is True def test_missing_base_url(self, confluence_source: ConfluenceSource) -> None: """Fehlende base_url erzeugt einen Fehler.""" config = SourceConfig( type="confluence", name="Bad Config", params={"token": "abc"}, ) result = confluence_source.extract(config, "bahn") assert len(result.errors) == 1 assert result.errors[0].error_type == "auth" assert "base_url" in result.errors[0].message def test_missing_token(self, confluence_source: ConfluenceSource) -> None: """Fehlender Token erzeugt einen Fehler.""" config = SourceConfig( type="confluence", name="No Token", params={"base_url": "https://confluence.example.com"}, ) result = confluence_source.extract(config, "bahn") assert len(result.errors) == 1 assert result.errors[0].error_type == "auth" assert "Token" in result.errors[0].message # --------------------------------------------------------------------------- # Tests: Einzelne Seiten abrufen # --------------------------------------------------------------------------- class TestPageExtraction: """Tests für das Abrufen einzelner Seiten.""" @patch.object(ConfluenceClient, "_request") def test_single_page_extraction( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Einzelne Seite wird korrekt extrahiert.""" mock_request.return_value = _make_page_response( page_id="12345", title="Sprint Planning W24", content_html="

Agenda

Punkt 1

", ) result = confluence_source.extract(page_config, "bahn") assert len(result.artifacts) == 2 # Two pages configured artifact = result.artifacts[0] assert artifact.metadata.title == "Sprint Planning W24" assert artifact.metadata.source["type"] == "confluence" assert artifact.metadata.source["space"] == "ACV2" assert "## Agenda" in artifact.content assert "Punkt 1" in artifact.content @patch.object(ConfluenceClient, "_request") def test_frontmatter_enrichment( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Frontmatter wird mit URL, Space, Titel und Last-Modified angereichert.""" mock_request.return_value = _make_page_response( page_id="12345", title="API Redesign", space_key="EINFACHBAHN", last_modified="2025-03-20T14:00:00.000+01:00", ) # Nur eine Seite konfigurieren page_config.params["pages"] = ["12345"] result = confluence_source.extract(page_config, "bahn") artifact = result.artifacts[0] assert artifact.metadata.source["url"].startswith("https://confluence.example.com") assert artifact.metadata.source["space"] == "EINFACHBAHN" assert artifact.metadata.source["page_id"] == "12345" assert artifact.metadata.source["version"] == "3" assert artifact.metadata.created == date(2025, 3, 20) assert artifact.metadata.source_context == "bahn" @patch.object(ConfluenceClient, "_request") def test_content_hash_computed( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Content-Hash wird berechnet.""" mock_request.return_value = _make_page_response(page_id="12345") page_config.params["pages"] = ["12345"] result = confluence_source.extract(page_config, "bahn") artifact = result.artifacts[0] assert artifact.metadata.content_hash.startswith("sha256:") assert len(artifact.metadata.content_hash) == 71 # sha256: + 64 hex chars @patch.object(ConfluenceClient, "_request") def test_attachments_listed( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Anhänge werden als Links im Artefakt vermerkt.""" mock_request.return_value = _make_page_response(page_id="12345") page_config.params["pages"] = ["12345"] result = confluence_source.extract(page_config, "bahn") artifact = result.artifacts[0] assert "## Anhänge" in artifact.content assert "diagram.png" in artifact.content # --------------------------------------------------------------------------- # Tests: Inkrementelle Updates # --------------------------------------------------------------------------- class TestIncrementalUpdates: """Tests für inkrementelle Verarbeitung via Versions-Nummer.""" @patch.object(ConfluenceClient, "_request") def test_skip_unchanged_page( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Unveränderte Seiten werden übersprungen.""" mock_request.return_value = _make_page_response( page_id="12345", version=3 ) page_config.params["pages"] = ["12345"] # Version 3 ist bereits bekannt confluence_source.set_known_versions({"12345": 3}) result = confluence_source.extract(page_config, "bahn") assert len(result.artifacts) == 0 assert result.skipped == 1 @patch.object(ConfluenceClient, "_request") def test_process_updated_page( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Aktualisierte Seiten werden verarbeitet.""" mock_request.return_value = _make_page_response( page_id="12345", version=5 ) page_config.params["pages"] = ["12345"] # Version 3 ist bekannt, aber Seite ist jetzt bei Version 5 confluence_source.set_known_versions({"12345": 3}) result = confluence_source.extract(page_config, "bahn") assert len(result.artifacts) == 1 assert result.skipped == 0 @patch.object(ConfluenceClient, "_request") def test_new_page_always_processed( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Neue Seiten (nicht in known_versions) werden immer verarbeitet.""" mock_request.return_value = _make_page_response( page_id="12345", version=1 ) page_config.params["pages"] = ["12345"] # Andere Seite ist bekannt, aber nicht diese confluence_source.set_known_versions({"99999": 10}) result = confluence_source.extract(page_config, "bahn") assert len(result.artifacts) == 1 # --------------------------------------------------------------------------- # Tests: Seitenbäume # --------------------------------------------------------------------------- class TestPageTreeTraversal: """Tests für die Seitenbaum-Traversierung.""" @patch.object(ConfluenceClient, "_request") def test_tree_traversal( self, mock_request: MagicMock, confluence_source: ConfluenceSource, tree_config: SourceConfig, ) -> None: """Seitenbaum wird rekursiv traversiert.""" # Request-Routing basierend auf URL-Pfad def side_effect(path: str) -> dict: if "/child/page" in path: if "99999" in path: return _make_child_response(["11111", "22222"]) return _make_child_response([]) # get_page calls if "99999" in path: return _make_page_response(page_id="99999", title="Root Page") if "11111" in path: return _make_page_response(page_id="11111", title="Child 1") if "22222" in path: return _make_page_response(page_id="22222", title="Child 2") return _make_page_response() mock_request.side_effect = side_effect result = confluence_source.extract(tree_config, "bahn") # Root + 2 Children = 3 Seiten assert len(result.artifacts) == 3 titles = {a.metadata.title for a in result.artifacts} assert "Root Page" in titles assert "Child 1" in titles assert "Child 2" in titles # --------------------------------------------------------------------------- # Tests: Space-Seiten # --------------------------------------------------------------------------- class TestSpaceExtraction: """Tests für das Abrufen von Space-Seiten.""" @patch.object(ConfluenceClient, "_request") def test_space_pages_extracted( self, mock_request: MagicMock, confluence_source: ConfluenceSource, source_config: SourceConfig, ) -> None: """Alle Seiten eines Spaces werden abgerufen.""" def side_effect(path: str) -> dict: if "spaceKey=ACV2" in path: return _make_space_response(["111", "222", "333"]) # Individual page requests page_id = path.split("/content/")[1].split("?")[0] return _make_page_response(page_id=page_id, title=f"Page {page_id}") mock_request.side_effect = side_effect result = confluence_source.extract(source_config, "bahn") assert len(result.artifacts) == 3 # --------------------------------------------------------------------------- # Tests: Fehlerbehandlung # --------------------------------------------------------------------------- class TestErrorHandling: """Tests für graceful Error Handling.""" @patch.object(ConfluenceClient, "_request") def test_api_error_skips_page( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """API-Fehler bei einzelner Seite überspringt diese und verarbeitet Rest.""" call_count = [0] def side_effect(path: str) -> dict: call_count[0] += 1 if "12345" in path: raise urllib.error.HTTPError( url="https://confluence.example.com/rest/api/content/12345", code=404, msg="Not Found", hdrs={}, # type: ignore fp=None, ) return _make_page_response(page_id="67890", title="Good Page") mock_request.side_effect = side_effect result = confluence_source.extract(page_config, "bahn") # Eine Seite erfolgreich, eine fehlgeschlagen assert len(result.artifacts) == 1 assert result.artifacts[0].metadata.title == "Good Page" assert len(result.errors) == 1 assert result.errors[0].error_type == "connection" assert "12345" in result.errors[0].message @patch.object(ConfluenceClient, "_request") def test_space_api_error_logged( self, mock_request: MagicMock, confluence_source: ConfluenceSource, source_config: SourceConfig, ) -> None: """API-Fehler beim Space-Abruf wird geloggt, Pipeline fährt fort.""" mock_request.side_effect = urllib.error.URLError("Connection refused") result = confluence_source.extract(source_config, "bahn") assert len(result.artifacts) == 0 assert len(result.errors) == 1 assert result.errors[0].error_type == "connection" assert result.errors[0].retry is True @patch.object(ConfluenceClient, "_request") def test_parse_error_skips_page( self, mock_request: MagicMock, confluence_source: ConfluenceSource, page_config: SourceConfig, ) -> None: """Parse-Fehler wird als Fehler geloggt, Seite übersprungen.""" # Ungültiges JSON-ähnliches Response (fehlende Felder) page_config.params["pages"] = ["12345"] mock_request.return_value = {"id": "12345"} # Missing required fields result = confluence_source.extract(page_config, "bahn") # Should still produce an artifact (with defaults) or error # The _parse_page_response handles missing fields gracefully assert len(result.artifacts) == 1 or len(result.errors) == 1 def test_no_pages_configured(self, confluence_source: ConfluenceSource) -> None: """Keine Seiten konfiguriert → leeres Ergebnis ohne Fehler.""" config = SourceConfig( type="confluence", name="Empty Config", params={ "base_url": "https://confluence.example.com", "token": "abc", }, ) result = confluence_source.extract(config, "bahn") assert len(result.artifacts) == 0 assert len(result.errors) == 0 # --------------------------------------------------------------------------- # Tests: Confluence Client # --------------------------------------------------------------------------- class TestConfluenceClient: """Tests für den ConfluenceClient.""" def test_bearer_auth_header(self) -> None: """Bearer-Token wird korrekt gesetzt (Server/DC).""" client = ConfluenceClient( base_url="https://conf.example.com", token="my-token", ) assert client._build_auth_header() == "Bearer my-token" def test_basic_auth_header(self) -> None: """Basic Auth wird korrekt gesetzt (Cloud).""" client = ConfluenceClient( base_url="https://conf.example.com", token="api-token", username="user@example.com", ) header = client._build_auth_header() assert header.startswith("Basic ") def test_base_url_trailing_slash_removed(self) -> None: """Trailing Slash wird von base_url entfernt.""" client = ConfluenceClient( base_url="https://conf.example.com/", token="token", ) assert client.base_url == "https://conf.example.com" # --------------------------------------------------------------------------- # Tests: Slugify # --------------------------------------------------------------------------- class TestSlugify: """Tests für die Slug-Generierung.""" def test_simple_title(self) -> None: assert ConfluenceSource._slugify("Sprint Planning") == "sprint-planning" def test_umlauts(self) -> None: assert ConfluenceSource._slugify("Über Änderungen") == "ueber-aenderungen" def test_special_chars(self) -> None: assert ConfluenceSource._slugify("API (v2) - Redesign!") == "api-v2-redesign" def test_empty_string(self) -> None: assert ConfluenceSource._slugify("") == "untitled" def test_only_special_chars(self) -> None: assert ConfluenceSource._slugify("---") == "untitled"