# SPDX-FileCopyrightText: 2025 Weibo, Inc.
#
# SPDX-License-Identifier: Apache-2.0

import json
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from executor.agents.dify.dify_agent import DifyAgent
from shared.models.execution import ExecutionRequest
from shared.status import TaskStatus

# Test constants
TASK_ID = 123
SUBTASK_ID = 456


def create_mock_emitter():
    """Create a mock emitter for testing."""
    emitter = MagicMock()
    emitter.in_progress = AsyncMock()
    emitter.start = AsyncMock()
    emitter.done = AsyncMock()
    emitter.error = AsyncMock()
    emitter.text_delta = AsyncMock()
    return emitter


class TestDifyAgent:
    """Test cases for DifyAgent"""

    @pytest.fixture(autouse=True)
    def mock_http_requests(self):
        """
        Mock all HTTP requests to prevent actual network calls during tests.
        - requests.get: DifyAgent.__init__ calls _get_app_mode() which makes GET to /v1/info
        - CallbackClient: Uses TracedSession (not raw requests.post), so we mock
          the entire class to prevent retry loops with exponential backoff on empty URLs.
        - EmitterBuilder: Mock to prevent asyncio event loop issues in tests.
        """
        with (
            patch("executor.agents.dify.dify_agent.requests.get") as mock_get,
            patch("executor.agents.base.CallbackClient") as mock_callback_cls,
            patch("shared.models.EmitterBuilder") as mock_builder_cls,
        ):
            # Mock GET response for _get_app_mode()
            mock_get_response = MagicMock()
            mock_get_response.status_code = 200
            mock_get_response.json.return_value = {"mode": "chat"}
            mock_get.return_value = mock_get_response

            # Mock CallbackClient instance
            mock_callback = MagicMock()
            mock_callback.send_event.return_value = {"status": "success"}
            mock_callback_cls.return_value = mock_callback

            # Mock EmitterBuilder to prevent asyncio issues
            mock_emitter = MagicMock()
            mock_emitter.in_progress.return_value = {"status": "success"}
            mock_builder = MagicMock()
            mock_builder.with_task.return_value = mock_builder
            mock_builder.with_transport.return_value = mock_builder
            mock_builder.with_executor_info.return_value = mock_builder
            mock_builder.build.return_value = mock_emitter
            mock_builder_cls.return_value = mock_builder

            yield {
                "get": mock_get,
                "callback": mock_callback,
                "emitter": mock_emitter,
            }

    @pytest.fixture
    def mock_emitter(self) -> MagicMock:
        """Create a mock emitter for testing"""
        return create_mock_emitter()

    @pytest.fixture
    def task_data(self) -> ExecutionRequest:
        """Sample task data for testing"""
        return ExecutionRequest(
            task_id=TASK_ID,
            subtask_id=SUBTASK_ID,
            prompt="Hello Dify",
            bot=[
                {
                    "agent_config": {
                        "env": {
                            "DIFY_API_KEY": "app-test-api-key",
                            "DIFY_BASE_URL": "https://api.dify.ai",
                            "DIFY_APP_ID": "app-default-123",
                        }
                    }
                }
            ],
            user={"user_name": "testuser"},
        )

    def test_init(self, task_data: ExecutionRequest, mock_emitter: MagicMock) -> None:
        """Test DifyAgent initialization"""
        agent = DifyAgent(task_data, mock_emitter)

        assert agent is not None
        assert agent.task_id == TASK_ID
        assert agent.prompt == "Hello Dify"
        # Without bot_prompt, dify_app_id and params come from config (empty)
        assert agent.dify_app_id == "app-default-123"  # From DIFY_APP_ID in config
        assert agent.params == {}  # No params in config
        assert agent.dify_config["api_key"] == "app-test-api-key"
        assert agent.dify_config["base_url"] == "https://api.dify.ai"

    def test_init_without_bot_prompt(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test DifyAgent initialization without bot_prompt"""
        agent = DifyAgent(task_data, mock_emitter)

        assert agent.dify_app_id == "app-default-123"  # Should use default from config
        assert agent.params == {}

    def test_parse_bot_prompt_valid(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test parsing valid bot_prompt"""
        agent = DifyAgent(task_data, mock_emitter)

        bot_prompt_json = json.dumps(
            {
                "difyAppId": "app-test-123",
                "params": {"customer_name": "John Doe", "language": "en-US"},
            }
        )
        app_id, params = agent._parse_bot_prompt(bot_prompt_json)

        assert app_id == "app-test-123"
        assert params == {"customer_name": "John Doe", "language": "en-US"}

    def test_parse_bot_prompt_invalid_json(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test parsing invalid JSON bot_prompt"""
        agent = DifyAgent(task_data, mock_emitter)

        app_id, params = agent._parse_bot_prompt("invalid json")

        assert app_id is None
        assert params == {}

    def test_parse_bot_prompt_empty(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test parsing empty bot_prompt"""
        agent = DifyAgent(task_data, mock_emitter)

        app_id, params = agent._parse_bot_prompt("")

        assert app_id is None
        assert params == {}

    def test_validate_config_success(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test config validation with valid config"""
        agent = DifyAgent(task_data, mock_emitter)

        result = agent._validate_config()

        assert result is True

    def test_validate_config_missing_api_key(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test config validation with missing API key"""
        task_data.bot[0]["agent_config"]["env"]["DIFY_API_KEY"] = ""
        agent = DifyAgent(task_data, mock_emitter)

        result = agent._validate_config()

        assert result is False

    def test_validate_config_missing_base_url(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test config validation with missing base URL"""
        task_data.bot[0]["agent_config"]["env"]["DIFY_BASE_URL"] = ""
        agent = DifyAgent(task_data, mock_emitter)

        result = agent._validate_config()

        assert result is False

    def test_validate_config_missing_app_id(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test config validation with missing app ID - app_id is now optional"""
        task_data.bot[0]["agent_config"]["env"]["DIFY_APP_ID"] = ""
        agent = DifyAgent(task_data, mock_emitter)

        result = agent._validate_config()

        # DIFY_APP_ID is no longer required since each API key corresponds to one app
        assert result is True

    @patch("executor.agents.dify.dify_agent.requests.post")
    def test_call_dify_api_success(
        self, mock_post: MagicMock, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test successful Dify API call"""
        # Mock streaming response
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.iter_lines.return_value = [
            b'data: {"event": "message", "answer": "Hello", "conversation_id": "conv-123"}',
            b'data: {"event": "message", "answer": " World"}',
            b'data: {"event": "message_end"}',
        ]
        mock_post.return_value = mock_response

        agent = DifyAgent(task_data, mock_emitter)
        result = agent._call_dify_api("Test query")

        assert result["answer"] == "Hello World"
        assert result["conversation_id"] == "conv-123"
        assert mock_post.called

    @patch("executor.agents.dify.dify_agent.requests.post")
    def test_call_dify_api_error_response(
        self, mock_post: MagicMock, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test Dify API call with error response"""
        # Mock error response
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.iter_lines.return_value = [
            b'data: {"event": "error", "message": "Invalid app ID"}'
        ]
        mock_post.return_value = mock_response

        agent = DifyAgent(task_data, mock_emitter)

        with pytest.raises(Exception) as exc_info:
            agent._call_dify_api("Test query")

        assert "Dify API error" in str(exc_info.value)

    @patch("executor.agents.dify.dify_agent.requests.post")
    def test_call_dify_api_http_error(
        self, mock_post: MagicMock, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test Dify API call with HTTP error"""
        # Mock HTTP error
        mock_response = MagicMock()
        mock_response.status_code = 404
        mock_response.raise_for_status.side_effect = Exception("404 Not Found")
        mock_post.return_value = mock_response

        agent = DifyAgent(task_data, mock_emitter)

        with pytest.raises(Exception):
            agent._call_dify_api("Test query")

    @patch.object(DifyAgent, "_call_dify_api")
    @patch.object(DifyAgent, "_validate_config")
    def test_execute_success(
        self,
        mock_validate: MagicMock,
        mock_call_api: MagicMock,
        task_data: ExecutionRequest,
        mock_emitter: MagicMock,
    ) -> None:
        """Test successful execution"""
        mock_validate.return_value = True
        mock_call_api.return_value = {
            "answer": "This is the answer from Dify",
            "conversation_id": "conv-123",
        }

        agent = DifyAgent(task_data, mock_emitter)
        result = agent.execute()

        assert result == TaskStatus.COMPLETED
        assert mock_validate.called
        assert mock_call_api.called

    @patch.object(DifyAgent, "_validate_config")
    def test_execute_invalid_config(
        self,
        mock_validate: MagicMock,
        task_data: ExecutionRequest,
        mock_emitter: MagicMock,
    ) -> None:
        """Test execution with invalid config"""
        mock_validate.return_value = False

        agent = DifyAgent(task_data, mock_emitter)
        result = agent.execute()

        assert result == TaskStatus.FAILED

    @patch.object(DifyAgent, "_call_dify_api")
    @patch.object(DifyAgent, "_validate_config")
    def test_execute_no_answer(
        self,
        mock_validate: MagicMock,
        mock_call_api: MagicMock,
        task_data: ExecutionRequest,
        mock_emitter: MagicMock,
    ) -> None:
        """Test execution with no answer from API"""
        mock_validate.return_value = True
        mock_call_api.return_value = {"answer": "", "conversation_id": "conv-123"}

        agent = DifyAgent(task_data, mock_emitter)
        result = agent.execute()

        assert result == TaskStatus.FAILED

    @patch.object(DifyAgent, "_call_dify_api")
    @patch.object(DifyAgent, "_validate_config")
    def test_execute_api_exception(
        self,
        mock_validate: MagicMock,
        mock_call_api: MagicMock,
        task_data: ExecutionRequest,
        mock_emitter: MagicMock,
    ) -> None:
        """Test execution with API exception"""
        mock_validate.return_value = True
        mock_call_api.side_effect = Exception("API call failed")

        agent = DifyAgent(task_data, mock_emitter)
        result = agent.execute()

        assert result == TaskStatus.FAILED

    def test_conversation_id_management(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test conversation ID management"""
        # Clear any existing conversation state before test
        DifyAgent.clear_conversation(task_data.task_id)

        agent = DifyAgent(task_data, mock_emitter)

        # Initially empty
        assert agent.conversation_id == ""

        # Save conversation ID
        agent._save_conversation_id("conv-test-123")

        # Create new agent instance for same task
        agent2 = DifyAgent(task_data, mock_emitter)
        assert agent2.conversation_id == "conv-test-123"

        # Clear conversation
        DifyAgent.clear_conversation(task_data.task_id)
        agent3 = DifyAgent(task_data, mock_emitter)
        assert agent3.conversation_id == ""

    def test_get_name(
        self, task_data: ExecutionRequest, mock_emitter: MagicMock
    ) -> None:
        """Test get_name method"""
        agent = DifyAgent(task_data, mock_emitter)

        assert agent.get_name() == "Dify"
