win_monitor/backend/tests/test_web_search.py

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()