mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-14 16:08:41 +00:00
5ffc2d3e27
* feat(mcp): add reliable task notifications and cancellation * feat(mcp): add background task chat UI * fix(mcp): hide and sanitize task notification prompts * fix(mcp): sanitize projected task names * fix(mcp): harden task notifications and details * fix(mcp): harden task lifecycle recovery * fix(mcp): gate task UI and isolate cancellations * test: scope plain-text response locator * fix(mcp): align task notification boundaries * fix(mcp): bound task delivery retries * fix background task notification races
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.gateway.deps import get_config
|
|
from app.gateway.routers import features
|
|
|
|
|
|
def _app_with_config(
|
|
*,
|
|
agents_api_enabled: bool,
|
|
browser_enabled: bool = False,
|
|
browser_extra: dict | None = None,
|
|
mcp_tasks_available: bool = False,
|
|
) -> FastAPI:
|
|
app = FastAPI()
|
|
app.state.mcp_tasks_available = mcp_tasks_available
|
|
app.include_router(features.router)
|
|
tools = (
|
|
[
|
|
SimpleNamespace(name="browser_navigate", model_extra=browser_extra or {}),
|
|
]
|
|
if browser_enabled
|
|
else []
|
|
)
|
|
fake_config = SimpleNamespace(agents_api=SimpleNamespace(enabled=agents_api_enabled), tools=tools)
|
|
app.dependency_overrides[get_config] = lambda: fake_config
|
|
return app
|
|
|
|
|
|
def test_features_reports_agents_api_enabled() -> None:
|
|
with TestClient(_app_with_config(agents_api_enabled=True)) as client:
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"agents_api": {"enabled": True},
|
|
"browser_control": {"enabled": False},
|
|
"mcp_tasks": {"enabled": False},
|
|
}
|
|
|
|
|
|
def test_features_reports_agents_api_disabled() -> None:
|
|
with TestClient(_app_with_config(agents_api_enabled=False)) as client:
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"agents_api": {"enabled": False},
|
|
"browser_control": {"enabled": False},
|
|
"mcp_tasks": {"enabled": False},
|
|
}
|
|
|
|
|
|
def test_features_reports_mcp_tasks_startup_capability() -> None:
|
|
with TestClient(_app_with_config(agents_api_enabled=True, mcp_tasks_available=True)) as client:
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json()["mcp_tasks"] == {"enabled": True}
|
|
|
|
|
|
def test_features_reports_browser_control_enabled_when_configured_and_runtime_available() -> None:
|
|
with (
|
|
patch("app.gateway.browser_capability.importlib.util.find_spec", return_value=object()),
|
|
TestClient(_app_with_config(agents_api_enabled=True, browser_enabled=True)) as client,
|
|
):
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json()["browser_control"] == {"enabled": True}
|
|
|
|
|
|
def test_features_reports_browser_control_disabled_when_runtime_missing() -> None:
|
|
with (
|
|
patch("app.gateway.browser_capability.importlib.util.find_spec", return_value=None),
|
|
TestClient(_app_with_config(agents_api_enabled=True, browser_enabled=True)) as client,
|
|
):
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json()["browser_control"] == {"enabled": False}
|
|
|
|
|
|
def test_features_reports_browser_control_disabled_for_unguarded_cdp() -> None:
|
|
with (
|
|
patch("app.gateway.browser_capability.importlib.util.find_spec", return_value=object()),
|
|
TestClient(
|
|
_app_with_config(
|
|
agents_api_enabled=True,
|
|
browser_enabled=True,
|
|
browser_extra={"cdp_url": "http://127.0.0.1:9222"},
|
|
),
|
|
) as client,
|
|
):
|
|
response = client.get("/api/features")
|
|
assert response.status_code == 200
|
|
assert response.json()["browser_control"] == {"enabled": False}
|