543 lines
19 KiB
Python
543 lines
19 KiB
Python
"""Tests for EnrichmentAgent – KI-gestützte Anreicherung.
|
||
|
||
Requirements: 9.1, 9.2, 9.3, 9.4, 9.5, 9.7
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from monorepo.knowledge.ingestion.enrichment import (
|
||
CONFIDENCE_THRESHOLD,
|
||
ActionItem,
|
||
EnrichmentAgent,
|
||
EnrichmentResult,
|
||
_compute_action_item_hash,
|
||
_detect_category,
|
||
_extract_action_items,
|
||
_extract_people,
|
||
_extract_projects,
|
||
_extract_title,
|
||
_slugify,
|
||
generate_action_items_section,
|
||
generate_wiki_links,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _slugify tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSlugify:
|
||
"""Tests for _slugify helper."""
|
||
|
||
def test_simple_name(self) -> None:
|
||
assert _slugify("Max Müller") == "max-mueller"
|
||
|
||
def test_umlaut_handling(self) -> None:
|
||
assert _slugify("André Knie") == "andre-knie"
|
||
|
||
def test_sharp_s(self) -> None:
|
||
assert _slugify("Hans Straße") == "hans-strasse"
|
||
|
||
def test_multiple_spaces(self) -> None:
|
||
assert _slugify("Max Müller") == "max-mueller"
|
||
|
||
def test_leading_trailing_spaces(self) -> None:
|
||
assert _slugify(" Max Müller ") == "max-mueller"
|
||
|
||
def test_single_word(self) -> None:
|
||
assert _slugify("Monorepo") == "monorepo"
|
||
|
||
def test_special_characters(self) -> None:
|
||
assert _slugify("Projekt (v2.0)") == "projekt-v2-0"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _compute_action_item_hash tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestComputeActionItemHash:
|
||
"""Tests for _compute_action_item_hash."""
|
||
|
||
def test_deterministic(self) -> None:
|
||
h1 = _compute_action_item_hash("Do the thing")
|
||
h2 = _compute_action_item_hash("Do the thing")
|
||
assert h1 == h2
|
||
|
||
def test_case_insensitive(self) -> None:
|
||
h1 = _compute_action_item_hash("Do the thing")
|
||
h2 = _compute_action_item_hash("do the thing")
|
||
assert h1 == h2
|
||
|
||
def test_trims_whitespace(self) -> None:
|
||
h1 = _compute_action_item_hash("Do the thing")
|
||
h2 = _compute_action_item_hash(" Do the thing ")
|
||
assert h1 == h2
|
||
|
||
def test_different_descriptions_differ(self) -> None:
|
||
h1 = _compute_action_item_hash("Do thing A")
|
||
h2 = _compute_action_item_hash("Do thing B")
|
||
assert h1 != h2
|
||
|
||
def test_returns_16_chars(self) -> None:
|
||
h = _compute_action_item_hash("some task")
|
||
assert len(h) == 16
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _detect_category tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDetectCategory:
|
||
"""Tests for _detect_category."""
|
||
|
||
def test_meeting_keywords(self) -> None:
|
||
content = "Meeting Protokoll vom 15.01.2025\nTeilnehmer: Max, Anna"
|
||
category, confidence = _detect_category(content)
|
||
assert category == "meeting"
|
||
assert confidence >= 0.5
|
||
|
||
def test_decision_keywords(self) -> None:
|
||
content = "Entscheidung: Wir haben beschlossen, die API umzubauen."
|
||
category, _ = _detect_category(content)
|
||
assert category == "decision"
|
||
|
||
def test_project_keywords(self) -> None:
|
||
content = "Projekt Roadmap Q1 2025\nMeilenstein 1: Release Sprint 3"
|
||
category, _ = _detect_category(content)
|
||
assert category == "project"
|
||
|
||
def test_reference_keywords(self) -> None:
|
||
content = "Dokumentation: How-To Guide für den Deployment-Prozess"
|
||
category, _ = _detect_category(content)
|
||
assert category == "reference"
|
||
|
||
def test_link_detection(self) -> None:
|
||
content = "https://docs.python.org/3/library/re.html"
|
||
category, confidence = _detect_category(content)
|
||
assert category == "link"
|
||
assert confidence >= 0.8
|
||
|
||
def test_unknown_content_defaults_inbox(self) -> None:
|
||
content = "Einfach nur ein Gedanke."
|
||
category, confidence = _detect_category(content)
|
||
assert category == "inbox"
|
||
assert confidence < CONFIDENCE_THRESHOLD
|
||
|
||
def test_source_type_hint(self) -> None:
|
||
content = "Some content about things"
|
||
category, _ = _detect_category(content, source_type="jira")
|
||
assert category == "project"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_people tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractPeople:
|
||
"""Tests for _extract_people."""
|
||
|
||
def test_at_mention(self) -> None:
|
||
content = "Assigned to @Max Müller for review."
|
||
people = _extract_people(content)
|
||
names = [n for n, _ in people]
|
||
assert any("Max" in n for n in names)
|
||
|
||
def test_name_pattern(self) -> None:
|
||
content = "Besprochen mit André Knie und Hans Schmidt."
|
||
people = _extract_people(content)
|
||
names = [n for n, _ in people]
|
||
assert any("André Knie" in n for n in names)
|
||
assert any("Hans Schmidt" in n for n in names)
|
||
|
||
def test_excludes_stop_names(self) -> None:
|
||
content = "Meeting Notes und Action Items besprochen."
|
||
people = _extract_people(content)
|
||
names = [n for n, _ in people]
|
||
assert "Action Items" not in names
|
||
assert "Meeting Notes" not in names
|
||
|
||
def test_empty_content(self) -> None:
|
||
people = _extract_people("")
|
||
assert people == []
|
||
|
||
def test_at_mention_high_confidence(self) -> None:
|
||
content = "Feedback von @André eingereicht."
|
||
people = _extract_people(content)
|
||
assert any(conf >= 0.9 for _, conf in people)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_projects tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractProjects:
|
||
"""Tests for _extract_projects."""
|
||
|
||
def test_explicit_project_reference(self) -> None:
|
||
content = "Betrifft Projekt Orchestrator-Monorepo."
|
||
projects = _extract_projects(content)
|
||
names = [n for n, _ in projects]
|
||
assert any("Orchestrator-Monorepo" in n for n in names)
|
||
|
||
def test_english_project_reference(self) -> None:
|
||
content = "This relates to Project Knowledge-Pipeline."
|
||
projects = _extract_projects(content)
|
||
names = [n for n, _ in projects]
|
||
assert any("Knowledge-Pipeline" in n for n in names)
|
||
|
||
def test_existing_wiki_link(self) -> None:
|
||
content = "See [[projects/my-project]] for details."
|
||
projects = _extract_projects(content)
|
||
names = [n for n, _ in projects]
|
||
assert "my-project" in names
|
||
|
||
def test_empty_content(self) -> None:
|
||
projects = _extract_projects("")
|
||
assert projects == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_action_items tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractActionItems:
|
||
"""Tests for _extract_action_items."""
|
||
|
||
def test_checkbox_pattern(self) -> None:
|
||
content = "Tasks:\n- [ ] Finish the report\n- [ ] Send email"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 2
|
||
assert items[0].description == "Finish the report"
|
||
assert items[1].description == "Send email"
|
||
|
||
def test_todo_pattern(self) -> None:
|
||
content = "TODO: Review PR #42\nSome other text"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
assert "Review PR #42" in items[0].description
|
||
|
||
def test_action_pattern(self) -> None:
|
||
content = "ACTION: Schedule follow-up meeting"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
assert "Schedule follow-up meeting" in items[0].description
|
||
|
||
def test_arrow_pattern(self) -> None:
|
||
content = "→ Create Jira ticket for this"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
assert "Create Jira ticket" in items[0].description
|
||
|
||
def test_deduplication(self) -> None:
|
||
content = "- [ ] Do the thing\n- [ ] Do the thing"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
|
||
def test_empty_content(self) -> None:
|
||
items = _extract_action_items("")
|
||
assert items == []
|
||
|
||
def test_assignee_detection(self) -> None:
|
||
content = "- [ ] Review document @André"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
assert items[0].assignee == "André"
|
||
|
||
def test_deadline_detection(self) -> None:
|
||
content = "- [ ] Submit report bis 2025-03-15"
|
||
items = _extract_action_items(content)
|
||
assert len(items) == 1
|
||
assert items[0].deadline == "2025-03-15"
|
||
|
||
def test_hash_auto_computed(self) -> None:
|
||
content = "TODO: Something important"
|
||
items = _extract_action_items(content)
|
||
assert items[0].hash != ""
|
||
assert len(items[0].hash) == 16
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_title tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractTitle:
|
||
"""Tests for _extract_title."""
|
||
|
||
def test_markdown_heading(self) -> None:
|
||
content = "# My Document Title\n\nSome content here."
|
||
assert _extract_title(content) == "My Document Title"
|
||
|
||
def test_h2_heading(self) -> None:
|
||
content = "## Second Level Heading\n\nContent."
|
||
assert _extract_title(content) == "Second Level Heading"
|
||
|
||
def test_first_line_fallback(self) -> None:
|
||
content = "Just some text without a heading\nMore text."
|
||
title = _extract_title(content)
|
||
assert "Just some text" in title
|
||
|
||
def test_empty_content(self) -> None:
|
||
assert _extract_title("") == "Untitled"
|
||
|
||
def test_whitespace_only(self) -> None:
|
||
assert _extract_title(" \n\n ") == "Untitled"
|
||
|
||
def test_long_first_line_truncated(self) -> None:
|
||
content = "A" * 200
|
||
title = _extract_title(content)
|
||
assert len(title) <= 80
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# generate_wiki_links tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateWikiLinks:
|
||
"""Tests for generate_wiki_links."""
|
||
|
||
def test_people_links(self) -> None:
|
||
links = generate_wiki_links(["Max Müller", "André Knie"], [])
|
||
assert links["people"] == [
|
||
"[[people/max-mueller]]",
|
||
"[[people/andre-knie]]",
|
||
]
|
||
|
||
def test_project_links(self) -> None:
|
||
links = generate_wiki_links([], ["Knowledge Pipeline"])
|
||
assert links["projects"] == ["[[projects/knowledge-pipeline]]"]
|
||
|
||
def test_empty_lists(self) -> None:
|
||
links = generate_wiki_links([], [])
|
||
assert links["people"] == []
|
||
assert links["projects"] == []
|
||
|
||
def test_combined(self) -> None:
|
||
links = generate_wiki_links(["Hans Schmidt"], ["Orchestrator"])
|
||
assert "[[people/hans-schmidt]]" in links["people"]
|
||
assert "[[projects/orchestrator]]" in links["projects"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# generate_action_items_section tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGenerateActionItemsSection:
|
||
"""Tests for generate_action_items_section."""
|
||
|
||
def test_empty_list(self) -> None:
|
||
assert generate_action_items_section([]) == ""
|
||
|
||
def test_single_item(self) -> None:
|
||
items = [ActionItem(description="Do the thing")]
|
||
section = generate_action_items_section(items)
|
||
assert "## Action Items" in section
|
||
assert "- [ ] Do the thing" in section
|
||
|
||
def test_item_with_assignee(self) -> None:
|
||
items = [ActionItem(description="Review code", assignee="Max")]
|
||
section = generate_action_items_section(items)
|
||
assert "@Max" in section
|
||
|
||
def test_item_with_deadline(self) -> None:
|
||
items = [ActionItem(description="Submit report", deadline="2025-03-15")]
|
||
section = generate_action_items_section(items)
|
||
assert "bis 2025-03-15" in section
|
||
|
||
def test_multiple_items(self) -> None:
|
||
items = [
|
||
ActionItem(description="Task A"),
|
||
ActionItem(description="Task B"),
|
||
ActionItem(description="Task C"),
|
||
]
|
||
section = generate_action_items_section(items)
|
||
assert section.count("- [ ]") == 3
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# EnrichmentAgent tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEnrichmentAgent:
|
||
"""Tests for EnrichmentAgent class."""
|
||
|
||
@pytest.fixture
|
||
def agent(self) -> EnrichmentAgent:
|
||
return EnrichmentAgent(provider="kiro")
|
||
|
||
def test_default_provider_is_kiro(self) -> None:
|
||
agent = EnrichmentAgent()
|
||
assert agent.provider == "kiro"
|
||
|
||
def test_provider_available_for_kiro(self) -> None:
|
||
agent = EnrichmentAgent(provider="kiro")
|
||
assert agent._provider_available is True
|
||
|
||
def test_provider_unavailable_for_litellm(self) -> None:
|
||
agent = EnrichmentAgent(provider="litellm")
|
||
assert agent._provider_available is False
|
||
|
||
def test_enrich_empty_content(self, agent: EnrichmentAgent) -> None:
|
||
result = agent.enrich("")
|
||
assert result.title == "Untitled"
|
||
assert result.category == "inbox"
|
||
assert result.confidence == 0.0
|
||
|
||
def test_enrich_meeting_content(self, agent: EnrichmentAgent) -> None:
|
||
content = """# Meeting Protokoll 2025-01-15
|
||
|
||
Teilnehmer: André Knie, Max Müller
|
||
|
||
## Agenda
|
||
1. Sprint Review
|
||
2. Nächste Schritte
|
||
|
||
## Action Items
|
||
- [ ] André erstellt Jira-Ticket
|
||
- [ ] Max reviewed den PR bis 2025-01-20
|
||
"""
|
||
result = agent.enrich(content)
|
||
assert result.title == "Meeting Protokoll 2025-01-15"
|
||
assert result.category == "meeting"
|
||
assert result.confidence > 0.5
|
||
assert len(result.action_items) >= 2
|
||
assert result.enrichment_pending is False
|
||
|
||
def test_enrich_detects_people(self, agent: EnrichmentAgent) -> None:
|
||
content = "Besprochen mit André Knie und Hans Schmidt im Meeting."
|
||
result = agent.enrich(content)
|
||
assert len(result.people) >= 1
|
||
|
||
def test_enrich_detects_projects(self, agent: EnrichmentAgent) -> None:
|
||
content = "Betrifft Projekt Orchestrator-Monorepo und die Pipeline."
|
||
result = agent.enrich(content)
|
||
assert len(result.projects) >= 1
|
||
|
||
def test_enrich_confidence_threshold(self, agent: EnrichmentAgent) -> None:
|
||
"""Entities below confidence threshold are excluded."""
|
||
content = "Some text mentioning @Max in a clear mention."
|
||
result = agent.enrich(content)
|
||
# All returned people must have passed confidence threshold
|
||
# (tested indirectly: if people are returned, they passed)
|
||
for person in result.people:
|
||
assert isinstance(person, str)
|
||
assert len(person) >= 2
|
||
|
||
def test_enrich_pending_when_provider_unavailable(self) -> None:
|
||
agent = EnrichmentAgent(provider="litellm")
|
||
result = agent.enrich("Some content to analyze")
|
||
assert result.enrichment_pending is True
|
||
assert result.category == "inbox"
|
||
|
||
def test_enrich_returns_tags(self, agent: EnrichmentAgent) -> None:
|
||
content = "# Python API Documentation\n\nUsing Docker and Kubernetes for deployment."
|
||
result = agent.enrich(content)
|
||
assert "python" in result.tags or "docker" in result.tags or "kubernetes" in result.tags
|
||
|
||
def test_enrich_link_category(self, agent: EnrichmentAgent) -> None:
|
||
content = "https://docs.python.org/3/library/re.html"
|
||
result = agent.enrich(content)
|
||
assert result.category == "link"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# classify_relevance tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestClassifyRelevance:
|
||
"""Tests for classify_relevance."""
|
||
|
||
@pytest.fixture
|
||
def agent(self) -> EnrichmentAgent:
|
||
return EnrichmentAgent(provider="kiro")
|
||
|
||
def test_empty_content_zero(self, agent: EnrichmentAgent) -> None:
|
||
assert agent.classify_relevance("") == 0.0
|
||
|
||
def test_whitespace_only_zero(self, agent: EnrichmentAgent) -> None:
|
||
assert agent.classify_relevance(" ") == 0.0
|
||
|
||
def test_short_content_low_score(self, agent: EnrichmentAgent) -> None:
|
||
score = agent.classify_relevance("Hi")
|
||
assert score <= 0.3
|
||
|
||
def test_structured_content_higher_score(self, agent: EnrichmentAgent) -> None:
|
||
content = """# Important Document
|
||
|
||
This is a longer document with:
|
||
- List items
|
||
- More structure
|
||
|
||
## Section Two
|
||
|
||
Some [linked content](https://example.com) here.
|
||
|
||
TODO: Do something important
|
||
"""
|
||
score = agent.classify_relevance(content)
|
||
assert score >= 0.6
|
||
|
||
def test_score_between_zero_and_one(self, agent: EnrichmentAgent) -> None:
|
||
content = "Medium length content with some words and information."
|
||
score = agent.classify_relevance(content)
|
||
assert 0.0 <= score <= 1.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ActionItem dataclass tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestActionItem:
|
||
"""Tests for ActionItem dataclass."""
|
||
|
||
def test_auto_hash_on_creation(self) -> None:
|
||
item = ActionItem(description="Do something")
|
||
assert item.hash != ""
|
||
assert len(item.hash) == 16
|
||
|
||
def test_preserves_explicit_hash(self) -> None:
|
||
item = ActionItem(description="Do something", hash="custom_hash_val!")
|
||
assert item.hash == "custom_hash_val!"
|
||
|
||
def test_optional_fields_default_none(self) -> None:
|
||
item = ActionItem(description="Task")
|
||
assert item.assignee is None
|
||
assert item.deadline is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# EnrichmentResult dataclass tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEnrichmentResult:
|
||
"""Tests for EnrichmentResult dataclass."""
|
||
|
||
def test_defaults(self) -> None:
|
||
result = EnrichmentResult(title="Test", category="inbox")
|
||
assert result.tags == []
|
||
assert result.people == []
|
||
assert result.projects == []
|
||
assert result.action_items == []
|
||
assert result.confidence == 0.0
|
||
assert result.enrichment_pending is False
|
||
|
||
def test_enrichment_pending_flag(self) -> None:
|
||
result = EnrichmentResult(
|
||
title="Test", category="inbox", enrichment_pending=True
|
||
)
|
||
assert result.enrichment_pending is True
|