Browser_Agent/tests/test_agent_tools.py

import pytest
from datetime import datetime, timedelta, timezone
from agent.tools import execute_tool
from dataclasses import dataclass, field

@dataclass
class MockTask:
    steps: list = None
    params: dict = field(default_factory=dict)
    
    def __post_init__(self):
        if self.steps is None:
            self.steps = []

@pytest.mark.asyncio
async def test_tool_navigate_ssrf(mocker):
    page = mocker.AsyncMock()
    
    # Valid url
    res = await execute_tool("navigate", {"url": "https://example.com"}, page)
    assert res["status"] == "success"
    page.goto.assert_called_with(
        "https://example.com",
        wait_until="domcontentloaded",
        timeout=30000,
    )
    
    # SSRF blocked url
    res = await execute_tool("navigate", {"url": "http://169.254.169.254"}, page)
    assert "error" in res
    assert "Sicherheitsverstoß" in res["error"]

@pytest.mark.asyncio
async def test_tool_request_user_permission(mocker):
    page = mocker.AsyncMock()
    page.url = "https://example.com/login"
    page.context.cookies.return_value = []
    task = MockTask()
    res = await execute_tool(
        "request_user_permission",
        {"reason": "Login"},
        page,
        task=task,
    )
    assert res["status"] == "confirmation_required"
    assert res["confirmation"]["origin"] == "https://example.com"
    assert task.params["_confirmation_checkpoint"]["id"] == res["confirmation"]["id"]

@pytest.mark.asyncio
async def test_tool_request_temp_mail_bypass(mocker):
    page = mocker.AsyncMock()
    
    res = await execute_tool("request_temp_mail_bypass", {"email_address": "test@temp-mail.io"}, page)
    assert res["status"] == "error"
    assert res["error_code"] == "confirmation_required"

@pytest.mark.asyncio
async def test_tool_request_browser_handoff(mocker):
    page = mocker.AsyncMock()
    page.url = "https://example.com/verify"
    page.context.cookies.return_value = [{"name": "auth", "value": "123"}]
    
    res = await execute_tool("request_browser_handoff", {"reason": "captcha"}, page)
    assert res["status"] == "handoff_required"
    assert "[HANDOFF_REQUIRED|https://example.com/verify|" in res["handoff_signal"]

@pytest.mark.asyncio
async def test_tool_click_security(mocker):
    page = mocker.AsyncMock()
    page.url = "https://example.com/checkout"
    task = MockTask()
    
    # Mock JS eval to return True (simulating a form submit button)
    page.evaluate.side_effect = [True]
    
    res = await execute_tool("click", {"node_id": "42"}, page, task=task)
    assert "error" in res
    assert res["error_code"] == "confirmation_required"
    
    confirmation_id = "confirm-1"
    task.params["_approved_confirmation"] = {
        "id": confirmation_id,
        "origin": "https://example.com",
        "expires_at": (datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat(),
        "consumed": False,
    }
    page.evaluate.side_effect = [True, {"x": 100, "y": 200}]
    res2 = await execute_tool(
        "click",
        {"node_id": "42", "confirmation_id": confirmation_id},
        page,
        task=task,
    )
    assert res2["status"] == "success"
    page.mouse.move.assert_awaited_once()
    page.mouse.click.assert_awaited_once_with(100, 200)
    assert "_approved_confirmation" not in task.params


@pytest.mark.asyncio
async def test_tool_click_stale_id_returns_without_default_timeout(mocker):
    page = mocker.AsyncMock()
    setattr(page, "mutation_allowed", True)
    page.evaluate.return_value = None

    res = await execute_tool("click", {"node_id": "ax-old"}, page)

    assert "nicht mehr vorhanden" in res["error"]
    page.click.assert_not_awaited()

@pytest.mark.asyncio
async def test_tool_visual_check_returns_only_structured_report(mocker):
    page = mocker.AsyncMock()
    page.screenshot.return_value = b"fake_image_data"
    mocker.patch(
        "agent.tools.vision_subagent.analyze",
        new=mocker.AsyncMock(return_value={
            "status": "pass",
            "summary": "Layout ist stabil.",
            "issues": [],
            "confidence": 0.95,
            "suggestedNextAction": "Keine Änderung.",
            "model": "google/gemini-2.5-flash-lite",
            "costUsd": 0.0003,
        }),
    )

    res = await execute_tool("visual_check", {"goal": "Prüfe das Layout"}, page)
    assert res["status"] == "success"
    assert res["visual_check"]["status"] == "pass"
    assert "image_base64" not in res

@pytest.mark.asyncio
async def test_tool_read_dom(mocker, monkeypatch):
    page = mocker.AsyncMock()
    
    # Mock extract_dom_text to avoid running actual JS
    async def mock_extract(*args, **kwargs):
        return "[BTN:Login] [LINK:Home]"
    
    import browser.accessibility
    monkeypatch.setattr(browser.accessibility, "extract_dom_text", mock_extract)
    
    res = await execute_tool("read_dom", {}, page)
    assert res["status"] == "success"
    assert res["dom_text"] == "[BTN:Login] [LINK:Home]"

@pytest.mark.asyncio
async def test_tool_find_element(mocker):
    page = mocker.AsyncMock()
    # evaluate returns a JSON string of elements
    page.evaluate.return_value = '[{"node_id": "ax-7", "tag": "BUTTON", "text": "Login", "x": 100, "y": 200}]'
    
    res = await execute_tool("find_element", {"query": "login"}, page)
    assert res["status"] == "success"
    assert len(res["elements"]) == 1
    assert res["elements"][0]["tag"] == "BUTTON"
    assert res["elements"][0]["node_id"] == "ax-7"
    assert res["elements"][0]["x"] == 100

@pytest.mark.asyncio
async def test_tool_read_text_content(mocker):
    page = mocker.AsyncMock()
    page.evaluate.return_value = "Hello World"
    
    res = await execute_tool("read_text_content", {}, page)
    assert res["status"] == "success"
    assert res["text_content"] == "Hello World"