307 lines
10 KiB
Python
307 lines
10 KiB
Python
"""Unit-Tests für die EmailSource-Quellstrategie.
|
|
|
|
Testet .eml-Parsing, Attachment-Erkennung und Artefakt-Generierung.
|
|
|
|
Requirements: 7.1, 7.2, 7.5, 7.6
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from monorepo.knowledge.sources.base import SourceConfig
|
|
from monorepo.knowledge.sources.email import (
|
|
EmailSource,
|
|
_extract_body,
|
|
_format_recipients,
|
|
_parse_email_date,
|
|
_slugify,
|
|
_strip_html,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def email_dir(tmp_path: Path) -> Path:
|
|
"""Creates a temporary directory with sample .eml files."""
|
|
email_dir = tmp_path / "emails"
|
|
email_dir.mkdir()
|
|
return email_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_eml_content() -> str:
|
|
"""A minimal valid .eml file content."""
|
|
return (
|
|
"From: sender@example.com\r\n"
|
|
"To: recipient@example.com\r\n"
|
|
"Cc: cc@example.com\r\n"
|
|
"Subject: Test Email Subject\r\n"
|
|
"Date: Mon, 15 Jan 2025 10:30:00 +0100\r\n"
|
|
"Message-ID: <test123@example.com>\r\n"
|
|
"Content-Type: text/plain; charset=utf-8\r\n"
|
|
"\r\n"
|
|
"This is the email body.\r\n"
|
|
"It has multiple lines.\r\n"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def multipart_eml_content() -> str:
|
|
"""A multipart .eml file with text and attachment."""
|
|
return (
|
|
"From: alice@example.com\r\n"
|
|
"To: bob@example.com\r\n"
|
|
"Subject: Meeting Notes\r\n"
|
|
"Date: Wed, 20 Feb 2025 14:00:00 +0000\r\n"
|
|
"Message-ID: <meeting456@example.com>\r\n"
|
|
"MIME-Version: 1.0\r\n"
|
|
'Content-Type: multipart/mixed; boundary="boundary123"\r\n'
|
|
"\r\n"
|
|
"--boundary123\r\n"
|
|
"Content-Type: text/plain; charset=utf-8\r\n"
|
|
"\r\n"
|
|
"Please find the notes attached.\r\n"
|
|
"--boundary123\r\n"
|
|
"Content-Type: application/pdf\r\n"
|
|
'Content-Disposition: attachment; filename="notes.pdf"\r\n'
|
|
"\r\n"
|
|
"FAKE PDF CONTENT\r\n"
|
|
"--boundary123--\r\n"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def source_config(email_dir: Path) -> SourceConfig:
|
|
"""A SourceConfig pointing to the email directory."""
|
|
return SourceConfig(
|
|
type="email",
|
|
name="Test Emails",
|
|
params={"directory": str(email_dir)},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Helper functions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSlugify:
|
|
def test_simple_text(self) -> None:
|
|
assert _slugify("Hello World") == "hello-world"
|
|
|
|
def test_special_chars_removed(self) -> None:
|
|
assert _slugify("Re: Meeting (Important!)") == "re-meeting-important"
|
|
|
|
def test_truncates_at_80_chars(self) -> None:
|
|
long_text = "a" * 100
|
|
assert len(_slugify(long_text)) == 80
|
|
|
|
def test_empty_string(self) -> None:
|
|
assert _slugify("") == "untitled"
|
|
|
|
def test_only_special_chars(self) -> None:
|
|
assert _slugify("!!!???") == "untitled"
|
|
|
|
|
|
class TestParseEmailDate:
|
|
def test_valid_rfc2822_date(self) -> None:
|
|
result = _parse_email_date("Mon, 15 Jan 2025 10:30:00 +0100")
|
|
assert result is not None
|
|
assert result.year == 2025
|
|
assert result.month == 1
|
|
assert result.day == 15
|
|
|
|
def test_none_input(self) -> None:
|
|
assert _parse_email_date(None) is None
|
|
|
|
def test_empty_string(self) -> None:
|
|
assert _parse_email_date("") is None
|
|
|
|
def test_invalid_date(self) -> None:
|
|
assert _parse_email_date("not a date") is None
|
|
|
|
|
|
class TestFormatRecipients:
|
|
def test_single_address(self) -> None:
|
|
result = _format_recipients("user@example.com")
|
|
assert result == ["user@example.com"]
|
|
|
|
def test_named_address(self) -> None:
|
|
result = _format_recipients("John Doe <john@example.com>")
|
|
assert result == ["John Doe <john@example.com>"]
|
|
|
|
def test_multiple_addresses(self) -> None:
|
|
result = _format_recipients("a@x.com, b@x.com")
|
|
assert len(result) == 2
|
|
|
|
def test_none_input(self) -> None:
|
|
assert _format_recipients(None) == []
|
|
|
|
def test_empty_string(self) -> None:
|
|
assert _format_recipients("") == []
|
|
|
|
|
|
class TestStripHtml:
|
|
def test_basic_tags(self) -> None:
|
|
result = _strip_html("<p>Hello <b>world</b></p>")
|
|
assert "Hello" in result
|
|
assert "world" in result
|
|
assert "<" not in result
|
|
|
|
def test_br_tags(self) -> None:
|
|
result = _strip_html("Line1<br>Line2")
|
|
assert "Line1" in result
|
|
assert "Line2" in result
|
|
|
|
def test_script_removal(self) -> None:
|
|
result = _strip_html("<script>alert('x')</script>Content")
|
|
assert "alert" not in result
|
|
assert "Content" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: EmailSource.extract()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEmailSourceExtract:
|
|
def test_extracts_simple_eml(
|
|
self, email_dir: Path, source_config: SourceConfig, sample_eml_content: str
|
|
) -> None:
|
|
"""Test that a simple .eml file is correctly parsed into an artifact."""
|
|
(email_dir / "test.eml").write_bytes(sample_eml_content.encode("utf-8"))
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
assert len(result.artifacts) == 1
|
|
assert len(result.errors) == 0
|
|
|
|
artifact = result.artifacts[0]
|
|
assert artifact.metadata.title == "Test Email Subject"
|
|
assert artifact.metadata.type == "email"
|
|
assert artifact.metadata.source["type"] == "email"
|
|
assert "sender@example.com" in artifact.metadata.people
|
|
assert artifact.metadata.created is not None
|
|
assert artifact.metadata.created.year == 2025
|
|
assert "This is the email body." in artifact.content
|
|
|
|
def test_extracts_multipart_with_attachment(
|
|
self, email_dir: Path, source_config: SourceConfig, multipart_eml_content: str
|
|
) -> None:
|
|
"""Test multipart email with attachment creates artifact with links."""
|
|
(email_dir / "meeting.eml").write_bytes(multipart_eml_content.encode("utf-8"))
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
assert len(result.artifacts) == 1
|
|
artifact = result.artifacts[0]
|
|
assert artifact.metadata.title == "Meeting Notes"
|
|
assert "notes.pdf" in artifact.content
|
|
assert "## Anhänge" in artifact.content
|
|
|
|
def test_empty_directory(self, email_dir: Path, source_config: SourceConfig) -> None:
|
|
"""Test that an empty directory returns no artifacts."""
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
assert len(result.artifacts) == 0
|
|
assert len(result.errors) == 0
|
|
|
|
def test_nonexistent_directory(self, tmp_path: Path) -> None:
|
|
"""Test error handling for non-existent directory."""
|
|
config = SourceConfig(
|
|
type="email",
|
|
name="Missing",
|
|
params={"directory": str(tmp_path / "nonexistent")},
|
|
)
|
|
source = EmailSource()
|
|
result = source.extract(config, "bahn")
|
|
|
|
assert len(result.artifacts) == 0
|
|
assert len(result.errors) == 1
|
|
assert "existiert nicht" in result.errors[0].message
|
|
|
|
def test_non_email_files_ignored(
|
|
self, email_dir: Path, source_config: SourceConfig
|
|
) -> None:
|
|
"""Test that non-email files are ignored."""
|
|
(email_dir / "readme.txt").write_text("Not an email")
|
|
(email_dir / "data.csv").write_text("a,b,c")
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
assert len(result.artifacts) == 0
|
|
|
|
def test_multiple_eml_files(
|
|
self, email_dir: Path, source_config: SourceConfig, sample_eml_content: str
|
|
) -> None:
|
|
"""Test processing multiple .eml files."""
|
|
(email_dir / "first.eml").write_bytes(sample_eml_content.encode("utf-8"))
|
|
(email_dir / "second.eml").write_bytes(
|
|
sample_eml_content.replace("Test Email Subject", "Second Email").encode("utf-8")
|
|
)
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
assert len(result.artifacts) == 2
|
|
|
|
def test_frontmatter_contains_required_fields(
|
|
self, email_dir: Path, source_config: SourceConfig, sample_eml_content: str
|
|
) -> None:
|
|
"""Test that artifact frontmatter has sender, recipients, date, source."""
|
|
(email_dir / "test.eml").write_bytes(sample_eml_content.encode("utf-8"))
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
artifact = result.artifacts[0]
|
|
# Check source dict
|
|
assert artifact.metadata.source["type"] == "email"
|
|
assert "message_id" in artifact.metadata.source
|
|
# Check people includes sender and recipients
|
|
assert "sender@example.com" in artifact.metadata.people
|
|
assert "recipient@example.com" in artifact.metadata.people
|
|
# Check date
|
|
assert artifact.metadata.created is not None
|
|
|
|
def test_supports_incremental_returns_false(self) -> None:
|
|
"""Email source does not support incremental processing."""
|
|
source = EmailSource()
|
|
assert source.supports_incremental() is False
|
|
|
|
def test_content_hash_is_set(
|
|
self, email_dir: Path, source_config: SourceConfig, sample_eml_content: str
|
|
) -> None:
|
|
"""Test that content hash is computed."""
|
|
(email_dir / "test.eml").write_bytes(sample_eml_content.encode("utf-8"))
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
artifact = result.artifacts[0]
|
|
assert artifact.metadata.content_hash.startswith("sha256:")
|
|
|
|
def test_malformed_eml_produces_error(
|
|
self, email_dir: Path, source_config: SourceConfig
|
|
) -> None:
|
|
"""Test that a completely invalid file doesn't crash but may produce minimal artifact."""
|
|
# An empty file is still parseable by Python's email module (produces empty Message)
|
|
(email_dir / "broken.eml").write_bytes(b"")
|
|
|
|
source = EmailSource()
|
|
result = source.extract(source_config, "bahn")
|
|
|
|
# Should still produce an artifact (with default values) or an error
|
|
assert len(result.artifacts) + len(result.errors) >= 0 # No crash
|