"""Tests for LLM processing pipeline (filter, classifier, deduplicator).""" from datetime import UTC, datetime from unittest.mock import MagicMock, patch from ai_usecases_explorer.collectors.base import RawItem from ai_usecases_explorer.models.usecase import Novelty, ScenarioType, UseCase from ai_usecases_explorer.processors.classifier import Classifier from ai_usecases_explorer.processors.deduplicator import Deduplicator from ai_usecases_explorer.processors.filter import ValueFilter def _raw_item(title: str = "Test", text: str = "Some content") -> RawItem: return RawItem( title=title, source_url="https://example.com/post", source_platform="hackernews", author="user", published_at=datetime.now(tz=UTC), raw_content=f"{title}\n\n{text}", ) def _usecase( id: str = "abc123def456", scenario: ScenarioType = ScenarioType.CODE_GEN_REVIEW, summary: str = "A test summary.", ) -> UseCase: now = datetime.now(tz=UTC) return UseCase( id=id, title="Test", summary=summary, source_url="https://example.com/post", source_platform="hackernews", author="user", published_at=now, collected_at=now, scenario_type=scenario, tools_used=["Claude"], has_local_data=False, has_workflow_opt=True, value_score=0.7, is_real_value=True, novelty=Novelty.NEW, similar_to_ids=[], raw_content="raw", ) # --------------------------------------------------------------------------- # ValueFilter # --------------------------------------------------------------------------- class TestValueFilter: def test_returns_true_for_high_value_item(self) -> None: api_response = { "is_real_value": True, "reason": "Uses local data", "has_local_data": True, "has_workflow_opt": False, } with patch("ai_usecases_explorer.processors.filter.ValueFilter._call_llm", return_value=api_response): vf = ValueFilter(client=MagicMock()) result = vf.run(_raw_item()) assert result["is_real_value"] is True assert result["has_local_data"] is True def test_returns_false_for_low_value_item(self) -> None: api_response = { "is_real_value": False, "reason": "Just asks ChatGPT-style questions", "has_local_data": False, "has_workflow_opt": False, } with patch("ai_usecases_explorer.processors.filter.ValueFilter._call_llm", return_value=api_response): vf = ValueFilter(client=MagicMock()) result = vf.run(_raw_item()) assert result["is_real_value"] is False def test_result_has_required_keys(self) -> None: api_response = { "is_real_value": True, "reason": "ok", "has_local_data": True, "has_workflow_opt": True, } with patch("ai_usecases_explorer.processors.filter.ValueFilter._call_llm", return_value=api_response): vf = ValueFilter(client=MagicMock()) result = vf.run(_raw_item()) assert "is_real_value" in result assert "has_local_data" in result assert "has_workflow_opt" in result assert "reason" in result # --------------------------------------------------------------------------- # Classifier # --------------------------------------------------------------------------- class TestClassifier: def test_returns_classification_result(self) -> None: api_response = { "scenario_type": "代码生成与审查", "tools_used": ["Claude", "VS Code"], "value_score": 0.8, "summary": "User uses Claude to review PRs automatically.", } with patch("ai_usecases_explorer.processors.classifier.Classifier._call_llm", return_value=api_response): clf = Classifier(client=MagicMock()) result = clf.run(_raw_item()) assert result["scenario_type"] == ScenarioType.CODE_GEN_REVIEW assert result["value_score"] == 0.8 assert "Claude" in result["tools_used"] def test_maps_string_to_scenario_type_enum(self) -> None: api_response = { "scenario_type": "工作流自动化", "tools_used": ["n8n"], "value_score": 0.6, "summary": "Automates daily tasks.", } with patch("ai_usecases_explorer.processors.classifier.Classifier._call_llm", return_value=api_response): clf = Classifier(client=MagicMock()) result = clf.run(_raw_item()) assert result["scenario_type"] == ScenarioType.WORKFLOW_AUTOMATION def test_falls_back_to_other_for_unknown_scenario(self) -> None: api_response = { "scenario_type": "未知类别", # not in enum "tools_used": [], "value_score": 0.3, "summary": "Something unknown.", } with patch("ai_usecases_explorer.processors.classifier.Classifier._call_llm", return_value=api_response): clf = Classifier(client=MagicMock()) result = clf.run(_raw_item()) assert result["scenario_type"] == ScenarioType.OTHER def test_result_has_summary(self) -> None: api_response = { "scenario_type": "其他", "tools_used": [], "value_score": 0.4, "summary": "A concise summary.", } with patch("ai_usecases_explorer.processors.classifier.Classifier._call_llm", return_value=api_response): clf = Classifier(client=MagicMock()) result = clf.run(_raw_item()) assert result["summary"] == "A concise summary." # --------------------------------------------------------------------------- # Deduplicator # --------------------------------------------------------------------------- class TestDeduplicator: def test_returns_new_when_no_existing(self) -> None: api_response = { "novelty": "new", "similar_to_ids": [], "novelty_reason": "First of this kind.", } with patch("ai_usecases_explorer.processors.deduplicator.Deduplicator._call_llm", return_value=api_response): dedup = Deduplicator(client=MagicMock()) result = dedup.run(_usecase(), existing=[]) assert result["novelty"] == Novelty.NEW assert result["similar_to_ids"] == [] def test_returns_similar_when_matching_exists(self) -> None: existing = [_usecase("existing001111", summary="Same kind of PR review tool.")] api_response = { "novelty": "similar", "similar_to_ids": ["existing001111"], "novelty_reason": "Same category already seen.", } with patch("ai_usecases_explorer.processors.deduplicator.Deduplicator._call_llm", return_value=api_response): dedup = Deduplicator(client=MagicMock()) result = dedup.run(_usecase(), existing=existing) assert result["novelty"] == Novelty.SIMILAR assert "existing001111" in result["similar_to_ids"] def test_returns_repeat_for_duplicate(self) -> None: existing = [_usecase("dup00000001a", summary="Identical content.")] api_response = { "novelty": "repeat", "similar_to_ids": ["dup00000001a"], "novelty_reason": "Exact duplicate.", } with patch("ai_usecases_explorer.processors.deduplicator.Deduplicator._call_llm", return_value=api_response): dedup = Deduplicator(client=MagicMock()) result = dedup.run(_usecase(), existing=existing) assert result["novelty"] == Novelty.REPEAT def test_skips_llm_when_no_existing(self) -> None: """When existing list is empty, skip LLM call and return NEW directly.""" dedup = Deduplicator(client=MagicMock()) result = dedup.run(_usecase(), existing=[]) assert result["novelty"] == Novelty.NEW assert result["similar_to_ids"] == []