from __future__ import annotations import unittest import json from pathlib import Path from unittest.mock import patch from app.automation.context import WorkflowContext from app.automation.nodes.web_search import WebSearchRunner, normalize_search_result, result_identity from app.automation.nodes.research import AiWebResearchRunner, compact_evidence, validate_json_data, validate_research_result from app.automation_service import web_search_workflow_template from app.automation_service import workflow_return_data from app.schemas import AutomationWorkflowSaveRequest class WebSearchHelpersTest(unittest.TestCase): def test_normalize_search_result_converts_percent_to_screen_point(self) -> None: result = normalize_search_result( { "title": "示例结果", "url": "https://example.com", "snippet": "摘要", "title_center_x_percent": 25, "title_center_y_percent": 40, }, scroll_page=2, width=1920, height=1080, ) self.assertIsNotNone(result) self.assertEqual(result["title_center_x"], 480) self.assertEqual(result["title_center_y"], 432) self.assertEqual(result["scroll_page"], 2) def test_result_identity_prefers_url_and_falls_back_to_title(self) -> None: self.assertEqual(result_identity({"url": "HTTPS://EXAMPLE.COM", "title": "标题"}), "https://example.com") self.assertEqual(result_identity({"url": "", "title": " 标题 "}), "标题") def test_ranking_ignores_invalid_and_duplicate_indexes(self) -> None: runner = WebSearchRunner( WorkflowContext(workflow_id=1, provider_id=1, model_id=1), {"query": "测试", "result_count": 2}, ) runner._text_json = lambda prompt: { "ranked_results": [ {"original_index": 1, "relevance_score": 9}, {"original_index": 1, "relevance_score": 8}, {"original_index": 99, "relevance_score": 7}, {"original_index": 0, "relevance_score": 6}, ] } ranked = runner._rank_results([{"title": "A"}, {"title": "B"}]) self.assertEqual([item["title"] for item in ranked], ["B", "A"]) def test_failed_title_location_keeps_search_page_state(self) -> None: runner = WebSearchRunner( WorkflowContext(workflow_id=1, provider_id=1, model_id=1), {"query": "测试", "click_attempts": 1}, ) runner._go_to_scroll_page = lambda scroll_page: None runner._capture = lambda: {"width": 1920, "height": 1080, "image_base64": "", "mime_type": "image/png"} runner._vision_json = lambda prompt, screenshot: {"found": False, "notes": "未找到标题"} result = runner._open_result({"title": "不存在的标题", "scroll_page": 0}) self.assertFalse(result["opened_detail_page"]) self.assertTrue(result["is_search_results_page"]) class WebSearchWorkflowTemplateTest(unittest.TestCase): def test_template_matches_workflow_schema(self) -> None: workflow = AutomationWorkflowSaveRequest.model_validate(web_search_workflow_template()) self.assertEqual(workflow.schema_version, "workflow/v1") self.assertEqual(workflow.workflow_key, "ai-web-research") self.assertEqual(workflow.variables["objective"]["default"], "") self.assertEqual([node.type for node in workflow.nodes], ["flow.start", "research.ai_web_research", "flow.end"]) def test_checked_in_workflow_matches_template(self) -> None: path = Path(__file__).resolve().parents[2] / "workflows" / "ai-web-research.workflow.json" checked_in = json.loads(path.read_text(encoding="utf-8")) self.assertEqual(checked_in, web_search_workflow_template()) class AiResearchHelpersTest(unittest.TestCase): def test_json_schema_validation_reports_missing_required_field(self) -> None: schema = { "type": "object", "required": ["summary"], "properties": {"summary": {"type": "string"}}, } result = validate_json_data({}, schema) self.assertFalse(result["schema_valid"]) self.assertTrue(result["errors"]) def test_compact_evidence_keeps_source_and_cleaned_content(self) -> None: evidence = compact_evidence( { "researched_details": [ { "visited_url": "https://example.com", "opened_detail_page": True, "result": {"title": "原始标题"}, "cleaned": {"clean_title": "清理标题", "clean_text": "正文", "key_points": ["要点"]}, } ] } ) self.assertEqual(evidence[0]["title"], "清理标题") self.assertEqual(evidence[0]["url"], "https://example.com") def test_research_validation_enforces_minimum_sources(self) -> None: result = validate_research_result( {"summary": "完成"}, { "type": "object", "required": ["summary"], "properties": {"summary": {"type": "string"}}, }, {"min_sources": 2}, [{"title": "A", "url": "https://example.com"}], ) self.assertTrue(result["schema_valid"]) self.assertFalse(result["constraints_valid"]) self.assertFalse(result["valid"]) def test_workflow_return_data_uses_configured_node(self) -> None: workflow = {"settings": {"return": {"node_id": "research"}}} result = {"outputs": {"research": {"data": {"answer": 1}}}} self.assertEqual(workflow_return_data(workflow, result), {"data": {"answer": 1}}) def test_ai_research_retries_until_assessment_is_valid(self) -> None: runner = AiWebResearchRunner( WorkflowContext(workflow_id=1, provider_id=1, model_id=1), { "objective": "测试目标", "output_schema": { "type": "object", "required": ["answer"], "properties": {"answer": {"type": "string"}}, }, "constraints": {"min_sources": 1}, "max_attempts": 2, }, ) runner._create_plan = lambda: {"queries": ["第一轮", "第二轮"]} assessments = iter( [ { "goal_achieved": False, "candidate_data": {}, "missing_information": ["答案"], "next_queries": ["第二轮"], }, { "goal_achieved": True, "candidate_data": {"answer": "完成"}, "missing_information": [], "next_queries": [], }, ] ) runner._assess_progress = lambda plan, queries, evidence: next(assessments) fake_output = { "result_count": 1, "researched_count": 1, "researched_details": [ { "visited_url": "https://example.com", "opened_detail_page": True, "result": {"title": "来源"}, "cleaned": {"clean_text": "证据"}, } ], } with patch("app.automation.nodes.research.WebSearchRunner") as search_runner: search_runner.return_value.run.return_value = fake_output result = runner.run() self.assertTrue(result["goal_achieved"]) self.assertEqual(result["attempts_used"], 2) self.assertEqual(result["data"], {"answer": "完成"}) if __name__ == "__main__": unittest.main()