wehub-resource-sync
378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""Unit tests for the cursor-native (terminal-injection) harness.
|
|
|
|
Covers the executor's text extraction + capability flags, the tmux bridge's pure
|
|
helpers (paste-payload encoding, bridge dir, spawn env, tmux.json round-trip),
|
|
and harness registration. The live tmux injection is exercised by the e2e gate,
|
|
not here, so these need no tmux or cursor-agent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from omnigent.cursor_native_bridge import (
|
|
BRIDGE_DIR_ENV_VAR,
|
|
FORK_HISTORY_CLOSE_TAG,
|
|
FORK_HISTORY_OPEN_TAG,
|
|
_paste_payload_bytes,
|
|
allow_mcp_tools_in_cli_config,
|
|
approve_mcp_server_for_workspace,
|
|
bridge_dir_for_session_id,
|
|
build_cursor_native_spawn_env,
|
|
build_mcp_config,
|
|
clear_fork_preamble,
|
|
cursor_project_key,
|
|
enable_mcp_for_workspace,
|
|
read_fork_preamble,
|
|
read_tmux_info,
|
|
wrap_fork_preamble,
|
|
write_fork_preamble,
|
|
write_mcp_bridge_config,
|
|
write_mcp_config,
|
|
write_tmux_target,
|
|
)
|
|
from omnigent.inner import cursor_native_executor as cne
|
|
from omnigent.inner.cursor_native_executor import (
|
|
CursorNativeExecutor,
|
|
_content_to_text,
|
|
_latest_user_text,
|
|
)
|
|
from omnigent.inner.executor import ExecutorError
|
|
|
|
|
|
class TestContentExtraction:
|
|
def test_string_content(self, tmp_path: Path) -> None:
|
|
assert _content_to_text("hello", tmp_path) == "hello"
|
|
|
|
def test_input_text_blocks(self, tmp_path: Path) -> None:
|
|
content = [
|
|
{"type": "input_text", "text": "one"},
|
|
{"type": "text", "text": "two"},
|
|
# invalid data URI -> materialize_attachment returns None -> no line
|
|
{"type": "input_image", "image_url": "data:..."},
|
|
]
|
|
assert _content_to_text(content, tmp_path) == "one\n\ntwo"
|
|
|
|
def test_real_image_attachment_materialized(self, tmp_path: Path) -> None:
|
|
# a tiny valid base64 PNG data URI should be written to disk + referenced
|
|
png = (
|
|
"data:image/png;base64,"
|
|
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
|
|
)
|
|
out = _content_to_text([{"type": "input_image", "image_url": png}], tmp_path)
|
|
assert out.startswith("[Attached: ")
|
|
assert str(tmp_path) in out
|
|
|
|
def test_empty_and_none(self, tmp_path: Path) -> None:
|
|
assert _content_to_text(None, tmp_path) == ""
|
|
assert _content_to_text([], tmp_path) == ""
|
|
|
|
def test_latest_user_text(self, tmp_path: Path) -> None:
|
|
messages = [
|
|
{"role": "user", "content": "first"},
|
|
{"role": "assistant", "content": "ok"},
|
|
{"role": "user", "content": "second"},
|
|
]
|
|
assert _latest_user_text(messages, tmp_path) == "second"
|
|
assert _latest_user_text([{"role": "assistant", "content": "x"}], tmp_path) == ""
|
|
|
|
|
|
class TestExecutorCapabilities:
|
|
def test_capability_flags(self, tmp_path: Path) -> None:
|
|
ex = CursorNativeExecutor(bridge_dir=tmp_path)
|
|
# Output is shown by the embedded terminal, not streamed by the executor.
|
|
assert ex.supports_streaming() is False
|
|
# Web-UI messages can be injected mid-turn (steering).
|
|
assert ex.supports_live_message_queue() is True
|
|
|
|
|
|
class TestForkPreamble:
|
|
"""The fork preamble file + sentinel framing (text-prefix replay)."""
|
|
|
|
def test_read_does_not_consume_clear_does(self, tmp_path: Path) -> None:
|
|
write_fork_preamble(tmp_path, "You: hi")
|
|
# Read is idempotent (the file survives) so a failed/retried injection
|
|
# can re-read it; only clear consumes.
|
|
assert read_fork_preamble(tmp_path) == "You: hi"
|
|
assert read_fork_preamble(tmp_path) == "You: hi"
|
|
clear_fork_preamble(tmp_path)
|
|
assert read_fork_preamble(tmp_path) is None
|
|
|
|
def test_empty_preamble_is_not_written(self, tmp_path: Path) -> None:
|
|
write_fork_preamble(tmp_path, "")
|
|
assert read_fork_preamble(tmp_path) is None
|
|
|
|
def test_read_missing_is_none_and_clear_is_noop(self, tmp_path: Path) -> None:
|
|
assert read_fork_preamble(tmp_path) is None
|
|
clear_fork_preamble(tmp_path) # no file -> no error
|
|
|
|
def test_wrap_fences_preamble_before_user_text(self) -> None:
|
|
wrapped = wrap_fork_preamble("Conversation so far:\nuser: hi", "do it")
|
|
assert wrapped.startswith(FORK_HISTORY_OPEN_TAG)
|
|
assert FORK_HISTORY_CLOSE_TAG in wrapped
|
|
# The user's real text follows the fenced block.
|
|
assert wrapped.endswith("do it")
|
|
assert wrapped.index(FORK_HISTORY_CLOSE_TAG) < wrapped.index("do it")
|
|
|
|
def test_wrap_defangs_sentinels_inside_preamble(self) -> None:
|
|
# A literal close tag in the replayed transcript must not produce a second
|
|
# real close tag inside the block (which would let the forwarder's strip
|
|
# stop early and leak history). The framed block holds exactly one pair.
|
|
wrapped = wrap_fork_preamble(f"You: see {FORK_HISTORY_CLOSE_TAG} here", "go")
|
|
assert wrapped.count(FORK_HISTORY_CLOSE_TAG) == 1
|
|
assert wrapped.count(FORK_HISTORY_OPEN_TAG) == 1
|
|
# The defanged form stays readable.
|
|
assert "[/omnigent_fork_history]" in wrapped
|
|
|
|
|
|
class TestRunTurnPreambleInjection:
|
|
"""run_turn prepends the fork preamble to the FIRST injected message only."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_turn_injects_preamble_then_consumes_it(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injected: list[str] = []
|
|
monkeypatch.setattr(
|
|
cne,
|
|
"inject_user_message",
|
|
lambda bridge_dir, content: injected.append(content),
|
|
)
|
|
write_fork_preamble(tmp_path, "You: earlier")
|
|
ex = CursorNativeExecutor(bridge_dir=tmp_path)
|
|
|
|
# First turn: the preamble rides into the injected text, fenced.
|
|
async for _ in ex.run_turn([{"role": "user", "content": "first"}], [], ""):
|
|
pass
|
|
assert FORK_HISTORY_OPEN_TAG in injected[0]
|
|
assert "You: earlier" in injected[0]
|
|
assert injected[0].endswith("first")
|
|
|
|
# Second turn: preamble consumed -> plain user text only.
|
|
async for _ in ex.run_turn([{"role": "user", "content": "second"}], [], ""):
|
|
pass
|
|
assert injected[1] == "second"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_injection_preserves_preamble_for_retry(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# If the FIRST injection fails (e.g. the TUI exited), the preamble must
|
|
# NOT be consumed — otherwise the forked history is lost permanently and
|
|
# a retried first turn launches with no prior context.
|
|
injected: list[str] = []
|
|
calls = {"n": 0}
|
|
|
|
def fake_inject(bridge_dir: Path, content: str) -> None:
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise RuntimeError("tmux target not advertised")
|
|
injected.append(content)
|
|
|
|
monkeypatch.setattr(cne, "inject_user_message", fake_inject)
|
|
write_fork_preamble(tmp_path, "You: earlier")
|
|
ex = CursorNativeExecutor(bridge_dir=tmp_path)
|
|
|
|
# First turn: injection fails -> ExecutorError, preamble still on disk.
|
|
events = [e async for e in ex.run_turn([{"role": "user", "content": "first"}], [], "")]
|
|
assert any(isinstance(e, ExecutorError) for e in events)
|
|
assert read_fork_preamble(tmp_path) == "You: earlier"
|
|
|
|
# Retry: injection succeeds, history rides along, THEN it's consumed.
|
|
async for _ in ex.run_turn([{"role": "user", "content": "first"}], [], ""):
|
|
pass
|
|
assert "You: earlier" in injected[0]
|
|
assert read_fork_preamble(tmp_path) is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_preamble_injects_plain_text(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injected: list[str] = []
|
|
monkeypatch.setattr(
|
|
cne,
|
|
"inject_user_message",
|
|
lambda bridge_dir, content: injected.append(content),
|
|
)
|
|
ex = CursorNativeExecutor(bridge_dir=tmp_path)
|
|
async for _ in ex.run_turn([{"role": "user", "content": "hello"}], [], ""):
|
|
pass
|
|
assert injected == ["hello"]
|
|
|
|
|
|
class TestPastePayload:
|
|
def test_newlines_become_cr(self) -> None:
|
|
assert _paste_payload_bytes("a\nb") == b"a\rb"
|
|
assert _paste_payload_bytes("a\r\nb") == b"a\rb"
|
|
assert _paste_payload_bytes("a\rb") == b"a\rb"
|
|
|
|
def test_tab_kept_other_control_dropped(self) -> None:
|
|
# tab kept (0x09), ESC (0x1b) and BEL (0x07) dropped.
|
|
assert _paste_payload_bytes("a\tb\x1b\x07c") == b"a\tbc"
|
|
|
|
def test_unicode_passthrough(self) -> None:
|
|
assert _paste_payload_bytes("café") == "café".encode()
|
|
|
|
|
|
class TestBridge:
|
|
def test_bridge_dir_is_deterministic_and_session_scoped(self) -> None:
|
|
a1 = bridge_dir_for_session_id("conv_a")
|
|
a2 = bridge_dir_for_session_id("conv_a")
|
|
b = bridge_dir_for_session_id("conv_b")
|
|
assert a1 == a2
|
|
assert a1 != b
|
|
assert "cursor-native" in str(a1)
|
|
|
|
def test_spawn_env_carries_bridge_dir(self) -> None:
|
|
env = build_cursor_native_spawn_env("conv_xyz")
|
|
assert env[BRIDGE_DIR_ENV_VAR] == str(bridge_dir_for_session_id("conv_xyz"))
|
|
# The cursor bridge has no active-session concept (unlike claude/pi), so
|
|
# no request-session-id guard env is emitted — only the bridge dir.
|
|
assert "HARNESS_CURSOR_NATIVE_REQUEST_SESSION_ID" not in env
|
|
assert list(env) == [BRIDGE_DIR_ENV_VAR]
|
|
|
|
def test_tmux_target_round_trip(self, tmp_path: Path) -> None:
|
|
write_tmux_target(tmp_path, socket_path=Path("/tmp/x/tmux.sock"), tmux_target="main")
|
|
info = read_tmux_info(tmp_path)
|
|
assert info == {"socket_path": "/tmp/x/tmux.sock", "tmux_target": "main"}
|
|
|
|
def test_read_tmux_info_missing(self, tmp_path: Path) -> None:
|
|
assert read_tmux_info(tmp_path) is None
|
|
|
|
def test_build_mcp_config_registers_omnigent_relay(self, tmp_path: Path) -> None:
|
|
config = build_mcp_config(tmp_path, python_executable="python-test")
|
|
server = config["mcpServers"]["omnigent"]
|
|
assert server["command"] == "python-test"
|
|
assert server["args"] == [
|
|
"-I",
|
|
"-m",
|
|
"omnigent.claude_native_bridge",
|
|
"serve-mcp",
|
|
"--bridge-dir",
|
|
str(tmp_path),
|
|
]
|
|
assert "sys_session_send" in server["autoApprove"]
|
|
assert "sys_os_shell" in server["autoApprove"]
|
|
assert server["autoApprove"] == sorted(server["autoApprove"])
|
|
assert server["env"]["TMPDIR"]
|
|
|
|
def test_write_mcp_config_is_workspace_scoped(self, tmp_path: Path, monkeypatch) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
bridge_dir = tmp_path / "bridge"
|
|
monkeypatch.setattr(
|
|
"omnigent.cursor_native_bridge.approve_mcp_server_for_workspace",
|
|
lambda _workspace: pytest.fail("approval must happen after tool relay starts"),
|
|
)
|
|
path = write_mcp_config(workspace, bridge_dir, python_executable="python-test")
|
|
|
|
assert path == workspace / ".cursor" / "mcp.json"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert payload["mcpServers"]["omnigent"]["command"] == "python-test"
|
|
assert json.loads((bridge_dir / "bridge.json").read_text(encoding="utf-8"))["token"]
|
|
|
|
def test_write_mcp_bridge_config_is_idempotent(self, tmp_path: Path) -> None:
|
|
write_mcp_bridge_config(tmp_path)
|
|
first = (tmp_path / "bridge.json").read_text(encoding="utf-8")
|
|
write_mcp_bridge_config(tmp_path)
|
|
assert (tmp_path / "bridge.json").read_text(encoding="utf-8") == first
|
|
|
|
def test_cursor_project_key_matches_cursor_workspace_state(self) -> None:
|
|
assert cursor_project_key(Path("/Users/corey.zumar")) == "Users-corey.zumar"
|
|
|
|
def test_enable_mcp_for_workspace_removes_disabled_entry(
|
|
self, tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
disabled_path = (
|
|
tmp_path / ".cursor" / "projects" / "Users-corey.zumar" / "mcp-disabled.json"
|
|
)
|
|
disabled_path.parent.mkdir(parents=True)
|
|
disabled_path.write_text('["omnigent", "other"]\n', encoding="utf-8")
|
|
|
|
enable_mcp_for_workspace(Path("/Users/corey.zumar"))
|
|
|
|
assert json.loads(disabled_path.read_text(encoding="utf-8")) == ["other"]
|
|
|
|
def test_allow_mcp_tools_in_cli_config_adds_specific_allow_rules(
|
|
self, tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
config_path = tmp_path / ".cursor" / "cli-config.json"
|
|
config_path.parent.mkdir(parents=True)
|
|
config_path.write_text(
|
|
'{"permissions": {"allow": ["Shell(ls)", "Mcp(omnigent:sys_os_read)"]}}\n',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
allow_mcp_tools_in_cli_config()
|
|
|
|
allow = json.loads(config_path.read_text(encoding="utf-8"))["permissions"]["allow"]
|
|
assert "Shell(ls)" in allow
|
|
assert allow.count("Mcp(omnigent:sys_os_read)") == 1
|
|
assert "Mcp(omnigent:sys_session_send)" in allow
|
|
|
|
def test_approve_mcp_server_for_workspace_uses_cursor_cli(
|
|
self, tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
calls: list[dict[str, object]] = []
|
|
|
|
monkeypatch.setattr(
|
|
"omnigent.cursor_native.resolve_cursor_executable",
|
|
lambda: "/bin/cursor-agent-test",
|
|
)
|
|
|
|
def fake_run(*args, **kwargs):
|
|
calls.append({"args": args, "kwargs": kwargs})
|
|
return subprocess.CompletedProcess(args[0], 0)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
|
|
approve_mcp_server_for_workspace(tmp_path)
|
|
|
|
assert calls == [
|
|
{
|
|
"args": (["/bin/cursor-agent-test", "mcp", "enable", "omnigent"],),
|
|
"kwargs": {
|
|
"cwd": tmp_path,
|
|
"stdin": subprocess.DEVNULL,
|
|
"stdout": subprocess.DEVNULL,
|
|
"stderr": subprocess.DEVNULL,
|
|
"timeout": 15,
|
|
"check": False,
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
class TestRegistration:
|
|
def test_harness_is_registered(self) -> None:
|
|
from omnigent.runtime.harnesses import _HARNESS_MODULES
|
|
|
|
assert _HARNESS_MODULES["cursor-native"] == "omnigent.inner.cursor_native_harness"
|
|
|
|
def test_harness_is_allowlisted(self) -> None:
|
|
from omnigent.spec._omnigent_compat import OMNIGENT_HARNESSES
|
|
|
|
assert "cursor-native" in OMNIGENT_HARNESSES
|
|
|
|
def test_cursor_native_is_terminal_native(self) -> None:
|
|
# cursor-native launches the cursor-agent TUI in an omnigent terminal
|
|
# (like claude/codex/pi-native), so the runner must treat it as a native
|
|
# terminal harness.
|
|
from omnigent.harness_aliases import is_native_harness
|
|
|
|
assert is_native_harness("cursor-native") is True
|
|
assert is_native_harness("native-cursor") is True
|
|
|
|
def test_native_coding_agent_record(self) -> None:
|
|
from omnigent.native_coding_agents import native_coding_agent_for_harness
|
|
|
|
agent = native_coding_agent_for_harness("cursor-native")
|
|
assert agent is not None
|
|
assert agent.terminal_name == "cursor"
|
|
assert agent.display_name == "Cursor"
|