179 lines
8.1 KiB
Python
179 lines
8.1 KiB
Python
"""执行器：执行结果判定、模型错误的用户可读文案、历史精简、内嵌模式开关。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from backend.browser import runner
|
||
|
||
|
||
@pytest.fixture
|
||
def no_fallback_key(monkeypatch):
|
||
monkeypatch.setattr(runner, "get_llm_config", lambda: {"fallback_api_key": ""})
|
||
|
||
|
||
@pytest.fixture
|
||
def with_fallback_key(monkeypatch):
|
||
monkeypatch.setattr(runner, "get_llm_config", lambda: {"fallback_api_key": "FALLBACK"})
|
||
|
||
|
||
class TestVerdict:
|
||
def test_agent_finished_successfully(self, no_fallback_key):
|
||
verdict, summary = runner.verdict_from_history(
|
||
{"is_done": True, "is_successful": True, "final_result": "已提交填报"}, 60)
|
||
assert verdict == "success" and summary == "已提交填报"
|
||
|
||
def test_done_without_self_assessment_counts_as_success(self, no_fallback_key):
|
||
# is_successful 为 None 表示 Agent 没自评，不能当失败
|
||
verdict, _ = runner.verdict_from_history({"is_done": True, "is_successful": None, "final_result": "结束"}, 60)
|
||
assert verdict == "success"
|
||
|
||
def test_agent_self_assessed_failure(self, no_fallback_key):
|
||
verdict, summary = runner.verdict_from_history(
|
||
{"is_done": True, "is_successful": False, "final_result": "没找到入口"}, 60)
|
||
assert verdict == "failed" and summary == "没找到入口"
|
||
|
||
def test_step_limit_reached(self, no_fallback_key):
|
||
verdict, summary = runner.verdict_from_history({"is_done": False, "errors": []}, 60)
|
||
assert verdict == "failed" and "未在 60 步内完成" in summary
|
||
|
||
def test_step_limit_quotes_the_last_error(self, no_fallback_key):
|
||
_, summary = runner.verdict_from_history(
|
||
{"is_done": False, "errors": ["element not found", "click intercepted"]}, 60)
|
||
assert "click intercepted" in summary
|
||
|
||
|
||
class TestModelErrorMessages:
|
||
"""402 / 401 必须翻译成用户能照着做的话，不能只说「未在 N 步内完成」。"""
|
||
|
||
@pytest.mark.parametrize("error", [
|
||
"Error code: 402 - {'error': {'message': 'Insufficient Balance'}}",
|
||
"Insufficient Balance",
|
||
])
|
||
def test_out_of_balance_without_fallback_key_suggests_configuring_one(self, no_fallback_key, error):
|
||
_, summary = runner.verdict_from_history({"is_done": False, "errors": [error]}, 60)
|
||
assert "余额不足" in summary
|
||
assert "DEEPSEEK_API_KEY_FALLBACK" in summary
|
||
assert "未在 60 步内完成" not in summary
|
||
|
||
def test_out_of_balance_with_fallback_key_says_both_are_exhausted(self, with_fallback_key):
|
||
_, summary = runner.verdict_from_history(
|
||
{"is_done": False, "errors": ["Error code: 402 - Insufficient Balance"]}, 60)
|
||
assert "主 Key 与兜底 Key 都返回 402" in summary
|
||
assert "DEEPSEEK_API_KEY_FALLBACK" not in summary # 已经配过了，别再让用户去配
|
||
|
||
@pytest.mark.parametrize("error", ["Error code: 401 - Unauthorized", "invalid_api_key"])
|
||
def test_invalid_key(self, no_fallback_key, error):
|
||
_, summary = runner.verdict_from_history({"is_done": False, "errors": [error]}, 60)
|
||
assert "Key 无效" in summary and "401" in summary
|
||
|
||
def test_ordinary_errors_keep_the_step_limit_wording(self, no_fallback_key):
|
||
_, summary = runner.verdict_from_history({"is_done": False, "errors": ["timeout waiting for selector"]}, 40)
|
||
assert "未在 40 步内完成" in summary and "timeout waiting for selector" in summary
|
||
|
||
|
||
class TestCompactHistory:
|
||
class FakeHistory:
|
||
def __init__(self, **overrides):
|
||
self._values = {
|
||
"number_of_steps": lambda: 23,
|
||
"total_duration_seconds": lambda: 187.44,
|
||
"is_done": lambda: True,
|
||
"is_successful": lambda: True,
|
||
"urls": lambda: ["https://a/", "https://a/", "https://b/"],
|
||
"action_names": lambda: ["navigate", "click", "input_text"],
|
||
"errors": lambda: [None, "one error"],
|
||
"final_result": lambda: "done",
|
||
**overrides,
|
||
}
|
||
|
||
def __getattr__(self, name):
|
||
return self._values[name]
|
||
|
||
def test_extracts_the_fields_the_ui_and_logs_need(self):
|
||
compact = runner._compact_history(self.FakeHistory())
|
||
assert compact["steps"] == 23
|
||
assert compact["duration_s"] == 187.4
|
||
assert compact["is_done"] is True
|
||
assert compact["final_result"] == "done"
|
||
|
||
def test_consecutive_duplicate_urls_are_collapsed(self):
|
||
assert runner._compact_history(self.FakeHistory())["urls"] == ["https://a/", "https://b/"]
|
||
|
||
def test_empty_errors_are_dropped(self):
|
||
assert runner._compact_history(self.FakeHistory())["errors"] == ["one error"]
|
||
|
||
def test_a_broken_history_object_does_not_raise(self):
|
||
def boom():
|
||
raise RuntimeError("browser-use 换了接口")
|
||
|
||
compact = runner._compact_history(self.FakeHistory(number_of_steps=boom, urls=boom))
|
||
assert compact["steps"] == 0 and compact["urls"] == []
|
||
|
||
|
||
class TestEmbeddedMode:
|
||
def test_needs_both_cdp_and_control_urls(self, monkeypatch):
|
||
monkeypatch.setattr(runner, "CDP_URL", "http://127.0.0.1:9333")
|
||
monkeypatch.setattr(runner, "CTRL_URL", "http://127.0.0.1:8797")
|
||
assert runner.embedded_mode() is True
|
||
|
||
@pytest.mark.parametrize("cdp,ctrl", [("", "http://127.0.0.1:8797"), ("http://127.0.0.1:9333", ""), ("", "")])
|
||
def test_missing_either_one_falls_back_to_standalone_chrome(self, monkeypatch, cdp, ctrl):
|
||
monkeypatch.setattr(runner, "CDP_URL", cdp)
|
||
monkeypatch.setattr(runner, "CTRL_URL", ctrl)
|
||
assert runner.embedded_mode() is False
|
||
|
||
|
||
class TestEmbeddedExclusions:
|
||
def test_tab_switching_actions_are_excluded(self):
|
||
# Agent 切标签页会跳到界面页 app://ui，内嵌模式下必须禁掉
|
||
assert "switch" in runner.EMBEDDED_EXCLUDED_ACTIONS
|
||
assert "close" in runner.EMBEDDED_EXCLUDED_ACTIONS
|
||
|
||
def test_file_actions_are_excluded(self):
|
||
for action in ("upload_file", "write_file", "read_file"):
|
||
assert action in runner.EMBEDDED_EXCLUDED_ACTIONS
|
||
|
||
def test_build_tools_applies_the_exclusions(self):
|
||
from backend.browser.tools import build_tools
|
||
|
||
tools = build_tools("task_test", exclude=runner.EMBEDDED_EXCLUDED_ACTIONS)
|
||
registered = {a for a in tools.registry.registry.actions}
|
||
assert "ask_user" in registered and "report_blocker" in registered and "save_artifact" in registered
|
||
assert "switch" not in registered
|
||
|
||
|
||
class TestDockBounds:
|
||
def test_round_trip(self):
|
||
bounds = {"left": 100, "top": 50, "width": 800, "height": 600}
|
||
assert runner.set_dock_bounds(bounds) == bounds
|
||
assert runner.get_dock_bounds() == bounds
|
||
|
||
def test_values_are_coerced_to_int(self):
|
||
assert runner.set_dock_bounds({"left": 100.7, "top": 50.2, "width": 800.9, "height": 600.1})["left"] == 100
|
||
|
||
def test_clearing_bounds(self):
|
||
runner.set_dock_bounds({"left": 0, "top": 0, "width": 800, "height": 600})
|
||
runner.set_dock_bounds(None)
|
||
assert runner.get_dock_bounds() is None
|
||
|
||
|
||
class TestWindowKwargs:
|
||
def test_position_and_size_are_passed_through(self):
|
||
kwargs = runner.window_kwargs_for({"left": 1200, "top": 100, "width": 800, "height": 600})
|
||
assert kwargs["window_position"] == {"width": 1200, "height": 100}
|
||
assert kwargs["window_size"] == {"width": 800, "height": 600}
|
||
|
||
def test_negative_coordinates_are_clamped_to_zero(self):
|
||
# 回归：多屏时 Electron 上报 top=-521，browser-use 校验 window_position >= 0，
|
||
# 不夹取会让用户的真实任务在启动浏览器时直接失败
|
||
kwargs = runner.window_kwargs_for({"left": -1200, "top": -521, "width": 800, "height": 600})
|
||
assert kwargs["window_position"] == {"width": 0, "height": 0}
|
||
|
||
def test_tiny_dock_area_gets_a_usable_minimum_size(self):
|
||
kwargs = runner.window_kwargs_for({"left": 0, "top": 0, "width": 10, "height": 10})
|
||
assert kwargs["window_size"] == {"width": 320, "height": 240}
|
||
|
||
def test_no_dock_bounds_means_no_window_arguments(self):
|
||
assert runner.window_kwargs_for(None) == {}
|