610 lines
21 KiB
Python
610 lines
21 KiB
Python
"""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 = "<p>Inhalt der Seite</p>",
|
|
) -> 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("<p>Einfacher Text</p>")
|
|
assert "Einfacher Text" in result
|
|
|
|
def test_headings(self) -> None:
|
|
result = _html_to_markdown("<h1>Titel</h1><h2>Untertitel</h2>")
|
|
assert "# Titel" in result
|
|
assert "## Untertitel" in result
|
|
|
|
def test_bold_and_italic(self) -> None:
|
|
result = _html_to_markdown("<strong>fett</strong> und <em>kursiv</em>")
|
|
assert "**fett**" in result
|
|
assert "*kursiv*" in result
|
|
|
|
def test_links(self) -> None:
|
|
result = _html_to_markdown('<a href="https://example.com">Link</a>')
|
|
assert "[Link](https://example.com)" in result
|
|
|
|
def test_unordered_list(self) -> None:
|
|
result = _html_to_markdown("<ul><li>Eins</li><li>Zwei</li></ul>")
|
|
assert "- Eins" in result
|
|
assert "- Zwei" in result
|
|
|
|
def test_code_inline(self) -> None:
|
|
result = _html_to_markdown("Nutze <code>git pull</code> bitte")
|
|
assert "`git pull`" in result
|
|
|
|
def test_line_breaks(self) -> None:
|
|
result = _html_to_markdown("Zeile 1<br/>Zeile 2")
|
|
assert "Zeile 1\nZeile 2" in result
|
|
|
|
def test_html_entities_decoded(self) -> None:
|
|
result = _html_to_markdown("<p>A & B < C</p>")
|
|
assert "A & B < C" in result
|
|
|
|
def test_confluence_macros_removed(self) -> None:
|
|
html = (
|
|
'<ac:structured-macro ac:name="info">'
|
|
"<ac:rich-text-body>Info text</ac:rich-text-body>"
|
|
"</ac:structured-macro>"
|
|
)
|
|
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="<h2>Agenda</h2><p>Punkt 1</p>",
|
|
)
|
|
|
|
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"
|