"""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