499 lines
17 KiB
Python
499 lines
17 KiB
Python
"""Unit-Tests für die PDF- und OCR-Quellstrategie.
|
||
|
||
Testet PDFSource mit verschiedenen Szenarien:
|
||
- PDF mit eingebettetem Text
|
||
- Bilddateien (direkt OCR)
|
||
- OCR-Fallback bei gescannten PDFs
|
||
- Fehlerbehandlung (fehlende Bibliothek, ungültige Dateien)
|
||
- source.file-Referenz in Artefakt-Metadaten
|
||
|
||
Requirements: 4.2, 4.3, 4.4, 4.5, 4.6
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
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.pdf import (
|
||
IMAGE_EXTENSIONS,
|
||
OCR_LANGUAGES,
|
||
PDF_EXTENSIONS,
|
||
PDFSource,
|
||
SUPPORTED_EXTENSIONS,
|
||
_extract_pdf_text,
|
||
_perform_ocr,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def pdf_source() -> PDFSource:
|
||
"""Erstellt eine PDFSource-Instanz."""
|
||
return PDFSource()
|
||
|
||
|
||
@pytest.fixture
|
||
def source_config(tmp_path: Path) -> SourceConfig:
|
||
"""Erstellt eine SourceConfig mit temporärem Verzeichnis."""
|
||
return SourceConfig(
|
||
type="file",
|
||
name="Test PDF Source",
|
||
params={"path": str(tmp_path)},
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_pdf(tmp_path: Path) -> Path:
|
||
"""Erstellt eine Dummy-PDF-Datei (Binärdaten)."""
|
||
pdf_file = tmp_path / "test-document.pdf"
|
||
# Minimales PDF-Platzhalter (wird für Mock-Tests gebraucht)
|
||
pdf_file.write_bytes(b"%PDF-1.4 dummy content")
|
||
return pdf_file
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_image(tmp_path: Path) -> Path:
|
||
"""Erstellt eine Dummy-Bilddatei."""
|
||
img_file = tmp_path / "test-image.png"
|
||
img_file.write_bytes(b"\x89PNG\r\n\x1a\n dummy image")
|
||
return img_file
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_jpg(tmp_path: Path) -> Path:
|
||
"""Erstellt eine Dummy-JPG-Datei."""
|
||
img_file = tmp_path / "whiteboard-photo.jpg"
|
||
img_file.write_bytes(b"\xff\xd8\xff\xe0 dummy jpg")
|
||
return img_file
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Unterstützte Formate
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSupportedExtensions:
|
||
"""Tests für Dateiformat-Erkennung."""
|
||
|
||
def test_pdf_extension_supported(self) -> None:
|
||
assert ".pdf" in SUPPORTED_EXTENSIONS
|
||
|
||
def test_png_extension_supported(self) -> None:
|
||
assert ".png" in SUPPORTED_EXTENSIONS
|
||
|
||
def test_jpg_extension_supported(self) -> None:
|
||
assert ".jpg" in SUPPORTED_EXTENSIONS
|
||
|
||
def test_jpeg_extension_supported(self) -> None:
|
||
assert ".jpeg" in SUPPORTED_EXTENSIONS
|
||
|
||
def test_pdf_and_image_sets_disjoint(self) -> None:
|
||
assert PDF_EXTENSIONS.isdisjoint(IMAGE_EXTENSIONS)
|
||
|
||
def test_ocr_languages_configured(self) -> None:
|
||
assert OCR_LANGUAGES == "deu+eng"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: PDFSource Interface
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPDFSourceInterface:
|
||
"""Tests für das SourceStrategy-Interface."""
|
||
|
||
def test_supports_incremental(self, pdf_source: PDFSource) -> None:
|
||
assert pdf_source.supports_incremental() is True
|
||
|
||
def test_extract_empty_directory(
|
||
self, pdf_source: PDFSource, source_config: SourceConfig
|
||
) -> None:
|
||
"""Leeres Verzeichnis liefert keine Artefakte."""
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
assert result.artifacts == []
|
||
assert result.errors == []
|
||
|
||
def test_extract_missing_path_param(self, pdf_source: PDFSource) -> None:
|
||
"""Fehlender path-Parameter erzeugt einen Fehler."""
|
||
config = SourceConfig(type="file", name="Bad Config", params={})
|
||
result = pdf_source.extract(config, "bahn")
|
||
assert len(result.errors) == 1
|
||
assert result.errors[0].error_type == "parse"
|
||
|
||
def test_extract_nonexistent_path(self, pdf_source: PDFSource) -> None:
|
||
"""Nicht-existenter Pfad erzeugt einen Fehler."""
|
||
config = SourceConfig(
|
||
type="file",
|
||
name="Missing Path",
|
||
params={"path": "/nonexistent/path/to/files"},
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
assert len(result.errors) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: PDF-Verarbeitung
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPDFProcessing:
|
||
"""Tests für die PDF-Textextraktion."""
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_pdf_with_embedded_text(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_pdf: Path,
|
||
) -> None:
|
||
"""PDF mit eingebettetem Text wird korrekt extrahiert."""
|
||
mock_extract.return_value = ("Dies ist der PDF-Text.", False)
|
||
|
||
source_config.params["path"] = str(sample_pdf)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert "Dies ist der PDF-Text." in artifact.content
|
||
assert artifact.metadata.ocr_failed is False
|
||
assert artifact.metadata.source["file"] == str(sample_pdf)
|
||
assert artifact.metadata.source["type"] == "pdf"
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_pdf_scanned_with_ocr_success(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_pdf: Path,
|
||
) -> None:
|
||
"""Gescanntes PDF: OCR-Fallback erfolgreich."""
|
||
mock_extract.return_value = ("", True) # Kein Text, OCR benötigt
|
||
mock_ocr.return_value = ("OCR-extrahierter Text", False)
|
||
|
||
source_config.params["path"] = str(sample_pdf)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert "OCR-extrahierter Text" in artifact.content
|
||
assert artifact.metadata.ocr_failed is False
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_pdf_scanned_with_ocr_failure(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_pdf: Path,
|
||
) -> None:
|
||
"""Gescanntes PDF: OCR fehlgeschlagen → ocr_failed=True."""
|
||
mock_extract.return_value = ("", True) # Kein Text, OCR benötigt
|
||
mock_ocr.return_value = ("", True) # OCR fehlgeschlagen
|
||
|
||
source_config.params["path"] = str(sample_pdf)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert artifact.metadata.ocr_failed is True
|
||
assert artifact.content == "# test document\n\n" # Leerer Inhalt
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_pdf_library_not_available(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_pdf: Path,
|
||
) -> None:
|
||
"""PDF-Bibliothek fehlt → versucht OCR als Fallback."""
|
||
mock_extract.side_effect = ImportError("No PDF lib")
|
||
|
||
source_config.params["path"] = str(sample_pdf)
|
||
|
||
with patch("monorepo.knowledge.sources.pdf._perform_ocr") as mock_ocr:
|
||
mock_ocr.return_value = ("", True)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
assert len(result.artifacts) == 1
|
||
assert result.artifacts[0].metadata.ocr_failed is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Bilddateien → OCR
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestImageProcessing:
|
||
"""Tests für die Verarbeitung von Bilddateien."""
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
def test_image_ocr_success(
|
||
self,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_image: Path,
|
||
) -> None:
|
||
"""Bilddatei wird via OCR verarbeitet."""
|
||
mock_ocr.return_value = ("Handschriftlicher Text", False)
|
||
|
||
source_config.params["path"] = str(sample_image)
|
||
result = pdf_source.extract(source_config, "privat")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert "Handschriftlicher Text" in artifact.content
|
||
assert artifact.metadata.ocr_failed is False
|
||
assert artifact.metadata.source["type"] == "image"
|
||
assert artifact.metadata.source["file"] == str(sample_image)
|
||
assert artifact.metadata.source_context == "privat"
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
def test_image_ocr_failure(
|
||
self,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_image: Path,
|
||
) -> None:
|
||
"""Bilddatei: OCR fehlgeschlagen → ocr_failed=True."""
|
||
mock_ocr.return_value = ("", True)
|
||
|
||
source_config.params["path"] = str(sample_image)
|
||
result = pdf_source.extract(source_config, "privat")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert artifact.metadata.ocr_failed is True
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
def test_jpg_processed_as_image(
|
||
self,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_jpg: Path,
|
||
) -> None:
|
||
"""JPG-Datei wird korrekt als Bild verarbeitet."""
|
||
mock_ocr.return_value = ("Whiteboard Notiz", False)
|
||
|
||
source_config.params["path"] = str(sample_jpg)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
assert len(result.artifacts) == 1
|
||
artifact = result.artifacts[0]
|
||
assert artifact.metadata.source["type"] == "image"
|
||
assert "whiteboard" in artifact.metadata.title.lower()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Source-Referenz (source.file)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSourceFileReference:
|
||
"""Tests für die Beibehaltung der source.file-Referenz.
|
||
|
||
Requirements: 4.5 – Referenz auf Originaldatei im Frontmatter.
|
||
"""
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
def test_source_file_preserved_for_image(
|
||
self,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_image: Path,
|
||
) -> None:
|
||
"""source.file enthält den vollständigen Pfad zur Originaldatei."""
|
||
mock_ocr.return_value = ("Some text", False)
|
||
|
||
source_config.params["path"] = str(sample_image)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
artifact = result.artifacts[0]
|
||
assert "file" in artifact.metadata.source
|
||
assert artifact.metadata.source["file"] == str(sample_image)
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_source_file_preserved_for_pdf(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
source_config: SourceConfig,
|
||
sample_pdf: Path,
|
||
) -> None:
|
||
"""source.file enthält den vollständigen Pfad zur Original-PDF."""
|
||
mock_extract.return_value = ("PDF text", False)
|
||
|
||
source_config.params["path"] = str(sample_pdf)
|
||
result = pdf_source.extract(source_config, "bahn")
|
||
|
||
artifact = result.artifacts[0]
|
||
assert "file" in artifact.metadata.source
|
||
assert artifact.metadata.source["file"] == str(sample_pdf)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Verzeichnis-Scan
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDirectoryScan:
|
||
"""Tests für die Dateisammlung aus Verzeichnissen."""
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_mixed_directory(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Verzeichnis mit gemischten Dateitypen."""
|
||
# Erstelle Testdateien
|
||
(tmp_path / "doc.pdf").write_bytes(b"%PDF-1.4 content")
|
||
(tmp_path / "photo.png").write_bytes(b"\x89PNG data")
|
||
(tmp_path / "notes.txt").write_text("Text file") # Nicht unterstützt
|
||
(tmp_path / "readme.md").write_text("# Readme") # Nicht unterstützt
|
||
|
||
mock_extract.return_value = ("PDF text", False)
|
||
mock_ocr.return_value = ("Image text", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Mixed", params={"path": str(tmp_path)}
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
|
||
# Nur PDF und PNG werden verarbeitet
|
||
assert len(result.artifacts) == 2
|
||
types = {a.metadata.source["type"] for a in result.artifacts}
|
||
assert "pdf" in types
|
||
assert "image" in types
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._perform_ocr")
|
||
def test_subdirectory_scan(
|
||
self,
|
||
mock_ocr: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Unterverzeichnisse werden rekursiv durchsucht."""
|
||
subdir = tmp_path / "subfolder"
|
||
subdir.mkdir()
|
||
(subdir / "nested.jpg").write_bytes(b"\xff\xd8 data")
|
||
|
||
mock_ocr.return_value = ("Nested text", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Nested", params={"directory": str(tmp_path)}
|
||
)
|
||
result = pdf_source.extract(config, "dhive")
|
||
|
||
assert len(result.artifacts) == 1
|
||
assert result.artifacts[0].metadata.source_context == "dhive"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: Metadaten
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestArtifactMetadata:
|
||
"""Tests für die generierten Artefakt-Metadaten."""
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_title_from_filename(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Titel wird aus dem Dateinamen abgeleitet."""
|
||
pdf_file = tmp_path / "sprint-planning-w24.pdf"
|
||
pdf_file.write_bytes(b"%PDF content")
|
||
mock_extract.return_value = ("Planning text", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Test", params={"path": str(pdf_file)}
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
|
||
assert result.artifacts[0].metadata.title == "sprint planning w24"
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_content_hash_computed(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Content-Hash wird berechnet."""
|
||
pdf_file = tmp_path / "doc.pdf"
|
||
pdf_file.write_bytes(b"%PDF content")
|
||
mock_extract.return_value = ("Some content", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Test", params={"path": str(pdf_file)}
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
|
||
artifact = result.artifacts[0]
|
||
assert artifact.metadata.content_hash.startswith("sha256:")
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_created_date_set(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Erstelldatum wird auf heute gesetzt."""
|
||
pdf_file = tmp_path / "doc.pdf"
|
||
pdf_file.write_bytes(b"%PDF content")
|
||
mock_extract.return_value = ("Text", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Test", params={"path": str(pdf_file)}
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
|
||
assert result.artifacts[0].metadata.created == date.today()
|
||
|
||
@patch("monorepo.knowledge.sources.pdf._extract_pdf_text")
|
||
def test_category_set_to_references(
|
||
self,
|
||
mock_extract: MagicMock,
|
||
pdf_source: PDFSource,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""Kategorie ist standardmäßig 'references'."""
|
||
pdf_file = tmp_path / "doc.pdf"
|
||
pdf_file.write_bytes(b"%PDF content")
|
||
mock_extract.return_value = ("Text", False)
|
||
|
||
config = SourceConfig(
|
||
type="file", name="Test", params={"path": str(pdf_file)}
|
||
)
|
||
result = pdf_source.extract(config, "bahn")
|
||
|
||
assert result.artifacts[0].metadata.category == "references"
|