"""Unit-Tests für die ChatSource-Quellstrategie. Testet Text- und JSON-Chat-Export-Parsing, Teilnehmer-Erkennung und Artefakt-Generierung. Requirements: 7.3, 7.5 """ from __future__ import annotations import json from pathlib import Path import pytest from monorepo.knowledge.sources.base import SourceConfig from monorepo.knowledge.sources.chat import ( ChatSource, _parse_date_str, _slugify, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def chat_dir(tmp_path: Path) -> Path: """Creates a temporary directory for chat export files.""" chat_dir = tmp_path / "chats" chat_dir.mkdir() return chat_dir @pytest.fixture def source_config(chat_dir: Path) -> SourceConfig: """A SourceConfig pointing to the chat directory.""" return SourceConfig( type="chat", name="Test Chats", params={"directory": str(chat_dir)}, ) @pytest.fixture def json_chat_content() -> str: """Sample JSON chat export.""" messages = [ { "sender": "Alice", "timestamp": "2025-01-15 10:00", "text": "Good morning!", }, { "sender": "Bob", "timestamp": "2025-01-15 10:05", "text": "Morning! Ready for the meeting?", }, { "sender": "Alice", "timestamp": "2025-01-15 10:06", "text": "Yes, let's go.", }, ] return json.dumps(messages) @pytest.fixture def text_chat_content() -> str: """Sample text chat export with timestamp pattern.""" return ( "[2025-02-20 14:30] Max: Hallo zusammen\n" "[2025-02-20 14:31] Anna: Hi Max!\n" "[2025-02-20 14:32] Max: Wann ist das nächste Meeting?\n" "[2025-02-20 14:35] Anna: Morgen um 10 Uhr.\n" ) @pytest.fixture def whatsapp_chat_content() -> str: """Sample WhatsApp-style text chat export.""" return ( "2025-03-10, 09:00 - Alice: Good morning team\n" "2025-03-10, 09:01 - Bob: Morning!\n" "2025-03-10, 09:05 - Charlie: Hi all, let's discuss the sprint\n" ) # --------------------------------------------------------------------------- # Tests: Helper functions # --------------------------------------------------------------------------- class TestParseDateStr: def test_iso_datetime(self) -> None: result = _parse_date_str("2025-01-15 10:30:00") assert result is not None assert result.year == 2025 assert result.month == 1 assert result.hour == 10 def test_iso_datetime_no_seconds(self) -> None: result = _parse_date_str("2025-01-15 10:30") assert result is not None assert result.minute == 30 def test_german_format(self) -> None: result = _parse_date_str("15.01.2025, 10:30") assert result is not None assert result.year == 2025 assert result.day == 15 def test_invalid_returns_none(self) -> None: assert _parse_date_str("not a date") is None def test_empty_returns_none(self) -> None: assert _parse_date_str("") is None class TestSlugify: def test_simple_text(self) -> None: assert _slugify("Team Chat") == "team-chat" def test_special_chars(self) -> None: result = _slugify("chat_export (1)") assert "chat" in result assert "export" in result def test_empty(self) -> None: assert _slugify("") == "chat-export" # --------------------------------------------------------------------------- # Tests: ChatSource.extract() with JSON # --------------------------------------------------------------------------- class TestChatSourceJson: def test_extracts_json_chat( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test JSON chat export parsing.""" (chat_dir / "team-chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 1 assert len(result.errors) == 0 artifact = result.artifacts[0] assert artifact.metadata.type == "chat" assert "Chat:" in artifact.metadata.title assert artifact.metadata.source["type"] == "chat" assert artifact.metadata.source["format"] == "json" def test_json_participants_extracted( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test that participants are extracted from JSON chat.""" (chat_dir / "team-chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") artifact = result.artifacts[0] assert "Alice" in artifact.metadata.people assert "Bob" in artifact.metadata.people def test_json_date_range( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test that date range is extracted from JSON chat.""" (chat_dir / "team-chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") artifact = result.artifacts[0] assert artifact.metadata.created is not None assert artifact.metadata.created.year == 2025 assert artifact.metadata.created.month == 1 def test_json_messages_in_content( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test that messages appear in artifact content.""" (chat_dir / "team-chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") artifact = result.artifacts[0] assert "Good morning!" in artifact.content assert "Alice" in artifact.content assert "Bob" in artifact.content def test_empty_json_array_skipped( self, chat_dir: Path, source_config: SourceConfig ) -> None: """Test that empty JSON array produces no artifact.""" (chat_dir / "empty.json").write_text("[]", encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 0 def test_non_array_json_skipped( self, chat_dir: Path, source_config: SourceConfig ) -> None: """Test that non-array JSON is skipped.""" (chat_dir / "object.json").write_text('{"key": "value"}', encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 0 def test_json_alternative_keys( self, chat_dir: Path, source_config: SourceConfig ) -> None: """Test that alternative JSON keys (author, message, date) work.""" messages = [ {"author": "User1", "date": "2025-03-01 09:00", "message": "Hello"}, {"from": "User2", "time": "2025-03-01 09:01", "content": "Hi there"}, ] (chat_dir / "alt.json").write_text(json.dumps(messages), encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 1 artifact = result.artifacts[0] assert "Hello" in artifact.content assert "Hi there" in artifact.content # --------------------------------------------------------------------------- # Tests: ChatSource.extract() with Text # --------------------------------------------------------------------------- class TestChatSourceText: def test_extracts_text_chat_bracket_format( self, chat_dir: Path, source_config: SourceConfig, text_chat_content: str ) -> None: """Test text chat export with [timestamp] format.""" (chat_dir / "team.txt").write_text(text_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 1 artifact = result.artifacts[0] assert artifact.metadata.type == "chat" assert artifact.metadata.source["format"] == "text" def test_text_chat_participants( self, chat_dir: Path, source_config: SourceConfig, text_chat_content: str ) -> None: """Test participant extraction from text chat.""" (chat_dir / "team.txt").write_text(text_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") artifact = result.artifacts[0] assert "Max" in artifact.metadata.people assert "Anna" in artifact.metadata.people def test_whatsapp_format( self, chat_dir: Path, source_config: SourceConfig, whatsapp_chat_content: str ) -> None: """Test WhatsApp-style chat export parsing.""" (chat_dir / "whatsapp.txt").write_text(whatsapp_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 1 artifact = result.artifacts[0] assert "Alice" in artifact.metadata.people assert "Bob" in artifact.metadata.people assert "Charlie" in artifact.metadata.people def test_unrecognized_format_still_creates_artifact( self, chat_dir: Path, source_config: SourceConfig ) -> None: """Test that unrecognized text format still creates an artifact.""" (chat_dir / "freeform.txt").write_text( "Just some notes\nfrom a conversation\nwith no timestamps", encoding="utf-8", ) source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 1 artifact = result.artifacts[0] assert "notes" in artifact.content # --------------------------------------------------------------------------- # Tests: ChatSource general behavior # --------------------------------------------------------------------------- class TestChatSourceGeneral: def test_empty_directory(self, chat_dir: Path, source_config: SourceConfig) -> None: """Test that empty directory returns no artifacts.""" source = ChatSource() 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 for non-existent directory.""" config = SourceConfig( type="chat", name="Missing", params={"directory": str(tmp_path / "nonexistent")}, ) source = ChatSource() 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_supports_incremental_returns_false(self) -> None: """Chat source does not support incremental processing.""" source = ChatSource() assert source.supports_incremental() is False def test_content_hash_is_set( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test that content hash is computed for chat artifacts.""" (chat_dir / "chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") artifact = result.artifacts[0] assert artifact.metadata.content_hash.startswith("sha256:") def test_multiple_files_processed( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str, text_chat_content: str, ) -> None: """Test that multiple chat files in directory are all processed.""" (chat_dir / "chat1.json").write_text(json_chat_content, encoding="utf-8") (chat_dir / "chat2.txt").write_text(text_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert len(result.artifacts) == 2 def test_category_is_inbox( self, chat_dir: Path, source_config: SourceConfig, json_chat_content: str ) -> None: """Test that chat artifacts are categorized as inbox by default.""" (chat_dir / "chat.json").write_text(json_chat_content, encoding="utf-8") source = ChatSource() result = source.extract(source_config, "bahn") assert result.artifacts[0].metadata.category == "inbox"