Files
obsidian-wiki/tests/test_mcp_service.py
T
Codex e7913d4c6b feat(mcp): add OpenCode-backed wiki query
Add a query tool group for the wiki-file-assets MCP server that queues fixed /wiki-query OpenCode jobs, exposes status/result/cancel/readiness tools, and keeps query disabled until explicitly enabled.

Document the new service switches and config, keep remote query gated in AGENTS.md, preserve token behavior, and pass proxy environment through the managed service so OpenCode smoke/query jobs can reach the model provider.

Validation: python3 -m unittest discover -v; python3 -m py_compile obsidian_wiki/mcp_query.py obsidian_wiki/mcp_server.py scripts/manage_mcp_service.py tests/test_mcp_query.py tests/test_mcp_server_remote.py tests/test_mcp_service.py; git diff --check.
2026-07-05 18:00:53 +00:00

505 lines
25 KiB
Python

import os
import stat
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import call, patch
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
import manage_mcp_service
class McpServiceConfigTests(unittest.TestCase):
def test_service_commands_use_one_shell_entrypoint(self):
entrypoint = SCRIPTS_DIR / "mcp_service.sh"
legacy_wrappers = [
"new_mcp_service_token.sh",
"start_mcp_service.sh",
"show_mcp_service_config.sh",
"stop_mcp_service.sh",
"uninstall_mcp_service.sh",
]
self.assertTrue(entrypoint.is_file())
self.assertTrue(os.access(entrypoint, os.X_OK))
content = entrypoint.read_text(encoding="utf-8")
self.assertIn('exec python3 "$SCRIPT_DIR/manage_mcp_service.py" "$@"', content)
for wrapper in legacy_wrappers:
self.assertFalse((SCRIPTS_DIR / wrapper).exists(), f"{wrapper} should be folded into mcp_service.sh")
def test_render_service_unit_uses_http_transport_without_embedding_token(self):
unit = manage_mcp_service.render_service_unit(
project_root=Path("/opt/obsidian-wiki"),
vault_path=Path("/data/bjd-vault"),
env_file=Path("/opt/obsidian-wiki/.runtime/obsidian-wiki-mcp.env"),
uv_path=Path("/home/nac/.local/bin/uv"),
)
self.assertIn("WorkingDirectory=/opt/obsidian-wiki", unit)
self.assertIn("EnvironmentFile=/opt/obsidian-wiki/.runtime/obsidian-wiki-mcp.env", unit)
self.assertIn("/home/nac/.local/bin/uv run --no-sync python -m obsidian_wiki.mcp_server", unit)
self.assertIn("--vault /data/bjd-vault", unit)
self.assertIn("--transport streamable-http", unit)
self.assertIn("--host 0.0.0.0", unit)
self.assertIn("--port 8765", unit)
self.assertIn("PassEnvironment=http_proxy https_proxy all_proxy no_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY", unit)
self.assertNotIn("OBSIDIAN_MCP_TOKEN=", unit)
def test_write_token_env_file_is_private(self):
with tempfile.TemporaryDirectory() as tmp:
env_path = Path(tmp) / ".runtime" / "obsidian-wiki-mcp.env"
manage_mcp_service.write_token_env_file(
env_path,
"test-token",
host="0.0.0.0",
port=8765,
path="/mcp",
enable_ingest=True,
)
content = env_path.read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_TOKEN=test-token\n", content)
self.assertIn("OBSIDIAN_MCP_HOST=0.0.0.0\n", content)
self.assertIn("OBSIDIAN_MCP_PORT=8765\n", content)
self.assertIn("OBSIDIAN_MCP_PATH=/mcp\n", content)
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", content)
self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600)
self.assertEqual(stat.S_IMODE(os.stat(env_path.parent).st_mode), 0o700)
def test_resolve_vault_for_service_uses_explicit_or_configured_vault(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
explicit_vault = Path(tmp) / "explicit-vault"
configured_vault = Path(tmp) / "configured-vault"
project_root.mkdir()
explicit_vault.mkdir()
configured_vault.mkdir()
self.assertEqual(
manage_mcp_service.resolve_vault_for_service(project_root, vault_path=explicit_vault),
explicit_vault.resolve(),
)
with patch.object(manage_mcp_service, "resolve_vault_path", return_value=str(configured_vault)):
self.assertEqual(
manage_mcp_service.resolve_vault_for_service(project_root),
configured_vault.resolve(),
)
def test_render_remote_config_prints_client_configuration(self):
output = manage_mcp_service.render_remote_config(
hosts=["192.168.1.10"],
token="test-token",
port=8765,
path="/mcp",
)
self.assertIn("http://192.168.1.10:8765/mcp", output)
self.assertIn("Authorization", output)
self.assertIn("Bearer test-token", output)
self.assertIn('"type": "streamable-http"', output)
self.assertIn('"wiki-file-assets"', output)
self.assertNotIn('"obsidian-wiki"', output)
def test_render_remote_config_prints_ingest_upload_endpoint_when_enabled(self):
output = manage_mcp_service.render_remote_config(
hosts=["192.168.1.10", "10.0.0.2"],
token="test-token",
port=8765,
path="/mcp",
enabled_tools=("file-assets", "ingest"),
)
self.assertIn("http://192.168.1.10:8765/mcp", output)
self.assertIn("http://192.168.1.10:8765/mcp-ingest/uploads", output)
self.assertIn("http://10.0.0.2:8765/mcp-ingest/uploads", output)
def test_new_token_writes_token_without_touching_systemd_when_unit_missing(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="new-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.new_token(project_root=project_root)
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
env_content = env_path.read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_TOKEN=new-token\n", env_content)
self.assertIn("OBSIDIAN_MCP_PORT=8765\n", env_content)
self.assertIn("http://10.0.0.2:8765/mcp", output)
run_systemctl.assert_not_called()
def test_new_token_restarts_service_when_unit_exists(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
paths = manage_mcp_service.resolve_service_paths(project_root, config_home=config_home)
paths.unit_path.parent.mkdir(parents=True)
paths.unit_path.write_text("unit\n", encoding="utf-8")
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="new-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.new_token(project_root=project_root)
env_content = paths.env_path.read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_TOKEN=new-token\n", env_content)
self.assertIn("http://10.0.0.2:8765/mcp", output)
run_systemctl.assert_called_once_with(["restart", manage_mcp_service.SERVICE_NAME])
def test_start_service_uses_saved_token_and_restarts_service(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
manage_mcp_service.write_token_env_file(
project_root / ".runtime" / "obsidian-wiki-mcp.env",
"saved-token",
host="0.0.0.0",
port=8765,
path="/mcp",
)
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token") as generate_token,
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
patch.object(manage_mcp_service, "resolve_vault_for_service", return_value=Path("/data/bjd-vault")),
patch.object(manage_mcp_service, "prepare_runtime") as prepare_runtime,
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.start_service(project_root=project_root)
unit_path = config_home / "systemd" / "user" / manage_mcp_service.SERVICE_NAME
unit_content = unit_path.read_text(encoding="utf-8")
self.assertIn("/home/nac/.local/bin/uv run --no-sync python -m obsidian_wiki.mcp_server", unit_content)
self.assertIn("--vault /data/bjd-vault", unit_content)
self.assertIn("EnvironmentFile=", unit_content)
self.assertIn("http://10.0.0.2:8765/mcp", output)
self.assertIn("Bearer saved-token", output)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets\n", env_content)
generate_token.assert_not_called()
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
run_systemctl.assert_has_calls(
[
call(["daemon-reload"]),
call(["enable", manage_mcp_service.SERVICE_NAME]),
call(["restart", manage_mcp_service.SERVICE_NAME]),
]
)
def test_start_service_generates_token_when_missing_and_prints_config(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="created-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.4"]),
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
patch.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"),
patch.object(manage_mcp_service, "prepare_runtime") as prepare_runtime,
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.start_service(
project_root=project_root,
host="0.0.0.0",
port=9876,
path="/wiki-mcp",
enable_ingest=True,
)
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
env_content = env_path.read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_TOKEN=created-token\n", env_content)
self.assertIn("OBSIDIAN_MCP_PORT=9876\n", env_content)
self.assertIn("OBSIDIAN_MCP_PATH=/wiki-mcp\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\n", env_content)
self.assertIn("http://10.0.0.4:9876/wiki-mcp", output)
self.assertIn("Authorization: Bearer created-token", output)
self.assertIn('"type": "streamable-http"', output)
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
run_systemctl.assert_has_calls(
[
call(["daemon-reload"]),
call(["enable", manage_mcp_service.SERVICE_NAME]),
call(["restart", manage_mcp_service.SERVICE_NAME]),
]
)
def test_start_service_preserves_saved_enabled_tools_when_flag_is_omitted(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
manage_mcp_service.write_token_env_file(
project_root / ".runtime" / "obsidian-wiki-mcp.env",
"saved-token",
host="0.0.0.0",
port=8765,
path="/mcp",
enabled_tools=("file-assets", "ingest"),
)
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
env_path.write_text(
env_path.read_text(encoding="utf-8")
+ "OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/custom/bin/opencode\n",
encoding="utf-8",
)
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
patch.object(manage_mcp_service, "resolve_vault_for_service", return_value=Path("/data/bjd-vault")),
patch.object(manage_mcp_service, "prepare_runtime"),
patch.object(manage_mcp_service, "run_systemctl"),
):
manage_mcp_service.start_service(project_root=project_root)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/custom/bin/opencode\n", env_content)
def test_start_service_can_enable_query_without_ingest(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="created-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.4"]),
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
patch.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"),
patch.object(manage_mcp_service, "prepare_runtime") as prepare_runtime,
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.start_service(
project_root=project_root,
enable_query=True,
)
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
env_content = env_path.read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,query\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content)
self.assertIn("OBSIDIAN_MCP_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\n", env_content)
self.assertNotIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=", env_content)
self.assertIn("Enabled tool groups:\nfile-assets,query", output)
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
run_systemctl.assert_has_calls(
[
call(["daemon-reload"]),
call(["enable", manage_mcp_service.SERVICE_NAME]),
call(["restart", manage_mcp_service.SERVICE_NAME]),
]
)
def test_main_tools_enable_disable_updates_enabled_tools_in_config(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="created-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch("builtins.print") as print_fn,
):
enable_status = manage_mcp_service.main(
["tools", "enable", "ingest", "--project-root", str(project_root)]
)
self.assertEqual(enable_status, 0)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
self.assertIn("ingest: enabled", print_fn.call_args.args[0])
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch("builtins.print") as print_fn,
):
disable_status = manage_mcp_service.main(
["tools", "disable", "ingest", "--project-root", str(project_root)]
)
self.assertEqual(disable_status, 0)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content)
self.assertIn("ingest: disabled", print_fn.call_args.args[0])
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch("builtins.print") as print_fn,
):
disable_files_status = manage_mcp_service.main(
["tools", "disable", "file-assets", "--project-root", str(project_root)]
)
self.assertEqual(disable_files_status, 0)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=\n", env_content)
self.assertIn("file-assets: disabled", print_fn.call_args.args[0])
def test_main_tools_enable_query_updates_enabled_tools_in_config(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "generate_token", return_value="created-token"),
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
patch.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"),
patch("builtins.print") as print_fn,
):
enable_status = manage_mcp_service.main(
["tools", "enable", "query", "--project-root", str(project_root)]
)
self.assertEqual(enable_status, 0)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,query\n", env_content)
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content)
self.assertIn("OBSIDIAN_MCP_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\n", env_content)
self.assertIn("query: enabled", print_fn.call_args.args[0])
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch("builtins.print") as print_fn,
):
disable_status = manage_mcp_service.main(
["tools", "disable", "query", "--project-root", str(project_root)]
)
self.assertEqual(disable_status, 0)
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets\n", env_content)
self.assertIn("query: disabled", print_fn.call_args.args[0])
def test_set_tool_enabled_imports_proxy_environment_when_restarting_query_service(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
project_root.mkdir()
config_home = Path(tmp) / "config"
paths = manage_mcp_service.resolve_service_paths(project_root, config_home=config_home)
paths.unit_path.parent.mkdir(parents=True)
paths.unit_path.write_text("unit\n", encoding="utf-8")
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home), "https_proxy": "http://proxy.local:8080"}, clear=True),
patch.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"),
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
output = manage_mcp_service.set_tool_enabled(project_root, "query", True)
self.assertIn("query: enabled", output)
run_systemctl.assert_has_calls(
[
call(["import-environment", "https_proxy"], check=False),
call(["restart", manage_mcp_service.SERVICE_NAME]),
]
)
def test_main_start_passes_enable_ingest_to_start_service(self):
with (
patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service,
patch("builtins.print") as print_fn,
):
status = manage_mcp_service.main(["start", "--enable-ingest"])
self.assertEqual(status, 0)
start_service.assert_called_once()
self.assertTrue(start_service.call_args.kwargs["enable_ingest"])
print_fn.assert_called_once_with("config\n", end="")
def test_main_start_passes_enable_query_to_start_service(self):
with (
patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service,
patch("builtins.print") as print_fn,
):
status = manage_mcp_service.main(["start", "--enable-query"])
self.assertEqual(status, 0)
start_service.assert_called_once()
self.assertTrue(start_service.call_args.kwargs["enable_query"])
print_fn.assert_called_once_with("config\n", end="")
def test_main_new_token_does_not_expect_enable_ingest_argument(self):
with (
patch.object(manage_mcp_service, "new_token", return_value="config\n") as new_token,
patch("builtins.print") as print_fn,
):
status = manage_mcp_service.main(["new-token"])
self.assertEqual(status, 0)
new_token.assert_called_once()
self.assertNotIn("enable_ingest", new_token.call_args.kwargs)
print_fn.assert_called_once_with("config\n", end="")
def test_uninstall_service_removes_only_managed_files(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
config_home = Path(tmp) / "config"
paths = manage_mcp_service.resolve_service_paths(project_root, config_home=config_home)
paths.unit_path.parent.mkdir(parents=True)
paths.env_path.parent.mkdir(parents=True)
paths.unit_path.write_text("unit\n", encoding="utf-8")
paths.env_path.write_text("OBSIDIAN_MCP_TOKEN=old\n", encoding="utf-8")
with (
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
):
manage_mcp_service.uninstall_service(project_root)
self.assertFalse(paths.unit_path.exists())
self.assertFalse(paths.env_path.exists())
run_systemctl.assert_has_calls(
[
call(["disable", "--now", manage_mcp_service.SERVICE_NAME], check=False),
call(["daemon-reload"], check=False),
call(["reset-failed", manage_mcp_service.SERVICE_NAME], check=False),
]
)
def test_show_config_reads_saved_token_and_prints_remote_config(self):
with tempfile.TemporaryDirectory() as tmp:
project_root = Path(tmp) / "project"
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
manage_mcp_service.write_token_env_file(env_path, "saved-token", host="0.0.0.0", port=9876, path="/wiki-mcp")
with patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.3"]):
output = manage_mcp_service.show_config(project_root)
self.assertIn("http://10.0.0.3:9876/wiki-mcp", output)
self.assertIn("Authorization: Bearer saved-token", output)
if __name__ == "__main__":
unittest.main()