wehub-resource-sync
60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
583 lines
21 KiB
Python
583 lines
21 KiB
Python
"""Tests for fastmcp list and fastmcp call CLI commands."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import mcp_types
|
|
import pytest
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.cli import client as client_module
|
|
from fastmcp.cli.client import (
|
|
Client,
|
|
_build_client,
|
|
_build_stdio_from_command,
|
|
_format_call_result_text,
|
|
_is_http_target,
|
|
_sanitize_untrusted_text,
|
|
call_command,
|
|
coerce_value,
|
|
format_tool_signature,
|
|
list_command,
|
|
parse_tool_arguments,
|
|
resolve_server_spec,
|
|
)
|
|
from fastmcp.client.client import CallToolResult
|
|
from fastmcp.client.transports.stdio import StdioTransport
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# coerce_value
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCoerceValue:
|
|
def test_integer(self):
|
|
assert coerce_value("42", {"type": "integer"}) == 42
|
|
|
|
def test_integer_negative(self):
|
|
assert coerce_value("-7", {"type": "integer"}) == -7
|
|
|
|
def test_integer_invalid(self):
|
|
with pytest.raises(ValueError, match="Expected integer"):
|
|
coerce_value("abc", {"type": "integer"})
|
|
|
|
def test_number(self):
|
|
assert coerce_value("3.14", {"type": "number"}) == 3.14
|
|
|
|
def test_number_integer_value(self):
|
|
assert coerce_value("5", {"type": "number"}) == 5.0
|
|
|
|
def test_number_invalid(self):
|
|
with pytest.raises(ValueError, match="Expected number"):
|
|
coerce_value("xyz", {"type": "number"})
|
|
|
|
def test_boolean_true_variants(self):
|
|
for val in ("true", "True", "TRUE", "1", "yes"):
|
|
assert coerce_value(val, {"type": "boolean"}) is True
|
|
|
|
def test_boolean_false_variants(self):
|
|
for val in ("false", "False", "FALSE", "0", "no"):
|
|
assert coerce_value(val, {"type": "boolean"}) is False
|
|
|
|
def test_boolean_invalid(self):
|
|
with pytest.raises(ValueError, match="Expected boolean"):
|
|
coerce_value("maybe", {"type": "boolean"})
|
|
|
|
def test_array(self):
|
|
assert coerce_value("[1, 2, 3]", {"type": "array"}) == [1, 2, 3]
|
|
|
|
def test_array_invalid(self):
|
|
with pytest.raises(ValueError, match="Expected JSON array"):
|
|
coerce_value("not-json", {"type": "array"})
|
|
|
|
def test_object(self):
|
|
assert coerce_value('{"a": 1}', {"type": "object"}) == {"a": 1}
|
|
|
|
def test_string(self):
|
|
assert coerce_value("hello", {"type": "string"}) == "hello"
|
|
|
|
def test_string_default(self):
|
|
"""Unknown or missing type treats value as string."""
|
|
assert coerce_value("hello", {}) == "hello"
|
|
|
|
def test_string_preserves_numeric_looking_values(self):
|
|
assert coerce_value("42", {"type": "string"}) == "42"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_tool_arguments
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestParseToolArguments:
|
|
SCHEMA: dict[str, Any] = {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer"},
|
|
"verbose": {"type": "boolean"},
|
|
},
|
|
"required": ["query"],
|
|
}
|
|
|
|
def test_basic_key_value(self):
|
|
result = parse_tool_arguments(("query=hello", "limit=10"), None, self.SCHEMA)
|
|
assert result == {"query": "hello", "limit": 10}
|
|
|
|
def test_input_json_only(self):
|
|
result = parse_tool_arguments((), '{"query": "hello", "limit": 5}', self.SCHEMA)
|
|
assert result == {"query": "hello", "limit": 5}
|
|
|
|
def test_key_value_overrides_input_json(self):
|
|
result = parse_tool_arguments(
|
|
("limit=20",), '{"query": "hello", "limit": 5}', self.SCHEMA
|
|
)
|
|
assert result == {"query": "hello", "limit": 20}
|
|
|
|
def test_value_containing_equals(self):
|
|
result = parse_tool_arguments(("query=a=b=c",), None, self.SCHEMA)
|
|
assert result == {"query": "a=b=c"}
|
|
|
|
def test_invalid_arg_format_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
parse_tool_arguments(("noequalssign",), None, self.SCHEMA)
|
|
|
|
def test_invalid_input_json_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
parse_tool_arguments((), "not-valid-json", self.SCHEMA)
|
|
|
|
def test_input_json_non_object_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
parse_tool_arguments((), "[1,2,3]", self.SCHEMA)
|
|
|
|
def test_single_json_object_as_positional(self):
|
|
result = parse_tool_arguments(
|
|
('{"query": "hello", "limit": 5}',), None, self.SCHEMA
|
|
)
|
|
assert result == {"query": "hello", "limit": 5}
|
|
|
|
def test_json_positional_ignored_when_input_json_set(self):
|
|
"""When --input-json is already provided, a JSON positional arg is not special."""
|
|
with pytest.raises(SystemExit):
|
|
parse_tool_arguments(('{"limit": 99}',), '{"query": "hello"}', self.SCHEMA)
|
|
|
|
def test_coercion_error_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
parse_tool_arguments(("limit=abc",), None, self.SCHEMA)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# format_tool_signature
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFormatToolSignature:
|
|
def _make_tool(
|
|
self,
|
|
name: str = "my_tool",
|
|
properties: dict[str, Any] | None = None,
|
|
required: list[str] | None = None,
|
|
output_schema: dict[str, Any] | None = None,
|
|
description: str | None = None,
|
|
) -> mcp_types.Tool:
|
|
input_schema: dict[str, Any] = {"type": "object"}
|
|
if properties is not None:
|
|
input_schema["properties"] = properties
|
|
if required is not None:
|
|
input_schema["required"] = required
|
|
return mcp_types.Tool(
|
|
name=name,
|
|
description=description,
|
|
input_schema=input_schema,
|
|
output_schema=output_schema,
|
|
)
|
|
|
|
def test_no_params(self):
|
|
tool = self._make_tool()
|
|
assert format_tool_signature(tool) == "my_tool()"
|
|
|
|
def test_required_param(self):
|
|
tool = self._make_tool(
|
|
properties={"query": {"type": "string"}},
|
|
required=["query"],
|
|
)
|
|
assert format_tool_signature(tool) == "my_tool(query: str)"
|
|
|
|
def test_optional_param_with_default(self):
|
|
tool = self._make_tool(
|
|
properties={"limit": {"type": "integer", "default": 10}},
|
|
)
|
|
assert format_tool_signature(tool) == "my_tool(limit: int = 10)"
|
|
|
|
def test_optional_param_without_default(self):
|
|
tool = self._make_tool(
|
|
properties={"limit": {"type": "integer"}},
|
|
)
|
|
assert format_tool_signature(tool) == "my_tool(limit: int = ...)"
|
|
|
|
def test_mixed_required_and_optional(self):
|
|
tool = self._make_tool(
|
|
properties={
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer", "default": 10},
|
|
},
|
|
required=["query"],
|
|
)
|
|
sig = format_tool_signature(tool)
|
|
assert sig == "my_tool(query: str, limit: int = 10)"
|
|
|
|
def test_with_output_schema(self):
|
|
tool = self._make_tool(
|
|
properties={"q": {"type": "string"}},
|
|
required=["q"],
|
|
output_schema={"type": "object"},
|
|
)
|
|
assert format_tool_signature(tool) == "my_tool(q: str) -> dict"
|
|
|
|
def test_anyof_type(self):
|
|
tool = self._make_tool(
|
|
properties={"value": {"anyOf": [{"type": "string"}, {"type": "integer"}]}},
|
|
required=["value"],
|
|
)
|
|
assert format_tool_signature(tool) == "my_tool(value: str | int)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_server_spec
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResolveServerSpec:
|
|
def test_http_url(self):
|
|
assert (
|
|
resolve_server_spec("http://localhost:8000/mcp")
|
|
== "http://localhost:8000/mcp"
|
|
)
|
|
|
|
def test_https_url(self):
|
|
assert (
|
|
resolve_server_spec("https://example.com/mcp") == "https://example.com/mcp"
|
|
)
|
|
|
|
def test_python_file_existing(self, tmp_path: Path):
|
|
py_file = tmp_path / "server.py"
|
|
py_file.write_text("# empty")
|
|
result = resolve_server_spec(str(py_file))
|
|
assert isinstance(result, StdioTransport)
|
|
assert result.command == "fastmcp"
|
|
assert result.args == ["run", str(py_file.resolve()), "--no-banner"]
|
|
|
|
def test_json_mcp_config(self, tmp_path: Path):
|
|
config_file = tmp_path / "mcp.json"
|
|
config = {"mcpServers": {"test": {"url": "http://localhost:8000"}}}
|
|
config_file.write_text(json.dumps(config))
|
|
result = resolve_server_spec(str(config_file))
|
|
assert isinstance(result, dict)
|
|
assert "mcpServers" in result
|
|
|
|
def test_json_fastmcp_config_exits(self, tmp_path: Path):
|
|
config_file = tmp_path / "fastmcp.json"
|
|
config_file.write_text(json.dumps({"source": {"type": "file"}}))
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec(str(config_file))
|
|
|
|
def test_json_not_found_exits(self, tmp_path: Path):
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec(str(tmp_path / "nonexistent.json"))
|
|
|
|
def test_directory_exits(self, tmp_path: Path):
|
|
"""Directories should not be treated as file paths."""
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec(str(tmp_path))
|
|
|
|
def test_unrecognised_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec("some_random_thing")
|
|
|
|
def test_command_returns_stdio_transport(self):
|
|
result = resolve_server_spec(None, command="npx -y @mcp/server")
|
|
assert isinstance(result, StdioTransport)
|
|
assert result.command == "npx"
|
|
assert result.args == ["-y", "@mcp/server"]
|
|
|
|
def test_command_single_word(self):
|
|
result = resolve_server_spec(None, command="myserver")
|
|
assert isinstance(result, StdioTransport)
|
|
assert result.command == "myserver"
|
|
assert result.args == []
|
|
|
|
def test_server_spec_and_command_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec("http://localhost:8000", command="npx server")
|
|
|
|
def test_neither_server_spec_nor_command_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
resolve_server_spec(None)
|
|
|
|
def test_transport_sse_rewrites_url(self):
|
|
result = resolve_server_spec("http://localhost:8000/mcp", transport="sse")
|
|
assert result == "http://localhost:8000/mcp/sse"
|
|
|
|
def test_transport_sse_no_duplicate_suffix(self):
|
|
result = resolve_server_spec("http://localhost:8000/sse", transport="sse")
|
|
assert result == "http://localhost:8000/sse"
|
|
|
|
def test_transport_sse_trailing_slash(self):
|
|
result = resolve_server_spec("http://localhost:8000/mcp/", transport="sse")
|
|
assert result == "http://localhost:8000/mcp/sse"
|
|
|
|
def test_transport_http_leaves_url_unchanged(self):
|
|
result = resolve_server_spec("http://localhost:8000/mcp", transport="http")
|
|
assert result == "http://localhost:8000/mcp"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_stdio_from_command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBuildStdioFromCommand:
|
|
def test_simple_command(self):
|
|
transport = _build_stdio_from_command("uvx my-server")
|
|
assert transport.command == "uvx"
|
|
assert transport.args == ["my-server"]
|
|
|
|
def test_quoted_args(self):
|
|
transport = _build_stdio_from_command("npx -y '@scope/server'")
|
|
assert transport.command == "npx"
|
|
assert transport.args == ["-y", "@scope/server"]
|
|
|
|
def test_empty_command_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
_build_stdio_from_command("")
|
|
|
|
def test_invalid_shell_syntax_exits(self):
|
|
with pytest.raises(SystemExit):
|
|
_build_stdio_from_command("npx 'unterminated")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _is_http_target
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIsHttpTarget:
|
|
def test_http_url(self):
|
|
assert _is_http_target("http://localhost:8000") is True
|
|
|
|
def test_https_url(self):
|
|
assert _is_http_target("https://example.com/mcp") is True
|
|
|
|
def test_file_path(self):
|
|
assert _is_http_target("/path/to/server.py") is False
|
|
|
|
def test_stdio_transport(self):
|
|
assert _is_http_target(StdioTransport(command="npx", args=[])) is False
|
|
|
|
def test_mcp_config_dict(self):
|
|
"""MCPConfig dicts are not HTTP targets — auth is per-server internally."""
|
|
assert _is_http_target({"mcpServers": {}}) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBuildClient:
|
|
def test_http_target_gets_oauth_by_default(self):
|
|
client = _build_client("http://localhost:8000/mcp")
|
|
# OAuth is applied during Client init via _set_auth
|
|
assert client.transport.auth is not None
|
|
|
|
def test_stdio_target_no_auth(self):
|
|
transport = StdioTransport(command="npx", args=["-y", "@mcp/server"])
|
|
client = _build_client(transport)
|
|
# Stdio transports don't support auth — no auth should be set
|
|
assert not hasattr(client.transport, "auth") or client.transport.auth is None
|
|
|
|
def test_explicit_auth_none_disables_oauth(self):
|
|
client = _build_client("http://localhost:8000/mcp", auth="none")
|
|
# "none" explicitly disables auth, even for HTTP targets
|
|
assert client.transport.auth is None
|
|
|
|
def test_mcp_config_no_auth(self):
|
|
"""MCPConfig dicts handle auth per-server; no top-level auth applied."""
|
|
client = _build_client({"mcpServers": {"test": {"url": "http://localhost"}}})
|
|
# MCPConfigTransport doesn't support _set_auth — no crash means success
|
|
assert client.transport is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Integration tests — invoke actual CLI commands via monkeypatched _build_client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _build_test_server() -> FastMCP:
|
|
"""Create a minimal FastMCP server for integration tests."""
|
|
server = FastMCP("TestServer")
|
|
|
|
@server.tool
|
|
def greet(name: str) -> str:
|
|
"""Say hello to someone."""
|
|
return f"Hello, {name}!"
|
|
|
|
@server.tool
|
|
def add(a: int, b: int) -> int:
|
|
"""Add two numbers."""
|
|
return a + b
|
|
|
|
@server.resource("test://greeting")
|
|
def greeting_resource() -> str:
|
|
"""A static greeting resource."""
|
|
return "Hello from resource!"
|
|
|
|
@server.prompt
|
|
def ask(topic: str) -> str:
|
|
"""Ask about a topic."""
|
|
return f"Tell me about {topic}"
|
|
|
|
return server
|
|
|
|
|
|
@pytest.fixture()
|
|
def _patch_client():
|
|
"""Patch resolve_server_spec and _build_client so CLI commands use the
|
|
in-process test server without needing a real transport."""
|
|
server = _build_test_server()
|
|
|
|
def fake_resolve(server_spec: Any, **kwargs: Any) -> str:
|
|
return "fake"
|
|
|
|
def fake_build_client(resolved: Any, **kwargs: Any) -> Client:
|
|
return Client(server)
|
|
|
|
with (
|
|
patch.object(client_module, "resolve_server_spec", side_effect=fake_resolve),
|
|
patch.object(client_module, "_build_client", side_effect=fake_build_client),
|
|
):
|
|
yield
|
|
|
|
|
|
class TestListCommandCLI:
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_list_tools(self, capsys: pytest.CaptureFixture[str]):
|
|
await list_command("fake://server")
|
|
captured = capsys.readouterr()
|
|
assert "greet" in captured.out
|
|
assert "add" in captured.out
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_list_json(self, capsys: pytest.CaptureFixture[str]):
|
|
await list_command("fake://server", json_output=True)
|
|
captured = capsys.readouterr()
|
|
data = json.loads(captured.out)
|
|
names = {t["name"] for t in data["tools"]}
|
|
assert "greet" in names
|
|
assert "add" in names
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_list_resources(self, capsys: pytest.CaptureFixture[str]):
|
|
await list_command("fake://server", resources=True)
|
|
captured = capsys.readouterr()
|
|
assert "test://greeting" in captured.out
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_list_prompts(self, capsys: pytest.CaptureFixture[str]):
|
|
await list_command("fake://server", prompts=True)
|
|
captured = capsys.readouterr()
|
|
assert "ask" in captured.out
|
|
|
|
|
|
class TestCallCommandCLI:
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_tool(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command("fake://server", "greet", "name=World")
|
|
captured = capsys.readouterr()
|
|
assert "Hello, World!" in captured.out
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_tool_json(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command("fake://server", "greet", "name=World", json_output=True)
|
|
captured = capsys.readouterr()
|
|
data = json.loads(captured.out)
|
|
assert data["is_error"] is False
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_tool_not_found(self):
|
|
with pytest.raises(SystemExit):
|
|
await call_command("fake://server", "nonexistent")
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_tool_missing_args(self):
|
|
with pytest.raises(SystemExit):
|
|
await call_command("fake://server", "greet")
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_resource_by_uri(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command("fake://server", "test://greeting")
|
|
captured = capsys.readouterr()
|
|
assert "Hello from resource!" in captured.out
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_resource_json(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command("fake://server", "test://greeting", json_output=True)
|
|
captured = capsys.readouterr()
|
|
data = json.loads(captured.out)
|
|
assert isinstance(data, list)
|
|
assert data[0]["text"] == "Hello from resource!"
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_prompt(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command("fake://server", "ask", "topic=Python", prompt=True)
|
|
captured = capsys.readouterr()
|
|
assert "Python" in captured.out
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_prompt_json(self, capsys: pytest.CaptureFixture[str]):
|
|
await call_command(
|
|
"fake://server", "ask", "topic=Python", prompt=True, json_output=True
|
|
)
|
|
captured = capsys.readouterr()
|
|
data = json.loads(captured.out)
|
|
assert "messages" in data
|
|
|
|
@pytest.mark.usefixtures("_patch_client")
|
|
async def test_call_prompt_not_found(self):
|
|
with pytest.raises(SystemExit):
|
|
await call_command("fake://server", "nonexistent", prompt=True)
|
|
|
|
async def test_call_missing_target(self):
|
|
with pytest.raises(SystemExit):
|
|
await call_command("fake://server", "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Structured content serialization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFormatCallResult:
|
|
def test_structured_content_uses_dict_not_data(
|
|
self, capsys: pytest.CaptureFixture[str]
|
|
):
|
|
"""structured_content (raw dict) is used for display, not data (which may
|
|
be a non-serializable dataclass)."""
|
|
result = CallToolResult(
|
|
content=[mcp_types.TextContent(type="text", text="ok")],
|
|
structured_content={"key": "value"},
|
|
meta=None,
|
|
data=object(), # non-serializable on purpose
|
|
is_error=False,
|
|
)
|
|
# Should not raise — uses structured_content, not data
|
|
_format_call_result_text(result)
|
|
captured = capsys.readouterr()
|
|
assert "value" in captured.out
|
|
|
|
def test_escapes_rich_markup_and_control_chars(
|
|
self, capsys: pytest.CaptureFixture[str]
|
|
):
|
|
result = CallToolResult(
|
|
content=[mcp_types.TextContent(type="text", text="[red]x[/red]\x1b[2J")],
|
|
structured_content=None,
|
|
meta=None,
|
|
data=None,
|
|
is_error=False,
|
|
)
|
|
|
|
_format_call_result_text(result)
|
|
captured = capsys.readouterr()
|
|
assert "[red]x[/red]" in captured.out
|
|
assert "\\x1b" in captured.out
|
|
assert "\x1b" not in captured.out
|
|
|
|
|
|
class TestSanitizeUntrustedText:
|
|
def test_sanitize_untrusted_text(self):
|
|
value = "[bold]hello[/bold]\x07"
|
|
sanitized = _sanitize_untrusted_text(value)
|
|
assert sanitized == "\\[bold]hello\\[/bold]\\x07"
|