wehub-resource-sync
4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
581 lines
22 KiB
Python
581 lines
22 KiB
Python
"""Unit tests for the OpenClaw bridge integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from integrations.catalog import classify_integrations as _classify_integrations
|
|
from integrations.openclaw import (
|
|
OpenClawConfig,
|
|
_tool_result_to_dict,
|
|
build_openclaw_config,
|
|
describe_openclaw_error,
|
|
list_openclaw_tools,
|
|
openclaw_config_from_env,
|
|
openclaw_runtime_unavailable_reason,
|
|
validate_openclaw_config,
|
|
)
|
|
from integrations.openclaw.tools.openclaw_mcp_tool import (
|
|
call_openclaw_bridge_tool,
|
|
search_openclaw_conversations,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OpenClawConfig
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOpenClawConfig:
|
|
def test_default_mode(self) -> None:
|
|
# mode defaults to streamable-http; without url the validator should reject
|
|
with pytest.raises(ValidationError):
|
|
OpenClawConfig()
|
|
|
|
def test_streamable_http_requires_url(self) -> None:
|
|
with pytest.raises(ValidationError, match="requires a non-empty url"):
|
|
OpenClawConfig(mode="streamable-http", url="")
|
|
|
|
def test_sse_requires_url(self) -> None:
|
|
with pytest.raises(ValidationError, match="requires a non-empty url"):
|
|
OpenClawConfig(mode="sse", url="")
|
|
|
|
def test_stdio_requires_command(self) -> None:
|
|
with pytest.raises(ValidationError, match="requires a non-empty command"):
|
|
OpenClawConfig(mode="stdio", command="")
|
|
|
|
def test_valid_streamable_http(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp")
|
|
assert config.mode == "streamable-http"
|
|
assert config.url == "https://openclaw.example.com/mcp"
|
|
assert config.is_configured is True
|
|
|
|
def test_valid_sse(self) -> None:
|
|
config = OpenClawConfig(mode="sse", url="https://openclaw.example.com/sse")
|
|
assert config.mode == "sse"
|
|
assert config.is_configured is True
|
|
|
|
def test_valid_stdio(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp")
|
|
assert config.mode == "stdio"
|
|
assert config.command == "openclaw-mcp"
|
|
assert config.is_configured is True
|
|
|
|
def test_url_trailing_slash_stripped(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp/")
|
|
assert config.url == "https://openclaw.example.com/mcp"
|
|
|
|
def test_auth_token_bearer_prefix_stripped(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp", auth_token="Bearer tok123")
|
|
assert config.auth_token == "tok123"
|
|
|
|
def test_mode_normalized_to_lowercase(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp", mode="Streamable-HTTP")
|
|
assert config.mode == "streamable-http"
|
|
|
|
def test_args_tuple_normalized(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp", args=["--port", "8080"])
|
|
assert config.args == ("--port", "8080")
|
|
|
|
def test_args_empty_strings_filtered(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp", args=[" ", "flag", ""])
|
|
assert config.args == ("flag",)
|
|
|
|
def test_headers_non_dict_becomes_empty(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp", headers="bad") # type: ignore[arg-type]
|
|
assert config.headers == {}
|
|
|
|
def test_request_headers_injects_bearer(self) -> None:
|
|
config = OpenClawConfig(
|
|
url="https://openclaw.example.com/mcp",
|
|
auth_token="mytoken",
|
|
)
|
|
assert config.request_headers["Authorization"] == "Bearer mytoken"
|
|
|
|
def test_request_headers_no_duplicate_authorization(self) -> None:
|
|
config = OpenClawConfig(
|
|
url="https://openclaw.example.com/mcp",
|
|
auth_token="mytoken",
|
|
headers={"Authorization": "Bearer override"},
|
|
)
|
|
assert config.request_headers["Authorization"] == "Bearer override"
|
|
|
|
def test_is_configured_false_when_no_url_no_command(self) -> None:
|
|
# Build via dict to bypass model_validator (mode=stdio, no command)
|
|
# The only way to get is_configured=False at runtime is an unconfigured stdio
|
|
# We can test the property branch by building a valid stdio and checking command=False
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp")
|
|
assert config.is_configured is True
|
|
# Simulate cleared command (property reads self.command)
|
|
assert bool("") is False # sanity
|
|
|
|
def test_timeout_must_be_positive(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
OpenClawConfig(url="https://openclaw.example.com/mcp", timeout_seconds=0)
|
|
|
|
def test_integration_id_defaults_empty(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp")
|
|
assert config.integration_id == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_openclaw_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBuildOpenClawConfig:
|
|
def test_none_raises_for_unconfigured(self) -> None:
|
|
# Empty dict triggers the model validator
|
|
with pytest.raises(ValidationError):
|
|
build_openclaw_config(None)
|
|
|
|
def test_basic_http(self) -> None:
|
|
config = build_openclaw_config(
|
|
{"url": "https://openclaw.example.com/mcp", "auth_token": "tok"}
|
|
)
|
|
assert config.url == "https://openclaw.example.com/mcp"
|
|
assert config.auth_token == "tok"
|
|
|
|
def test_stdio_config(self) -> None:
|
|
config = build_openclaw_config(
|
|
{"mode": "stdio", "command": "openclaw-mcp", "args": ["--port", "8080"]}
|
|
)
|
|
assert config.mode == "stdio"
|
|
assert config.command == "openclaw-mcp"
|
|
assert config.args == ("--port", "8080")
|
|
|
|
def test_integration_id_round_trip(self) -> None:
|
|
config = build_openclaw_config(
|
|
{
|
|
"url": "https://openclaw.example.com/mcp",
|
|
"integration_id": "int-42",
|
|
}
|
|
)
|
|
assert config.integration_id == "int-42"
|
|
|
|
def test_helper_metadata_is_ignored(self) -> None:
|
|
config = build_openclaw_config(
|
|
{
|
|
"url": "https://openclaw.example.com/mcp",
|
|
"connection_verified": True,
|
|
"search_query": "checkout-api",
|
|
}
|
|
)
|
|
assert config.url == "https://openclaw.example.com/mcp"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# openclaw_config_from_env
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOpenClawConfigFromEnv:
|
|
def test_returns_none_when_no_url_or_command(self) -> None:
|
|
with patch.dict(os.environ, {}, clear=False):
|
|
for var in ("OPENCLAW_MCP_URL", "OPENCLAW_MCP_COMMAND"):
|
|
os.environ.pop(var, None)
|
|
assert openclaw_config_from_env() is None
|
|
|
|
@patch.dict(
|
|
os.environ,
|
|
{
|
|
"OPENCLAW_MCP_URL": "https://openclaw.example.com/mcp",
|
|
"OPENCLAW_MCP_AUTH_TOKEN": "env-token",
|
|
"OPENCLAW_MCP_MODE": "streamable-http",
|
|
},
|
|
)
|
|
def test_streamable_http_from_env(self) -> None:
|
|
config = openclaw_config_from_env()
|
|
assert config is not None
|
|
assert config.url == "https://openclaw.example.com/mcp"
|
|
assert config.auth_token == "env-token"
|
|
assert config.mode == "streamable-http"
|
|
|
|
@patch.dict(
|
|
os.environ,
|
|
{
|
|
"OPENCLAW_MCP_MODE": "stdio",
|
|
"OPENCLAW_MCP_COMMAND": "openclaw-mcp",
|
|
"OPENCLAW_MCP_ARGS": "--port 8080",
|
|
},
|
|
)
|
|
def test_stdio_from_env(self) -> None:
|
|
config = openclaw_config_from_env()
|
|
assert config is not None
|
|
assert config.mode == "stdio"
|
|
assert config.command == "openclaw-mcp"
|
|
assert config.args == ("--port", "8080")
|
|
|
|
@patch.dict(os.environ, {"OPENCLAW_MCP_MODE": "stdio"})
|
|
def test_stdio_without_command_returns_none(self) -> None:
|
|
os.environ.pop("OPENCLAW_MCP_COMMAND", None)
|
|
assert openclaw_config_from_env() is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# validate_openclaw_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidateOpenClawConfig:
|
|
def _http_config(self) -> OpenClawConfig:
|
|
return OpenClawConfig(url="https://openclaw.example.com/mcp")
|
|
|
|
def test_ok_result_on_success(self) -> None:
|
|
config = self._http_config()
|
|
mock_tools = [{"name": "conversations_get"}, {"name": "conversations_list"}]
|
|
|
|
with patch("integrations.openclaw.list_openclaw_tools", return_value=mock_tools):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is True
|
|
assert result.detail == (
|
|
"OpenClaw bridge connected via streamable-http (https://openclaw.example.com/mcp); "
|
|
"discovered 2 tool(s)."
|
|
)
|
|
assert result.tool_names == ("conversations_get", "conversations_list")
|
|
|
|
def test_failed_result_on_exception(self) -> None:
|
|
config = self._http_config()
|
|
|
|
with patch(
|
|
"integrations.openclaw.list_openclaw_tools",
|
|
side_effect=RuntimeError("connection refused"),
|
|
):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is False
|
|
assert "connection refused" in result.detail
|
|
|
|
def test_local_dashboard_url_returns_stdio_hint(self) -> None:
|
|
config = OpenClawConfig(mode="streamable-http", url="http://127.0.0.1:18789/")
|
|
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is False
|
|
assert "Control UI/Gateway" in result.detail
|
|
assert "openclaw" in result.detail
|
|
assert "mcp serve" in result.detail
|
|
|
|
def test_exception_group_is_unwrapped(self) -> None:
|
|
config = self._http_config()
|
|
nested = ExceptionGroup(
|
|
"unhandled errors in a TaskGroup",
|
|
[RuntimeError("HTTP 404 from POST https://openclaw.example.com/mcp")],
|
|
)
|
|
|
|
with patch("integrations.openclaw.list_openclaw_tools", side_effect=nested):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is False
|
|
assert "HTTP 404" in result.detail
|
|
assert "TaskGroup" not in result.detail
|
|
|
|
def test_unconfigured_returns_error_without_connecting(self) -> None:
|
|
# Manually create an object that bypasses the model_validator for testing
|
|
config = OpenClawConfig.__new__(OpenClawConfig)
|
|
object.__setattr__(config, "mode", "streamable-http")
|
|
object.__setattr__(config, "url", "")
|
|
object.__setattr__(config, "command", "")
|
|
object.__setattr__(config, "auth_token", "")
|
|
object.__setattr__(config, "args", ())
|
|
object.__setattr__(config, "headers", {})
|
|
object.__setattr__(config, "timeout_seconds", 15.0)
|
|
object.__setattr__(config, "integration_id", "")
|
|
|
|
result = validate_openclaw_config(config)
|
|
assert result.ok is False
|
|
assert "not configured" in result.detail.lower()
|
|
|
|
def test_stdio_endpoint_uses_command_in_detail(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp")
|
|
mock_tools = [{"name": "conversations_list"}]
|
|
|
|
with (
|
|
patch("integrations.openclaw.shutil.which", return_value="/tmp/openclaw-mcp"),
|
|
patch("integrations.openclaw.list_openclaw_tools", return_value=mock_tools),
|
|
):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is True
|
|
assert "openclaw-mcp" in result.detail
|
|
|
|
def test_stdio_missing_binary_fails_before_listing_tools(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw")
|
|
|
|
with (
|
|
patch("integrations.openclaw.shutil.which", return_value=None),
|
|
patch("integrations.openclaw.list_openclaw_tools") as mock_list_tools,
|
|
):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is False
|
|
assert "Command not found" in result.detail
|
|
mock_list_tools.assert_not_called()
|
|
|
|
def test_stdio_node_requirement_fails_before_listing_tools(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw", args=["mcp", "serve"])
|
|
completed = subprocess.CompletedProcess(
|
|
args=["openclaw", "--help"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="openclaw: Node.js v22.12+ is required (current: v22.1.0).",
|
|
)
|
|
|
|
with (
|
|
patch("integrations.openclaw.shutil.which", return_value="/opt/homebrew/bin/openclaw"),
|
|
patch("integrations.openclaw.subprocess.run", return_value=completed),
|
|
patch("integrations.openclaw.list_openclaw_tools") as mock_list_tools,
|
|
):
|
|
result = validate_openclaw_config(config)
|
|
|
|
assert result.ok is False
|
|
assert "Node.js v22.12+" in result.detail
|
|
mock_list_tools.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _tool_result_to_dict
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestToolResultToDict:
|
|
def _make_result(
|
|
self,
|
|
*,
|
|
is_error: bool = False,
|
|
content_items: list | None = None,
|
|
) -> MagicMock:
|
|
result = MagicMock()
|
|
result.isError = is_error
|
|
result.content = content_items or []
|
|
result.structuredContent = None
|
|
return result
|
|
|
|
def test_empty_content(self) -> None:
|
|
result = self._make_result()
|
|
parsed = _tool_result_to_dict(result)
|
|
assert parsed["is_error"] is False
|
|
assert parsed["text"] == ""
|
|
assert parsed["content"] == []
|
|
|
|
def test_text_content_extracted(self) -> None:
|
|
from mcp import types # type: ignore[import-not-found]
|
|
|
|
text_item = MagicMock(spec=types.TextContent)
|
|
text_item.text = "Hello from OpenClaw"
|
|
result = self._make_result(content_items=[text_item])
|
|
parsed = _tool_result_to_dict(result)
|
|
assert "Hello from OpenClaw" in parsed["text"]
|
|
assert parsed["content"][0]["type"] == "text"
|
|
|
|
def test_is_error_propagated(self) -> None:
|
|
result = self._make_result(is_error=True)
|
|
parsed = _tool_result_to_dict(result)
|
|
assert parsed["is_error"] is True
|
|
|
|
|
|
class TestDescribeOpenClawError:
|
|
def test_legacy_command_receives_bridge_hint(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw-mcp")
|
|
|
|
detail = describe_openclaw_error(
|
|
FileNotFoundError(2, "No such file", "openclaw-mcp"), config
|
|
)
|
|
|
|
assert "Command not found" in detail
|
|
assert "openclaw mcp serve" in detail
|
|
|
|
def test_gateway_unavailable_receives_gateway_hint(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw", args=["mcp", "serve"])
|
|
completed = subprocess.CompletedProcess(
|
|
args=["openclaw", "--help"],
|
|
returncode=0,
|
|
stdout="OpenClaw CLI help",
|
|
stderr="",
|
|
)
|
|
|
|
with patch("integrations.openclaw.subprocess.run", return_value=completed):
|
|
detail = describe_openclaw_error(RuntimeError("Connection closed"), config)
|
|
|
|
assert "openclaw gateway status" in detail
|
|
assert "openclaw gateway run" in detail
|
|
|
|
def test_runtime_unavailable_reason_detects_missing_stdio_command(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw")
|
|
|
|
with patch("integrations.openclaw.shutil.which", return_value=None):
|
|
detail = openclaw_runtime_unavailable_reason(config)
|
|
|
|
assert detail is not None
|
|
assert "Command not found" in detail
|
|
|
|
def test_runtime_unavailable_reason_detects_node_version_requirement(self) -> None:
|
|
config = OpenClawConfig(mode="stdio", command="openclaw", args=["mcp", "serve"])
|
|
completed = subprocess.CompletedProcess(
|
|
args=["openclaw", "--help"],
|
|
returncode=1,
|
|
stdout="",
|
|
stderr=(
|
|
"openclaw: Node.js v22.12+ is required (current: v22.1.0).\n"
|
|
"If you use nvm, run:\n"
|
|
" nvm install 22\n"
|
|
" nvm use 22\n"
|
|
" nvm alias default 22\n"
|
|
),
|
|
)
|
|
|
|
with (
|
|
patch("integrations.openclaw.shutil.which", return_value="/opt/homebrew/bin/openclaw"),
|
|
patch("integrations.openclaw.subprocess.run", return_value=completed),
|
|
):
|
|
detail = openclaw_runtime_unavailable_reason(config)
|
|
|
|
assert detail is not None
|
|
assert "Node.js v22.12+" in detail
|
|
assert "nvm use 22" in detail
|
|
assert "current shell" in detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_openclaw_tools (integration via mock)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListOpenClawTools:
|
|
def test_returns_tool_list(self) -> None:
|
|
config = OpenClawConfig(url="https://openclaw.example.com/mcp")
|
|
|
|
mock_tool = MagicMock()
|
|
mock_tool.name = "conversations_list"
|
|
mock_tool.description = "List conversations"
|
|
mock_tool.inputSchema = {"type": "object"}
|
|
|
|
with patch(
|
|
"integrations.openclaw._list_tools_async",
|
|
new=AsyncMock(return_value=[mock_tool]),
|
|
):
|
|
tools = list_openclaw_tools(config)
|
|
|
|
assert len(tools) == 1
|
|
assert tools[0]["name"] == "conversations_list"
|
|
assert tools[0]["description"] == "List conversations"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _classify_integrations (resolve_integrations node)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestClassifyOpenClawIntegration:
|
|
def _make_record(self, credentials: dict) -> dict:
|
|
return {
|
|
"id": "openclaw-1",
|
|
"service": "openclaw",
|
|
"status": "active",
|
|
"credentials": credentials,
|
|
}
|
|
|
|
def test_streamable_http_classified(self) -> None:
|
|
record = self._make_record(
|
|
{
|
|
"url": "https://openclaw.example.com/mcp",
|
|
"mode": "streamable-http",
|
|
"auth_token": "tok",
|
|
}
|
|
)
|
|
from tools.investigation.stages.gather_evidence.tools import availability_view
|
|
|
|
resolved = _classify_integrations([record])
|
|
assert "openclaw" in resolved
|
|
assert resolved["openclaw"].url == "https://openclaw.example.com/mcp"
|
|
# connection_verified is set at the tool-availability boundary
|
|
view = availability_view(resolved)
|
|
assert view["openclaw"]["connection_verified"] is True
|
|
|
|
def test_stdio_classified(self) -> None:
|
|
record = self._make_record({"mode": "stdio", "command": "openclaw-mcp"})
|
|
resolved = _classify_integrations([record])
|
|
assert "openclaw" in resolved
|
|
assert resolved["openclaw"].command == "openclaw-mcp"
|
|
|
|
def test_invalid_config_skipped(self) -> None:
|
|
# No url and non-stdio mode → ValidationError inside _classify_integrations → skipped
|
|
record = self._make_record({"mode": "streamable-http", "url": ""})
|
|
resolved = _classify_integrations([record])
|
|
assert "openclaw" not in resolved
|
|
|
|
def test_inactive_integration_skipped(self) -> None:
|
|
record = {
|
|
"id": "openclaw-2",
|
|
"service": "openclaw",
|
|
"status": "inactive",
|
|
"credentials": {"url": "https://openclaw.example.com/mcp"},
|
|
}
|
|
resolved = _classify_integrations([record])
|
|
assert "openclaw" not in resolved
|
|
|
|
def test_integration_id_preserved(self) -> None:
|
|
record = self._make_record(
|
|
{"url": "https://openclaw.example.com/mcp", "mode": "streamable-http"}
|
|
)
|
|
resolved = _classify_integrations([record])
|
|
assert resolved["openclaw"].integration_id == "openclaw-1"
|
|
|
|
|
|
class TestOpenClawExtractParams:
|
|
def test_extract_params_maps_plain_config_keys(self) -> None:
|
|
params = call_openclaw_bridge_tool.__opensre_registered_tool__.extract_params(
|
|
{
|
|
"openclaw": {
|
|
"connection_verified": True,
|
|
"url": "https://openclaw.example.com/mcp",
|
|
"mode": "streamable-http",
|
|
"auth_token": "tok",
|
|
"command": "openclaw",
|
|
"args": ["mcp", "serve"],
|
|
}
|
|
}
|
|
)
|
|
|
|
assert params["openclaw_url"] == "https://openclaw.example.com/mcp"
|
|
assert params["openclaw_mode"] == "streamable-http"
|
|
assert params["openclaw_token"] == "tok"
|
|
assert params["openclaw_command"] == "openclaw"
|
|
assert params["openclaw_args"] == ["mcp", "serve"]
|
|
|
|
def test_extract_params_keeps_prefixed_keys_for_detected_sources(self) -> None:
|
|
params = call_openclaw_bridge_tool.__opensre_registered_tool__.extract_params(
|
|
{
|
|
"openclaw": {
|
|
"connection_verified": True,
|
|
"openclaw_mode": "stdio",
|
|
"openclaw_command": "openclaw",
|
|
"openclaw_args": ["mcp", "serve"],
|
|
}
|
|
}
|
|
)
|
|
|
|
assert params["openclaw_mode"] == "stdio"
|
|
assert params["openclaw_command"] == "openclaw"
|
|
assert params["openclaw_args"] == ["mcp", "serve"]
|
|
|
|
def test_search_params_maps_search_query_aliases(self) -> None:
|
|
params = search_openclaw_conversations.__opensre_registered_tool__.extract_params(
|
|
{
|
|
"openclaw": {
|
|
"connection_verified": True,
|
|
"url": "https://openclaw.example.com/mcp",
|
|
"mode": "streamable-http",
|
|
"search_query": "checkout-api",
|
|
}
|
|
}
|
|
)
|
|
|
|
assert params["search"] == "checkout-api"
|
|
assert params["limit"] == 10
|