wehub-resource-sync
281 lines
9.4 KiB
Python
281 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import sys
|
|
from collections.abc import AsyncIterator, Callable
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from opensquilla.engine.commands import Surface
|
|
|
|
|
|
class _FakeOutputHandle:
|
|
approval_surface = Surface.CLI_GATEWAY
|
|
|
|
async def write_through(self, payload: str) -> None:
|
|
return None
|
|
|
|
async def send_message(self, message_type: str, payload: dict) -> None:
|
|
return None
|
|
|
|
def stream_output(self):
|
|
@asynccontextmanager
|
|
async def _cm() -> AsyncIterator[Callable[[str], None]]:
|
|
yield lambda _payload: None
|
|
|
|
return _cm()
|
|
|
|
|
|
class _FakeOpenTuiSurface:
|
|
output_handle = _FakeOutputHandle()
|
|
|
|
def __init__(self) -> None:
|
|
self.writes: list[str] = []
|
|
|
|
async def next_line(self) -> str | None:
|
|
return None
|
|
|
|
def set_cancel_callback(self, cb: Callable[[], None] | None) -> None:
|
|
return None
|
|
|
|
def set_shutdown_callback(self, cb: Callable[[], None] | None) -> None:
|
|
return None
|
|
|
|
def emit_eof(self) -> None:
|
|
return None
|
|
|
|
async def write_through(self, payload: str) -> None:
|
|
self.writes.append(payload)
|
|
|
|
async def send_message(self, message_type: str, payload: dict) -> None:
|
|
self.writes.append(f"{message_type}:{payload.get('text', '')}")
|
|
|
|
@property
|
|
def redraw_callback(self) -> Callable[[], None]:
|
|
return lambda: None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_opentui_chat_runtime_exposes_tui_output_and_reuses_runtime(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui.opentui import runtime as opentui_runtime
|
|
|
|
scope: dict[str, Any] = {
|
|
"model": "model-a",
|
|
"session_key": "session-a",
|
|
"tool_ctx": SimpleNamespace(workspace_dir="/tmp/opentui-workspace"),
|
|
}
|
|
captured: dict[str, Any] = {}
|
|
fake_surface = _FakeOpenTuiSurface()
|
|
|
|
@asynccontextmanager
|
|
async def fake_open_opentui_surface(**kwargs: Any):
|
|
captured["surface_kwargs"] = kwargs
|
|
yield fake_surface
|
|
|
|
async def fake_run_tui_runtime(**kwargs: Any):
|
|
captured["runtime_kwargs"] = kwargs
|
|
async with kwargs["surface_factory"]() as yielded:
|
|
assert yielded is fake_surface
|
|
hooks = kwargs["hooks"]
|
|
assert not opentui_runtime.get_tui_output(scope)
|
|
hooks.expose_surface(fake_surface)
|
|
output = opentui_runtime.get_tui_output(scope)
|
|
captured["output"] = output
|
|
captured["manager"] = getattr(output, "plugin_manager", None)
|
|
hooks.clear_exposed_surface()
|
|
return object()
|
|
|
|
monkeypatch.setattr(opentui_runtime, "open_opentui_surface", fake_open_opentui_surface)
|
|
monkeypatch.setattr(opentui_runtime, "run_tui_runtime", fake_run_tui_runtime)
|
|
|
|
async def fake_dispatch(_value: str) -> bool:
|
|
return True
|
|
|
|
await opentui_runtime.run_opentui_chat_runtime(
|
|
surface=Surface.CLI_GATEWAY,
|
|
scope=scope,
|
|
dispatch=fake_dispatch,
|
|
queue_max_size=8,
|
|
)
|
|
|
|
assert captured["surface_kwargs"] == {
|
|
"surface": Surface.CLI_GATEWAY,
|
|
"model": "model-a",
|
|
"session_id": "session-a",
|
|
"workspace_dir": "/tmp/opentui-workspace",
|
|
}
|
|
assert captured["runtime_kwargs"]["dispatch"] is fake_dispatch
|
|
assert captured["runtime_kwargs"]["config"].concurrent_input_during_turn is True
|
|
assert opentui_runtime.get_tui_output(scope) is None
|
|
assert getattr(captured["output"], "_output_handle", None) is fake_surface.output_handle
|
|
assert captured["manager"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_opentui_chat_runtime_forwards_workspace_dir_from_tool_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui.opentui import runtime as opentui_runtime
|
|
|
|
scope: dict[str, Any] = {
|
|
"model": "model-a",
|
|
"session_key": "session-a",
|
|
"tool_ctx": SimpleNamespace(workspace_dir="/tmp/workspace-a"),
|
|
}
|
|
captured: dict[str, Any] = {}
|
|
fake_surface = _FakeOpenTuiSurface()
|
|
|
|
@asynccontextmanager
|
|
async def fake_open_opentui_surface(**kwargs: Any):
|
|
captured["surface_kwargs"] = kwargs
|
|
yield fake_surface
|
|
|
|
async def fake_run_tui_runtime(**kwargs: Any):
|
|
async with kwargs["surface_factory"]() as yielded:
|
|
assert yielded is fake_surface
|
|
|
|
monkeypatch.setattr(opentui_runtime, "open_opentui_surface", fake_open_opentui_surface)
|
|
monkeypatch.setattr(opentui_runtime, "run_tui_runtime", fake_run_tui_runtime)
|
|
|
|
async def fake_dispatch(_value: str) -> bool:
|
|
return True
|
|
|
|
await opentui_runtime.run_opentui_chat_runtime(
|
|
surface=Surface.CLI_GATEWAY,
|
|
scope=scope,
|
|
dispatch=fake_dispatch,
|
|
queue_max_size=8,
|
|
)
|
|
|
|
assert captured["surface_kwargs"]["workspace_dir"] == "/tmp/workspace-a"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_opentui_chat_runtime_uses_footer_native_echo_hooks(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui.adapters import runtime_helpers
|
|
from opensquilla.cli.tui.opentui import runtime as opentui_runtime
|
|
|
|
assert runtime_helpers.classify_chat_input("/help") is not None
|
|
|
|
scope: dict[str, Any] = {"model": "model-a", "session_key": "session-a"}
|
|
fake_surface = _FakeOpenTuiSurface()
|
|
|
|
@asynccontextmanager
|
|
async def fake_open_opentui_surface(**_kwargs: Any):
|
|
yield fake_surface
|
|
|
|
async def fake_run_tui_runtime(**kwargs: Any):
|
|
hooks = kwargs["hooks"]
|
|
await hooks.on_user_input_echo(fake_surface, "hello opentui")
|
|
await hooks.on_user_input_echo(fake_surface, "中文输入 CJK混合ASCII")
|
|
await hooks.on_queued_turn_start(fake_surface)
|
|
return object()
|
|
|
|
monkeypatch.setattr(opentui_runtime, "open_opentui_surface", fake_open_opentui_surface)
|
|
monkeypatch.setattr(opentui_runtime, "run_tui_runtime", fake_run_tui_runtime)
|
|
|
|
async def fake_dispatch(_value: str) -> bool:
|
|
return True
|
|
|
|
await opentui_runtime.run_opentui_chat_runtime(
|
|
surface=Surface.CLI_GATEWAY,
|
|
scope=scope,
|
|
dispatch=fake_dispatch,
|
|
queue_max_size=8,
|
|
)
|
|
|
|
joined_writes = "".join(fake_surface.writes)
|
|
assert "你 / you" not in joined_writes
|
|
assert "prompt.echo:hello opentui" in joined_writes
|
|
assert "中文输入 CJK混合ASCII" in joined_writes
|
|
assert "running queued input" in joined_writes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_opentui_chat_runtime_reprints_exit_notices_to_real_stderr(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""The surface-error notice carries the only copy of a sidecar crash
|
|
reason, and the Goodbye notice is emitted with the bridge already doomed.
|
|
Both must reach the real terminal stderr after teardown instead of dying
|
|
with the captured console."""
|
|
from opensquilla.cli.tui.opentui import runtime as opentui_runtime
|
|
|
|
scope: dict[str, Any] = {"model": "model-a", "session_key": "session-a"}
|
|
fake_surface = _FakeOpenTuiSurface()
|
|
|
|
@asynccontextmanager
|
|
async def fake_open_opentui_surface(**_kwargs: Any):
|
|
yield fake_surface
|
|
|
|
async def fake_run_tui_runtime(**kwargs: Any):
|
|
hooks = kwargs["hooks"]
|
|
hooks.notice("[red]Input surface error: OpenTUI host exited with code 7[/red]")
|
|
hooks.notice("[yellow]Goodbye.[/yellow]")
|
|
return object()
|
|
|
|
monkeypatch.setattr(opentui_runtime, "open_opentui_surface", fake_open_opentui_surface)
|
|
monkeypatch.setattr(opentui_runtime, "run_tui_runtime", fake_run_tui_runtime)
|
|
|
|
fake_stderr = io.StringIO()
|
|
monkeypatch.setattr(sys, "stderr", fake_stderr)
|
|
|
|
async def fake_dispatch(_value: str) -> bool:
|
|
return True
|
|
|
|
await opentui_runtime.run_opentui_chat_runtime(
|
|
surface=Surface.CLI_GATEWAY,
|
|
scope=scope,
|
|
dispatch=fake_dispatch,
|
|
queue_max_size=8,
|
|
)
|
|
|
|
output = fake_stderr.getvalue()
|
|
assert "Input surface error: OpenTUI host exited with code 7" in output
|
|
assert "Goodbye" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forward_console_notice_prunes_completed_pending_tasks() -> None:
|
|
"""The session-scoped pending set must shed tasks as their sends finish —
|
|
every captured console line schedules one, so retaining completed tasks
|
|
grows without bound over a long interactive session."""
|
|
from opensquilla.cli.tui.backend.output_binding import TuiOutputBinding
|
|
from opensquilla.cli.tui.opentui import runtime as opentui_runtime
|
|
|
|
scope: dict[str, Any] = {}
|
|
TuiOutputBinding(scope).expose(_FakeOutputHandle())
|
|
pending: set[asyncio.Task[None]] = set()
|
|
|
|
for index in range(5):
|
|
opentui_runtime.forward_console_notice(scope, f"line-{index}", pending_tasks=pending)
|
|
assert len(pending) == 5
|
|
|
|
await asyncio.gather(*pending)
|
|
# Done callbacks run one call_soon hop after completion.
|
|
await asyncio.sleep(0)
|
|
assert pending == set()
|
|
|
|
|
|
def test_opentui_notice_renders_through_captured_console() -> None:
|
|
from opensquilla.cli.tui.opentui.notice_capture import capture_stdout_as_notices
|
|
from opensquilla.cli.tui.opentui.runtime import opentui_notice
|
|
|
|
lines: list[str] = []
|
|
# Runtime notices must render as clean styled host notices, never as raw Rich
|
|
# markup. With the capture installed, console.print is forwarded line-by-line.
|
|
with capture_stdout_as_notices(lines.append):
|
|
opentui_notice({}, "[yellow]Hello[/yellow]")
|
|
|
|
joined = "".join(lines)
|
|
assert "Hello" in joined
|
|
assert "[yellow]" not in joined # markup must be rendered out, not leaked verbatim
|