deerflow-code/offline-backend-20260512/backend/tests/test_browser_act.py

"""Tests for the interactive browser_act tool and its act-task queue support."""

from __future__ import annotations

import asyncio

import pytest

from deerflow.browser_context import tasks as task_queue
from deerflow.config.browser_context_config import BrowserContextConfig
from deerflow.tools.builtins import browser_act_tool as act_mod


@pytest.fixture(autouse=True)
def _isolated_task_home(tmp_path, monkeypatch):
    """Point the file-backed task queue at a throwaway directory."""
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    yield


def _enable_actions(monkeypatch, *, enabled=True, actions_enabled=True):
    cfg = BrowserContextConfig(
        enabled=enabled,
        actions_enabled=actions_enabled,
        action_result_timeout_seconds=5,
        task_poll_interval_seconds=0.25,
        task_lease_seconds=30,
    )
    monkeypatch.setattr(act_mod, "_settings", lambda: cfg)
    return cfg


# --- task queue layer -------------------------------------------------------


def test_act_task_round_trip_carries_session_and_observation():
    task = task_queue.create_browser_task(
        user_id="u1",
        url="https://example.com",
        kind="act",
        session_id="sess-1",
        action={"type": "navigate", "url": "https://example.com"},
    )
    assert task["kind"] == "act"
    assert task["session_id"] == "sess-1"

    claimed = task_queue.claim_next_browser_task(user_id="u1", worker_id="w1")
    assert claimed is not None
    assert claimed["kind"] == "act"
    assert claimed["session_id"] == "sess-1"
    assert claimed["action"] == {"type": "navigate", "url": "https://example.com"}

    observation = {
        "action_status": "ok",
        "url": "https://example.com/home",
        "title": "Home",
        "elements": [{"index": 0, "tag": "button", "label": "Login"}],
        "text": "welcome",
    }
    done = task_queue.complete_browser_task(
        task_id=task["id"],
        user_id="u1",
        observation=observation,
    )
    assert done is not None
    assert done["status"] == "done"
    assert done["result"]["observation"] == observation


def test_fetch_task_default_kind_is_fetch():
    task = task_queue.create_browser_task(user_id="u1", url="https://example.com")
    assert task["kind"] == "fetch"
    claimed = task_queue.claim_next_browser_task(user_id="u1", worker_id="w1")
    assert claimed["kind"] == "fetch"
    assert claimed["session_id"] == ""
    assert claimed["action"] is None


# --- tool validation --------------------------------------------------------


async def _invoke(**kwargs) -> str:
    return await act_mod.browser_act_tool.ainvoke(kwargs)


@pytest.mark.asyncio
async def test_browser_act_disabled_when_actions_off(monkeypatch):
    _enable_actions(monkeypatch, enabled=True, actions_enabled=False)
    out = await _invoke(action="navigate", url="https://example.com")
    assert "actions_enabled is false" in out


@pytest.mark.asyncio
async def test_browser_act_unknown_action(monkeypatch):
    _enable_actions(monkeypatch)
    out = await _invoke(action="frobnicate")
    assert "unknown action" in out


@pytest.mark.asyncio
async def test_browser_act_navigate_requires_url(monkeypatch):
    _enable_actions(monkeypatch)
    out = await _invoke(action="navigate")
    assert "requires an exact http(s) URL" in out


@pytest.mark.asyncio
async def test_browser_act_non_navigate_needs_session(monkeypatch):
    _enable_actions(monkeypatch)
    out = await _invoke(action="click", index=0)
    assert "no session_id yet" in out


@pytest.mark.asyncio
async def test_browser_act_click_requires_index(monkeypatch):
    _enable_actions(monkeypatch)
    out = await _invoke(action="click", session_id="sess-1")
    assert "requires a non-negative element index" in out


# --- tool success path (simulated extension) --------------------------------


@pytest.mark.asyncio
async def test_browser_act_returns_formatted_observation(monkeypatch):
    _enable_actions(monkeypatch)

    async def fake_extension():
        # Wait for the tool to create the pending act task, then complete it.
        for _ in range(40):
            claimed = task_queue.claim_next_browser_task(user_id="default", worker_id="ext")
            if claimed is not None:
                task_queue.complete_browser_task(
                    task_id=claimed["id"],
                    user_id="default",
                    observation={
                        "action_status": "ok",
                        "url": "https://example.com/home",
                        "title": "Home Page",
                        "elements": [
                            {"index": 0, "tag": "input", "type": "search", "label": "Search", "value": ""},
                            {"index": 1, "tag": "button", "label": "Go"},
                        ],
                        "text": "Welcome to the home page.",
                    },
                )
                return
            await asyncio.sleep(0.1)

    ext = asyncio.create_task(fake_extension())
    out = await _invoke(action="navigate", url="https://example.com")
    await ext

    assert "Browser session:" in out
    assert "Home Page" in out
    assert "[0] input[search]" in out
    assert "[1] button" in out
    assert "Welcome to the home page." in out