mcp-autogui-multinode

import pytest
from core.config import settings
from core.logger import logger
from fastmcp import Client
from mcp import types
from mcp import StdioServerParameters
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).parent.parent
MCP_LOCAL_SCRIPT = PROJECT_ROOT / "mcp_local.py"

MCP_BASE_URL = f"http://localhost:{settings.mcp_port}/mcp"

# must start the mcp server before running the tests
# example:
# First start the mcp server:
# uv run python mcp_local.py http
# Then run the tests:
# uv run pytest tests/test_local_mcp_client.py

@pytest.mark.asyncio
async def test_move_mouse_with_client():
    """Test moving the mouse via client"""
    logger.info("Testing move_mouse with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("move_mouse", {"x": 100, "y": 200})
        logger.info("MCP move_mouse response: {}", response)
        assert response is not None
        # Response should be a dict with output/error
        if isinstance(response, dict):
            assert "output" in response or "error" in response
        elif hasattr(response, 'content'):
            # Handle FastMCP response format
            assert response.content is not None

@pytest.mark.asyncio
@pytest.mark.parametrize("button_type,description", [
    ("left", "Left click"),
    ("right", "Right click"),
    ("middle", "Middle click"),
    ("double_left", "Double click"),
])
async def test_click_mouse_types_with_client(button_type, description):
    """Test different mouse click types: left, right, and double click"""
    logger.info("Testing click_mouse ({}) with client, {}", description, MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("click_mouse", {
            "x": 100,
            "y": 200,
            "button": button_type
        })
        logger.info("MCP click_mouse ({}) response: {}", description, response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response
        elif hasattr(response, 'content'):
            assert response.content is not None


@pytest.mark.asyncio
async def test_get_cursor_position_with_client():
    """Test getting cursor position via client"""
    logger.info("Testing get_cursor_position with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("get_cursor_position", {})
        logger.info("MCP get_cursor_position response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response
            # Cursor position should have PositionX and PositionY
            if response.get("output"):
                output = response["output"]
                assert isinstance(output, dict)
                assert "PositionX" in output or "PositionY" in output


@pytest.mark.asyncio
async def test_take_screenshot_with_client():
    """Test taking screenshot via client"""
    logger.info("Testing take_screenshot with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("take_screenshot", {})
        logger.info("MCP take_screenshot response: {}", response)
        assert response is not None
        if isinstance(response, list):
            # Screenshot should return base64 encoded image
            if len(response) > 0:
                for item in response:
                    if item.type == "image":
                        image = item.data
                        assert image is not None
                    elif item.type == "text":
                        text = item.text
                        assert text is not None
                    else:
                        assert False


@pytest.mark.asyncio
async def test_press_mouse_with_client():
    """Test pressing mouse button via client"""
    logger.info("Testing press_mouse with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("press_mouse", {"x": 100, "y": 190, "button": "left"})
        logger.info("MCP press_mouse response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response


@pytest.mark.asyncio
async def test_release_mouse_with_client():
    """Test releasing mouse button via client"""
    logger.info("Testing release_mouse with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("release_mouse", {"x": 100, "y": 200, "button": "left"})
        logger.info("MCP release_mouse response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response


@pytest.mark.asyncio
async def test_drag_mouse_with_client():
    """Test dragging mouse via client"""
    logger.info("Testing drag_mouse with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("drag_mouse", {
            "source_x": 100,
            "source_y": 200,
            "target_x": 300,
            "target_y": 400
        })
        logger.info("MCP drag_mouse response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response


@pytest.mark.asyncio
async def test_scroll_with_client():
    """Test scrolling via client"""
    logger.info("Testing scroll with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        # 移动到屏幕中央的安全区域，避免在系统UI区域滚动
        await client.call_tool("move_mouse", {"x": 500, "y": 500})
        await client.call_tool("wait", {"duration": 100})
        # 在安全位置进行滚动，使用屏幕中央坐标
        response = await client.call_tool("scroll", {
            "scroll_direction": "up",
            "scroll_amount": 3,
            "x": 500,  # 使用屏幕中央位置，远离系统UI区域
            "y": 500   # 避免在顶部菜单栏或底部Dock区域
        })
        logger.info("MCP scroll response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response
        elif hasattr(response, 'content'):
            assert response.content is not None

@pytest.mark.asyncio
async def test_scroll_down_with_client():
    """Test scrolling down via client"""
    logger.info("Testing scroll down with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        # 先移动到屏幕中央的安全区域（避免在系统UI区域滚动）
        # 使用较大的 y 坐标，避免在顶部菜单栏区域
        await client.call_tool("move_mouse", {"x": 500, "y": 500})
        # 添加短暂延迟，确保鼠标位置稳定
        await client.call_tool("wait", {"duration": 100})
        # 在安全位置进行滚动，明确指定坐标，使用较小的滚动量
        response = await client.call_tool("scroll", {
            "scroll_direction": "down",
            "scroll_amount": 1,
            "x": 500,  # 明确指定坐标，避免使用当前鼠标位置
            "y": 500   # 使用屏幕中央位置，远离系统UI区域
        })
        logger.info("MCP scroll down response: {}", response)
        assert response is not None
        if isinstance(response, types.Content):
            assert response.text is not None


@pytest.mark.asyncio
async def test_press_key_with_client():
    """Test pressing key via client"""
    logger.info("Testing press_key with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("press_key", {"key": "enter"})
        logger.info("MCP press_key response: {}", response)
        assert response is not None
        if isinstance(response, types.Content):
            assert response.text is not None


@pytest.mark.asyncio
async def test_press_key_combination_with_client():
    """Test pressing key combination via client"""
    logger.info("Testing press_key combination with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("press_key", {"key": "ctrl space"})
        logger.info("MCP press_key combination response: {}", response)
        assert response is not None
        if isinstance(response, types.Content):
            assert response.text is not None


@pytest.mark.asyncio
async def test_type_text_with_client():
    """Test typing text via client"""
    logger.info("Testing type_text with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("type_text", {"text": "Hello, World!"})
        logger.info("MCP type_text response: {}", response)
        assert response is not None
        if isinstance(response, types.Content):
            assert response.text is not None


@pytest.mark.asyncio
async def test_wait_with_client():
    """Test waiting via client"""
    logger.info("Testing wait with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("wait", {"duration": 100})
        logger.info("MCP wait response: {}", response)
        assert response is not None
        if isinstance(response, types.Content):
            assert response.text is not None


@pytest.mark.asyncio
async def test_click_mouse_right_button_with_client():
    """Test right clicking mouse via client"""
    logger.info("Testing click_mouse right button with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("click_mouse", {"x": 100, "y": 200, "button": "right"})
        logger.info("MCP click_mouse right button response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response
        elif hasattr(response, 'content'):
            assert response.content is not None


@pytest.mark.asyncio
async def test_click_mouse_double_click_with_client():
    """Test double clicking mouse via client"""
    logger.info("Testing click_mouse double click with client, {}", MCP_BASE_URL)
    client = Client(MCP_BASE_URL)
    async with client:
        response = await client.call_tool("click_mouse", {"x": 100, "y": 200, "button": "double_left"})
        logger.info("MCP click_mouse double click response: {}", response)
        assert response is not None
        if isinstance(response, dict):
            assert "output" in response or "error" in response
        elif hasattr(response, 'content'):
            assert response.content is not None