wehub-resource-sync
250 lines
8.9 KiB
Python
250 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import inspect
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
from opensquilla.cli.repl import standalone_slash_adapter
|
|
from opensquilla.cli.repl.session_state import ChatSessionState
|
|
from opensquilla.cli.repl.stream import TurnResult, UsageSummary
|
|
from opensquilla.cli.tui.contracts import TuiOutputHandle
|
|
from opensquilla.engine.commands import Surface
|
|
|
|
|
|
def test_standalone_runtime_has_no_raw_prompt_application_dependency(monkeypatch) -> None:
|
|
monkeypatch.delitem(
|
|
sys.modules,
|
|
"opensquilla.cli.repl.standalone_runtime",
|
|
raising=False,
|
|
)
|
|
|
|
original_import = __import__
|
|
blocked_module = "prompt" + "_toolkit"
|
|
|
|
def _guarded_import(name, globals=None, locals=None, fromlist=(), level=0): # noqa: ANN001
|
|
if name == blocked_module or name.startswith(f"{blocked_module}."):
|
|
raise AssertionError(f"standalone runtime imported {blocked_module} via {name}")
|
|
return original_import(name, globals, locals, fromlist, level)
|
|
|
|
monkeypatch.setattr("builtins.__import__", _guarded_import)
|
|
|
|
module = importlib.import_module("opensquilla.cli.repl.standalone_runtime")
|
|
source = inspect.getsource(module)
|
|
|
|
assert "ChatApplication" not in source
|
|
|
|
|
|
def test_standalone_session_context_mirrors_state_to_legacy_scope() -> None:
|
|
from opensquilla.cli.repl.standalone_runtime import StandaloneSessionContext
|
|
|
|
original_tool_ctx = object()
|
|
replacement_tool_ctx = object()
|
|
state = ChatSessionState(
|
|
session_key="agent:main:standalone:original",
|
|
model="standalone/original",
|
|
)
|
|
replacement_state = ChatSessionState(
|
|
session_key="agent:main:standalone:replacement",
|
|
model="standalone/replacement",
|
|
)
|
|
|
|
context = StandaloneSessionContext.create(state=state, tool_ctx=original_tool_ctx)
|
|
|
|
assert context.scope["session_key"] == "agent:main:standalone:original"
|
|
assert context.scope["tool_ctx"] is original_tool_ctx
|
|
assert context.scope["state"] is state
|
|
assert context.scope["model"] == "standalone/original"
|
|
|
|
context.replace_session(
|
|
session_key="agent:main:standalone:replacement",
|
|
tool_ctx=replacement_tool_ctx,
|
|
state=replacement_state,
|
|
model="standalone/replacement",
|
|
)
|
|
|
|
assert context.session_key == "agent:main:standalone:replacement"
|
|
assert context.tool_ctx is replacement_tool_ctx
|
|
assert context.model == "standalone/replacement"
|
|
assert context.scope["session_key"] == "agent:main:standalone:replacement"
|
|
assert context.scope["tool_ctx"] is replacement_tool_ctx
|
|
assert context.scope["state"] is replacement_state
|
|
assert context.scope["model"] == "standalone/replacement"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_runtime_mirrors_turn_model_update_to_legacy_scope(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.repl import standalone_runtime
|
|
|
|
class _FakeSessionManager:
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main") -> object:
|
|
return SimpleNamespace(session_key=session_key, agent_id=agent_id)
|
|
|
|
class _FakeServices:
|
|
def __init__(self) -> None:
|
|
self.config = None
|
|
self.session_manager = _FakeSessionManager()
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
services = _FakeServices()
|
|
turn_runner = object()
|
|
output = cast(TuiOutputHandle, object())
|
|
pending_provider = SimpleNamespace(drain_pending=lambda: [])
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
def fake_build_turn_runner_from_services(_services: object) -> object:
|
|
return turn_runner
|
|
|
|
async def fake_stream_response(
|
|
active_turn_runner: object,
|
|
session_key: str,
|
|
tool_ctx: object,
|
|
message: str,
|
|
model: str | None = None,
|
|
svc: object = None,
|
|
timeout: float | None = None,
|
|
*,
|
|
tui_output: TuiOutputHandle | None = None,
|
|
pending_input_provider: object | None = None,
|
|
) -> TurnResult:
|
|
captured["stream"] = {
|
|
"turn_runner": active_turn_runner,
|
|
"session_key": session_key,
|
|
"tool_ctx": tool_ctx,
|
|
"message": message,
|
|
"model": model,
|
|
"svc": svc,
|
|
"timeout": timeout,
|
|
"tui_output": tui_output,
|
|
"pending_input_provider": pending_input_provider,
|
|
}
|
|
return TurnResult(
|
|
text="assistant reply",
|
|
usage=UsageSummary(input_tokens=5, output_tokens=8),
|
|
model_after="standalone/after",
|
|
)
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch,
|
|
) -> None:
|
|
captured["surface"] = surface
|
|
captured["initial_scope"] = dict(scope)
|
|
scope["pending_input_provider"] = pending_provider
|
|
|
|
assert await dispatch("hello") is True
|
|
|
|
captured["scope_after_message"] = dict(scope)
|
|
captured["state_after_message"] = scope["state"]
|
|
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.build_turn_runner_from_services",
|
|
fake_build_turn_runner_from_services,
|
|
)
|
|
|
|
deps = standalone_runtime.StandaloneRuntimeDependencies(
|
|
stream_response=fake_stream_response,
|
|
image_command_handler=fake_stream_response,
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
slash_services_factory=lambda _svc: standalone_slash_adapter.StandaloneSlashServices(),
|
|
sync_slash_adapter_io=lambda: None,
|
|
get_tui_output=lambda _scope: output,
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model="openrouter/test",
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert captured["surface"] is Surface.CLI_STANDALONE
|
|
assert captured["initial_scope"]["session_key"] == "agent:main:standalone:test"
|
|
assert captured["initial_scope"]["model"] == "openrouter/test"
|
|
assert captured["stream"]["turn_runner"] is turn_runner
|
|
assert captured["stream"]["session_key"] == "agent:main:standalone:test"
|
|
assert captured["stream"]["message"] == "hello"
|
|
assert captured["stream"]["model"] == "openrouter/test"
|
|
assert captured["stream"]["svc"] is services
|
|
assert captured["stream"]["timeout"] == 7.25
|
|
assert captured["stream"]["tui_output"] is output
|
|
assert captured["stream"]["pending_input_provider"] is pending_provider
|
|
assert captured["scope_after_message"]["model"] == "standalone/after"
|
|
assert captured["state_after_message"].model == "standalone/after"
|
|
assert captured["state_after_message"].usage.input_tokens == 5
|
|
assert captured["state_after_message"].usage.output_tokens == 8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_runtime_matches_exit_with_standalone_surface(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.repl import standalone_runtime
|
|
|
|
class _FakeSessionManager:
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main") -> object:
|
|
return SimpleNamespace(session_key=session_key, agent_id=agent_id)
|
|
|
|
class _FakeServices:
|
|
def __init__(self) -> None:
|
|
self.config = None
|
|
self.session_manager = _FakeSessionManager()
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
surfaces: list[Surface] = []
|
|
|
|
def fake_is_exit_command(value: str, surface: Surface) -> bool:
|
|
surfaces.append(surface)
|
|
return value == "/exit"
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return _FakeServices()
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch,
|
|
) -> None:
|
|
assert surface is Surface.CLI_STANDALONE
|
|
assert await dispatch("/exit") is False
|
|
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.build_turn_runner_from_services",
|
|
lambda _services: object(),
|
|
)
|
|
monkeypatch.setattr(standalone_runtime, "is_exit_command", fake_is_exit_command)
|
|
|
|
deps = standalone_runtime.StandaloneRuntimeDependencies(
|
|
stream_response=lambda *args, **kwargs: None, # type: ignore[arg-type]
|
|
image_command_handler=lambda *args, **kwargs: None, # type: ignore[arg-type]
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
slash_services_factory=lambda _svc: standalone_slash_adapter.StandaloneSlashServices(),
|
|
sync_slash_adapter_io=lambda: None,
|
|
get_tui_output=lambda _scope: None,
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model=None,
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
)
|
|
|
|
assert surfaces == [Surface.CLI_STANDALONE]
|