wehub-resource-sync
249 lines
9.1 KiB
Python
249 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Run local system scenarios for MCP, plugins, bridge, and slash commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
from openharness.commands.registry import CommandContext, create_default_command_registry
|
|
from openharness.config.settings import Settings, load_settings
|
|
from openharness.engine.messages import ConversationMessage, TextBlock
|
|
from openharness.engine.query_engine import QueryEngine
|
|
from openharness.mcp.client import McpClientManager
|
|
from openharness.mcp.config import load_mcp_server_configs
|
|
from openharness.mcp.types import McpStdioServerConfig
|
|
from openharness.permissions import PermissionChecker
|
|
from openharness.plugins import load_plugins
|
|
from openharness.plugins.installer import install_plugin_from_path, uninstall_plugin
|
|
from openharness.state import AppState, AppStateStore
|
|
from openharness.bridge import build_sdk_url, decode_work_secret, encode_work_secret, spawn_session
|
|
from openharness.bridge.types import WorkSecret
|
|
from openharness.tools import create_default_tool_registry
|
|
from openharness.tools.base import ToolExecutionContext
|
|
|
|
|
|
FIXTURE_SERVER = Path(__file__).resolve().parents[1] / "tests" / "fixtures" / "fake_mcp_server.py"
|
|
|
|
|
|
class FakeApiClient:
|
|
async def stream_message(self, request):
|
|
del request
|
|
raise AssertionError("No model call expected in local system scenarios")
|
|
|
|
|
|
def _make_command_context(cwd: Path) -> CommandContext:
|
|
tool_registry = create_default_tool_registry()
|
|
engine = QueryEngine(
|
|
api_client=FakeApiClient(),
|
|
tool_registry=tool_registry,
|
|
permission_checker=PermissionChecker(load_settings().permission),
|
|
cwd=cwd,
|
|
model="claude-test",
|
|
system_prompt="system",
|
|
)
|
|
engine.load_messages(
|
|
[
|
|
ConversationMessage(role="user", content=[TextBlock(text="one")]),
|
|
ConversationMessage(role="assistant", content=[TextBlock(text="two")]),
|
|
ConversationMessage(role="user", content=[TextBlock(text="three")]),
|
|
]
|
|
)
|
|
return CommandContext(
|
|
engine=engine,
|
|
cwd=str(cwd),
|
|
tool_registry=tool_registry,
|
|
app_state=AppStateStore(
|
|
AppState(model="claude-test", permission_mode="default", theme="default", keybindings={})
|
|
),
|
|
)
|
|
|
|
|
|
def _write_plugin(source_root: Path) -> Path:
|
|
plugin_dir = source_root / "fixture-plugin"
|
|
(plugin_dir / "skills").mkdir(parents=True)
|
|
(plugin_dir / "plugin.json").write_text(
|
|
json.dumps({"name": "fixture-plugin", "version": "1.0.0", "description": "Fixture plugin"}),
|
|
encoding="utf-8",
|
|
)
|
|
(plugin_dir / "skills" / "fixture.md").write_text(
|
|
"# FixtureSkill\nFixture skill content for local scenario.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(plugin_dir / "mcp.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {
|
|
"fixture": {
|
|
"type": "stdio",
|
|
"command": sys.executable,
|
|
"args": [str(FIXTURE_SERVER)],
|
|
}
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return plugin_dir
|
|
|
|
|
|
async def _run_mcp_flow(temp_root: Path) -> None:
|
|
manager = McpClientManager(
|
|
{"fixture": McpStdioServerConfig(command=sys.executable, args=[str(FIXTURE_SERVER)])}
|
|
)
|
|
await manager.connect_all()
|
|
try:
|
|
registry = create_default_tool_registry(manager)
|
|
tool = registry.get("mcp__fixture__hello")
|
|
result = await tool.execute(
|
|
tool.input_model.model_validate({"name": "system"}),
|
|
ToolExecutionContext(cwd=temp_root),
|
|
)
|
|
if result.output != "fixture-hello:system":
|
|
raise AssertionError(result.output)
|
|
print("[mcp] PASS")
|
|
finally:
|
|
await manager.close()
|
|
|
|
|
|
async def _run_plugin_flow(temp_root: Path) -> None:
|
|
plugin_source = _write_plugin(temp_root / "plugin-source")
|
|
install_plugin_from_path(plugin_source)
|
|
project = temp_root / "project"
|
|
project.mkdir()
|
|
try:
|
|
plugins = load_plugins(Settings(), project)
|
|
manager = McpClientManager(load_mcp_server_configs(Settings(), plugins))
|
|
await manager.connect_all()
|
|
try:
|
|
registry = create_default_tool_registry(manager)
|
|
skill_tool = registry.get("skill")
|
|
skill_result = await skill_tool.execute(
|
|
skill_tool.input_model.model_validate({"name": "FixtureSkill"}),
|
|
ToolExecutionContext(cwd=project),
|
|
)
|
|
mcp_tool = registry.get("mcp__fixture-plugin_fixture__hello")
|
|
mcp_result = await mcp_tool.execute(
|
|
mcp_tool.input_model.model_validate({"name": "plugin"}),
|
|
ToolExecutionContext(cwd=project),
|
|
)
|
|
if "Fixture skill content" not in skill_result.output or mcp_result.output != "fixture-hello:plugin":
|
|
raise AssertionError("plugin flow failed")
|
|
print("[plugin] PASS")
|
|
finally:
|
|
await manager.close()
|
|
finally:
|
|
uninstall_plugin("fixture-plugin")
|
|
|
|
|
|
async def _run_plugin_command_flow(temp_root: Path) -> None:
|
|
registry = create_default_command_registry()
|
|
context = _make_command_context(temp_root)
|
|
plugin_source = _write_plugin(temp_root / "plugin-command-source")
|
|
|
|
for raw in [
|
|
f"/plugin install {plugin_source}",
|
|
"/plugin disable fixture-plugin",
|
|
"/plugin enable fixture-plugin",
|
|
"/plugin uninstall fixture-plugin",
|
|
]:
|
|
command, args = registry.lookup(raw)
|
|
result = await command.handler(args, context)
|
|
if result.message is None:
|
|
raise AssertionError(f"no result for {raw}")
|
|
print("[plugin-commands] PASS")
|
|
|
|
|
|
async def _run_bridge_flow(temp_root: Path) -> None:
|
|
handle = await spawn_session(
|
|
session_id="local-bridge",
|
|
command="printf 'bridge-system-ok' > bridge.txt",
|
|
cwd=temp_root,
|
|
)
|
|
await handle.process.wait()
|
|
secret = WorkSecret(version=1, session_ingress_token="tok", api_base_url="http://localhost:8080")
|
|
encoded = encode_work_secret(secret)
|
|
decoded = decode_work_secret(encoded)
|
|
url = build_sdk_url(decoded.api_base_url, "abc")
|
|
if (temp_root / "bridge.txt").read_text(encoding="utf-8") != "bridge-system-ok":
|
|
raise AssertionError("bridge file missing")
|
|
if url != "ws://localhost:8080/v2/session_ingress/ws/abc":
|
|
raise AssertionError(url)
|
|
print("[bridge] PASS")
|
|
|
|
|
|
async def _run_command_flow(temp_root: Path) -> None:
|
|
registry = create_default_command_registry()
|
|
context = _make_command_context(temp_root)
|
|
(temp_root / "src").mkdir()
|
|
(temp_root / "src" / "demo.py").write_text("print('demo')\n", encoding="utf-8")
|
|
for raw in [
|
|
"/memory add Notes :: local command note",
|
|
"/output-style set minimal",
|
|
"/vim on",
|
|
"/voice on",
|
|
"/plan on",
|
|
"/effort high",
|
|
"/passes 2",
|
|
"/tasks run printf 'local-command-task'",
|
|
"/init",
|
|
]:
|
|
command, args = registry.lookup(raw)
|
|
await command.handler(args, context)
|
|
doctor_command, doctor_args = registry.lookup("/doctor")
|
|
doctor_result = await doctor_command.handler(doctor_args, context)
|
|
if "- output_style: minimal" not in doctor_result.message:
|
|
raise AssertionError(doctor_result.message)
|
|
|
|
for raw, expected in [
|
|
("/files demo.py", "src/demo.py"),
|
|
("/session", "Session directory:"),
|
|
("/session tag local-smoke", "local-smoke.json"),
|
|
("/bridge show", "Bridge summary:"),
|
|
("/privacy-settings", "Privacy settings:"),
|
|
("/rate-limit-options", "Rate limit options:"),
|
|
("/release-notes", "Release Notes"),
|
|
("/upgrade", "Upgrade instructions:"),
|
|
]:
|
|
command, args = registry.lookup(raw)
|
|
result = await command.handler(args, context)
|
|
if expected not in (result.message or ""):
|
|
raise AssertionError(f"{raw} failed: {result.message}")
|
|
|
|
print("[commands] PASS")
|
|
|
|
|
|
async def main() -> int:
|
|
with tempfile.TemporaryDirectory(prefix="openharness-local-system-") as temp_dir:
|
|
temp_root = Path(temp_dir)
|
|
previous = {
|
|
"OPENHARNESS_CONFIG_DIR": os.environ.get("OPENHARNESS_CONFIG_DIR"),
|
|
"OPENHARNESS_DATA_DIR": os.environ.get("OPENHARNESS_DATA_DIR"),
|
|
}
|
|
os.environ["OPENHARNESS_CONFIG_DIR"] = str(temp_root / "config")
|
|
os.environ["OPENHARNESS_DATA_DIR"] = str(temp_root / "data")
|
|
try:
|
|
await _run_mcp_flow(temp_root)
|
|
await _run_plugin_flow(temp_root)
|
|
await _run_plugin_command_flow(temp_root)
|
|
await _run_bridge_flow(temp_root)
|
|
await _run_command_flow(temp_root)
|
|
print("Local system scenarios passed")
|
|
return 0
|
|
finally:
|
|
for key, value in previous.items():
|
|
if value is None:
|
|
os.environ.pop(key, None)
|
|
else:
|
|
os.environ[key] = value
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|