wehub-resource-sync
1807 lines
67 KiB
Python
1807 lines
67 KiB
Python
"""Unit tests for QwenExecutor (ACP / JSON-RPC 2.0 mode).
|
|
|
|
Tests cover:
|
|
- Executor construction and attribute defaults
|
|
- ACP protocol helpers (_rpc, _notify, _send)
|
|
- Session lifecycle (_ensure_initialized, _ensure_session)
|
|
- run_turn event translation (agent_message_chunk → TextChunk, TurnComplete)
|
|
- run_turn error paths (ACP error response, session-not-found retry reset)
|
|
- Process cleanup (close())
|
|
- Harness registry and alias wiring
|
|
- FastAPI app shape
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from omnigent.inner.executor import ExecutorError, TextChunk, TurnComplete
|
|
from omnigent.inner.qwen_executor import QwenExecutor
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Construction / attribute defaults
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_executor_default_attributes() -> None:
|
|
"""Constructor stores arguments and initialises state correctly."""
|
|
executor = QwenExecutor(qwen_path="qwen")
|
|
assert executor._qwen_path == "qwen"
|
|
assert executor._model is None
|
|
assert executor._proc is None
|
|
assert executor._session_id is None
|
|
assert executor._initialized is False
|
|
assert executor._rpc_id == 0
|
|
|
|
|
|
def test_executor_with_custom_model() -> None:
|
|
"""Custom model is stored on the instance."""
|
|
executor = QwenExecutor(model="qwen/qwen-plus", qwen_path="qwen")
|
|
assert executor._model == "qwen/qwen-plus"
|
|
|
|
|
|
def test_executor_cwd_defaults_to_cwd() -> None:
|
|
"""When no cwd is supplied the executor uses the process cwd."""
|
|
executor = QwenExecutor()
|
|
assert executor._cwd == os.getcwd()
|
|
|
|
|
|
def test_executor_explicit_cwd() -> None:
|
|
"""An explicit cwd is stored as-is."""
|
|
executor = QwenExecutor(cwd="/tmp")
|
|
assert executor._cwd == "/tmp"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# close() with no process
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_with_no_process_is_a_noop() -> None:
|
|
"""close() is safe to call when no subprocess was started."""
|
|
executor = QwenExecutor()
|
|
await executor.close() # must not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# close() with a live process
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_terminates_process() -> None:
|
|
"""close() terminates the subprocess and clears _proc."""
|
|
executor = QwenExecutor()
|
|
|
|
# asyncio.Process.terminate() is synchronous; stdin.close() is sync too.
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdin = MagicMock()
|
|
mock_proc.returncode = None
|
|
|
|
# wait() must be a coroutine.
|
|
async def fake_wait() -> int:
|
|
return 0
|
|
|
|
mock_proc.wait = fake_wait
|
|
executor._proc = mock_proc
|
|
|
|
await executor.close()
|
|
|
|
mock_proc.terminate.assert_called_once()
|
|
assert executor._proc is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_kills_when_terminate_raises() -> None:
|
|
"""close() falls back to kill() if terminate() raises."""
|
|
executor = QwenExecutor()
|
|
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdin = MagicMock()
|
|
mock_proc.terminate.side_effect = OSError("gone")
|
|
mock_proc.returncode = None
|
|
|
|
executor._proc = mock_proc
|
|
|
|
await executor.close() # must not propagate the OSError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _rpc_id increments
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rpc_id_increments_monotonically() -> None:
|
|
"""Each _rpc call uses a unique, incrementing id."""
|
|
executor = QwenExecutor()
|
|
|
|
sent: list[dict] = []
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
sent.append(msg)
|
|
# Immediately resolve the future so _rpc returns.
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result({"jsonrpc": "2.0", "id": msg["id"], "result": {}})
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
await executor._rpc("initialize", {"protocolVersion": 1})
|
|
await executor._rpc("session/new", {"sessionId": "x", "cwd": "/", "mcpServers": []})
|
|
|
|
assert sent[0]["id"] == 1
|
|
assert sent[1]["id"] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _read_stdout — dispatches responses vs notifications
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_stdout_resolves_pending_future() -> None:
|
|
"""_read_stdout resolves the matching _pending future on a response."""
|
|
executor = QwenExecutor()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
fut: asyncio.Future = loop.create_future()
|
|
executor._pending[42] = fut
|
|
|
|
response_line = json.dumps({"jsonrpc": "2.0", "id": 42, "result": {"ok": True}}) + "\n"
|
|
|
|
# Fake stdout that yields one line then EOF (b"" on the second readline).
|
|
mock_stdout = AsyncMock()
|
|
calls = [response_line.encode(), b""]
|
|
mock_stdout.readline = AsyncMock(side_effect=calls)
|
|
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = mock_stdout
|
|
executor._proc = mock_proc
|
|
|
|
# Run reader until it sees EOF (second readline returns b"").
|
|
await executor._read_stdout()
|
|
|
|
assert fut.done()
|
|
assert fut.result()["result"]["ok"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_stdout_puts_notifications_on_queue() -> None:
|
|
"""_read_stdout enqueues notifications (no id) onto the queue."""
|
|
executor = QwenExecutor()
|
|
|
|
notification = {
|
|
"jsonrpc": "2.0",
|
|
"method": "session/update",
|
|
"params": {
|
|
"sessionId": "sess-1",
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "hello"},
|
|
},
|
|
},
|
|
}
|
|
notification_line = json.dumps(notification) + "\n"
|
|
|
|
mock_stdout = AsyncMock()
|
|
mock_stdout.readline = AsyncMock(side_effect=[notification_line.encode(), b""])
|
|
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = mock_stdout
|
|
executor._proc = mock_proc
|
|
|
|
await executor._read_stdout()
|
|
|
|
assert not executor._queue.empty()
|
|
msg = executor._queue.get_nowait()
|
|
assert msg["method"] == "session/update"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_stdout_does_not_resolve_future_for_colliding_request() -> None:
|
|
"""A server request whose id collides with a pending one is queued, not resolved.
|
|
|
|
qwen mints its own request ids; one can equal an outstanding _rpc_id. Such a
|
|
message has a "method", so it must route to the queue, not our future.
|
|
"""
|
|
executor = QwenExecutor()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
fut: asyncio.Future = loop.create_future()
|
|
executor._pending[2] = fut # our pending session/prompt, id=2
|
|
|
|
# qwen sends a permission *request* that happens to reuse id=2.
|
|
request = {
|
|
"jsonrpc": "2.0",
|
|
"id": 2,
|
|
"method": "session/request_permission",
|
|
"params": {"toolCall": {"_meta": {"toolName": "run_shell_command"}}},
|
|
}
|
|
mock_stdout = AsyncMock()
|
|
mock_stdout.readline = AsyncMock(side_effect=[(json.dumps(request) + "\n").encode(), b""])
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = mock_stdout
|
|
executor._proc = mock_proc
|
|
|
|
await executor._read_stdout()
|
|
|
|
# The colliding request must NOT resolve our future with a result; it
|
|
# routes to the queue instead. (The trailing EOF wakes the still-pending
|
|
# future with EOFError so callers fail fast — see the EOF path — so the
|
|
# future may be done, but never with a result.)
|
|
assert fut.exception() is not None and not fut.cancelled()
|
|
assert isinstance(fut.exception(), EOFError)
|
|
assert 2 in executor._pending
|
|
queued = executor._queue.get_nowait()
|
|
assert queued["method"] == "session/request_permission"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_stdout_wakes_pending_futures_on_eof() -> None:
|
|
"""A clean EOF (subprocess crash) wakes in-flight futures with EOFError.
|
|
|
|
Without this, a qwen process that dies mid-turn closes stdout (an EOF, not
|
|
an exception), the reader exits normally, and the pending session/prompt
|
|
future never resolves — run_turn blocks until the idle timeout instead of
|
|
failing fast.
|
|
"""
|
|
executor = QwenExecutor()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
fut: asyncio.Future = loop.create_future()
|
|
executor._pending[7] = fut # in-flight session/prompt
|
|
|
|
# stdout closes immediately (process died) — first readline is EOF.
|
|
mock_stdout = AsyncMock()
|
|
mock_stdout.readline = AsyncMock(side_effect=[b""])
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = mock_stdout
|
|
executor._proc = mock_proc
|
|
|
|
await executor._read_stdout()
|
|
|
|
assert fut.done()
|
|
assert isinstance(fut.exception(), EOFError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _sandbox_launch_path — confine the qwen process tree per os_env.sandbox
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_sandbox_launch_path_bare_when_no_sandbox() -> None:
|
|
"""No os_env, or sandbox type 'none', spawns the bare qwen binary."""
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
|
|
# os_env not provided at all.
|
|
bare = QwenExecutor(qwen_path="/usr/bin/qwen")
|
|
assert bare._sandbox_launch_path(["PATH"]) == "/usr/bin/qwen"
|
|
# os_env present but sandbox explicitly disabled.
|
|
executor = QwenExecutor(
|
|
qwen_path="/usr/bin/qwen",
|
|
os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="none")),
|
|
)
|
|
assert executor._sandbox_launch_path(["PATH"]) == "/usr/bin/qwen"
|
|
|
|
|
|
def test_sandbox_launch_path_wraps_active_policy(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
"""An active sandbox wraps qwen in a launcher with its roots + env baked in.
|
|
|
|
Mirrors pi: the whole qwen process tree is confined, so even an allowed
|
|
tool call can't escape the spec's read/write roots. Asserts the launcher is
|
|
returned and the policy carries qwen's own paths and our spawn env names.
|
|
"""
|
|
from omnigent.inner import sandbox as sandbox_mod
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
from omnigent.inner.sandbox import SandboxPolicy
|
|
|
|
captured: dict = {}
|
|
|
|
def _fake_resolve(_os_env, cwd: Path) -> SandboxPolicy:
|
|
return SandboxPolicy(
|
|
backend_type="linux_bwrap",
|
|
active=True,
|
|
read_roots=[cwd.resolve(strict=False)],
|
|
write_roots=[cwd.resolve(strict=False)],
|
|
write_files=[],
|
|
allow_network=True,
|
|
)
|
|
|
|
def _fake_launcher(target: str, sandbox: SandboxPolicy) -> str:
|
|
captured["target"] = target
|
|
captured["policy"] = sandbox
|
|
return "/fake/launcher"
|
|
|
|
monkeypatch.setattr(sandbox_mod, "resolve_sandbox", _fake_resolve)
|
|
monkeypatch.setattr(sandbox_mod, "create_exec_launcher", _fake_launcher)
|
|
|
|
executor = QwenExecutor(
|
|
cwd=str(tmp_path),
|
|
qwen_path="/usr/bin/qwen",
|
|
os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="linux_bwrap")),
|
|
)
|
|
path = executor._sandbox_launch_path(("PATH", "OPENAI_BASE_URL"))
|
|
|
|
assert path == "/fake/launcher"
|
|
assert captured["target"] == "/usr/bin/qwen"
|
|
policy = captured["policy"]
|
|
# qwen's config dir is a write root so it can start inside the jail.
|
|
assert any(str(p).endswith(".qwen") for p in policy.write_roots)
|
|
# Our deliberate spawn env names are pruneproof in the launcher.
|
|
assert policy.spawn_env_allowlist is not None
|
|
assert "PATH" in policy.spawn_env_allowlist
|
|
assert "OPENAI_BASE_URL" in policy.spawn_env_allowlist
|
|
|
|
|
|
def test_sandbox_launch_path_falls_back_when_backend_unavailable(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path
|
|
) -> None:
|
|
"""A backend failure degrades to the bare binary, never blocks startup."""
|
|
from omnigent.inner import sandbox as sandbox_mod
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
|
|
def _boom(_os_env, _cwd) -> None:
|
|
raise NotImplementedError("no bwrap on this platform")
|
|
|
|
monkeypatch.setattr(sandbox_mod, "resolve_sandbox", _boom)
|
|
|
|
executor = QwenExecutor(
|
|
cwd=str(tmp_path),
|
|
qwen_path="/usr/bin/qwen",
|
|
os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="linux_bwrap")),
|
|
)
|
|
assert executor._sandbox_launch_path(["PATH"]) == "/usr/bin/qwen"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_process_resets_handshake_state(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A (re)start clears the one-way init latch so the fresh process re-handshakes.
|
|
|
|
After a crash the error paths reset session state but ``_initialized`` is a
|
|
one-way latch — without resetting it on restart, ``_ensure_initialized``
|
|
would skip ``initialize`` against the new subprocess and qwen would reject
|
|
the subsequent ``session/new``. ``_image_supported`` (derived from the init
|
|
response) is stale for the same reason.
|
|
"""
|
|
executor = QwenExecutor(qwen_path="/usr/bin/qwen")
|
|
# Simulate the post-crash state: handshake flags left latched from the
|
|
# previous (now-dead) subprocess.
|
|
executor._initialized = True
|
|
executor._image_supported = True
|
|
|
|
async def _fake_subprocess_exec(*_args, **_kwargs):
|
|
proc = MagicMock()
|
|
proc.stdout = AsyncMock()
|
|
proc.stdout.readline = AsyncMock(return_value=b"")
|
|
proc.stderr = AsyncMock()
|
|
proc.stderr.readline = AsyncMock(return_value=b"")
|
|
return proc
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_subprocess_exec)
|
|
|
|
await executor._start_process()
|
|
|
|
assert executor._initialized is False
|
|
assert executor._image_supported is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _ensure_session resets on "Session not found"
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_session_uses_server_assigned_id() -> None:
|
|
"""_ensure_session stores the sessionId from the server response, not ours."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True # skip initialize
|
|
|
|
server_session_id = "server-assigned-uuid"
|
|
|
|
async def fake_rpc(method: str, params: dict, timeout: float = 30.0) -> dict:
|
|
if method == "session/new":
|
|
return {"jsonrpc": "2.0", "id": 1, "result": {"sessionId": server_session_id}}
|
|
return {"jsonrpc": "2.0", "id": 1, "result": {}}
|
|
|
|
executor._rpc = fake_rpc # type: ignore[method-assign]
|
|
|
|
sid = await executor._ensure_session()
|
|
assert sid == server_session_id
|
|
assert executor._session_id == server_session_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_session_cached_after_first_call() -> None:
|
|
"""_ensure_session does not make a second RPC call once session is set."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "cached-sid"
|
|
|
|
rpc_calls: list[str] = []
|
|
|
|
async def fake_rpc(method: str, params: dict, timeout: float = 30.0) -> dict:
|
|
rpc_calls.append(method)
|
|
return {"result": {}}
|
|
|
|
executor._rpc = fake_rpc # type: ignore[method-assign]
|
|
|
|
sid = await executor._ensure_session()
|
|
assert sid == "cached-sid"
|
|
assert rpc_calls == [] # no RPC was made
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_turn — success path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_yields_text_chunks_and_turn_complete() -> None:
|
|
"""run_turn yields TextChunk events for agent_message_chunk notifications
|
|
and a TurnComplete when the session/prompt response arrives.
|
|
|
|
The fake_send callback:
|
|
1. Enqueues the streaming notification immediately (so the event loop
|
|
processes it before checking fut.done()).
|
|
2. Schedules the future resolution on the *next* event-loop iteration
|
|
via ``loop.call_soon`` so the notification is always consumed first.
|
|
"""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-abc"
|
|
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
session_id = executor._session_id
|
|
loop = asyncio.get_event_loop()
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
req_id = msg["id"]
|
|
# 1. Put the streaming notification on the queue first.
|
|
notification = {
|
|
"jsonrpc": "2.0",
|
|
"method": "session/update",
|
|
"params": {
|
|
"sessionId": session_id,
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "Hello!"},
|
|
},
|
|
},
|
|
}
|
|
await executor._queue.put(notification)
|
|
|
|
# 2. Resolve the future on the next loop iteration so the
|
|
# notification is consumed before fut.done() is True.
|
|
def _resolve() -> None:
|
|
fut = executor._pending.get(req_id)
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {"stopReason": "end_turn"},
|
|
}
|
|
)
|
|
|
|
loop.call_soon(_resolve)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
messages = [{"role": "user", "content": "say hi"}]
|
|
events = []
|
|
async for event in executor.run_turn(messages, [], "Be helpful"):
|
|
events.append(event)
|
|
|
|
text_chunks = [e for e in events if isinstance(e, TextChunk)]
|
|
turn_completes = [e for e in events if isinstance(e, TurnComplete)]
|
|
|
|
assert len(text_chunks) == 1
|
|
assert text_chunks[0].text == "Hello!"
|
|
assert len(turn_completes) == 1
|
|
assert turn_completes[0].response == "Hello!"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token usage — parsed from qwen's _meta.usage and emitted on TurnComplete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_accumulate_usage_maps_and_splits_cached() -> None:
|
|
"""_meta.usage maps to wire keys; cached tokens split out of input_tokens.
|
|
|
|
qwen's inputTokens (Gemini promptTokenCount) is inclusive of cached tokens,
|
|
but compute_llm_cost wants the non-cached portion in input_tokens.
|
|
"""
|
|
acc: dict[str, int] = {}
|
|
QwenExecutor._accumulate_usage(
|
|
acc,
|
|
{
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": ""},
|
|
"_meta": {
|
|
"usage": {
|
|
"inputTokens": 1000,
|
|
"outputTokens": 200,
|
|
"totalTokens": 1200,
|
|
"cachedReadTokens": 300,
|
|
"thoughtTokens": 50,
|
|
}
|
|
},
|
|
},
|
|
)
|
|
assert acc == {
|
|
"input_tokens": 700, # 1000 - 300 cached
|
|
"output_tokens": 200,
|
|
"total_tokens": 1200,
|
|
"cache_read_input_tokens": 300,
|
|
}
|
|
|
|
|
|
def test_accumulate_usage_sums_across_calls_and_ignores_non_usage() -> None:
|
|
"""Multiple emissions sum; updates without _meta.usage are ignored."""
|
|
acc: dict[str, int] = {}
|
|
# A plain text chunk carries no usage — must not perturb the accumulator.
|
|
QwenExecutor._accumulate_usage(acc, {"content": {"type": "text", "text": "hi"}})
|
|
assert acc == {}
|
|
for _ in range(2):
|
|
QwenExecutor._accumulate_usage(
|
|
acc,
|
|
{"_meta": {"usage": {"inputTokens": 100, "outputTokens": 10, "totalTokens": 110}}},
|
|
)
|
|
# Summed across the two calls; no cached key when cachedReadTokens absent.
|
|
assert acc == {"input_tokens": 200, "output_tokens": 20, "total_tokens": 220}
|
|
|
|
|
|
def test_accumulate_usage_clamps_negative_input() -> None:
|
|
"""A malformed cached > input never drives input_tokens negative."""
|
|
acc: dict[str, int] = {}
|
|
QwenExecutor._accumulate_usage(
|
|
acc,
|
|
{"_meta": {"usage": {"inputTokens": 100, "cachedReadTokens": 500, "totalTokens": 100}}},
|
|
)
|
|
assert acc["input_tokens"] == 0
|
|
assert acc["cache_read_input_tokens"] == 500
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_emits_usage_on_turn_complete(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A usage-bearing chunk surfaces as TurnComplete.usage and notifies cost."""
|
|
notified: list[dict] = []
|
|
monkeypatch.setattr(
|
|
"omnigent.inner.qwen_executor._notify_usage_from_dict",
|
|
lambda model, usage: notified.append({"model": model, "usage": usage}),
|
|
)
|
|
|
|
executor = QwenExecutor(model="qwen/qwen3-coder")
|
|
executor._initialized = True
|
|
executor._session_id = "sess-usage"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
session_id = executor._session_id
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
base = {
|
|
"jsonrpc": "2.0",
|
|
"method": "session/update",
|
|
"params": {"sessionId": session_id},
|
|
}
|
|
# 1. Real text chunk.
|
|
await executor._queue.put(
|
|
{
|
|
**base,
|
|
"params": {
|
|
"sessionId": session_id,
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "done"},
|
|
},
|
|
},
|
|
}
|
|
)
|
|
# 2. Usage-bearing chunk (empty text + _meta.usage).
|
|
await executor._queue.put(
|
|
{
|
|
**base,
|
|
"params": {
|
|
"sessionId": session_id,
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": ""},
|
|
"_meta": {
|
|
"usage": {
|
|
"inputTokens": 500,
|
|
"outputTokens": 80,
|
|
"totalTokens": 580,
|
|
"cachedReadTokens": 100,
|
|
}
|
|
},
|
|
},
|
|
},
|
|
}
|
|
)
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": msg["id"], "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
events = [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "")]
|
|
completes = [e for e in events if isinstance(e, TurnComplete)]
|
|
assert len(completes) == 1
|
|
assert completes[0].response == "done"
|
|
assert completes[0].usage == {
|
|
"input_tokens": 400, # 500 - 100 cached
|
|
"output_tokens": 80,
|
|
"total_tokens": 580,
|
|
"cache_read_input_tokens": 100,
|
|
}
|
|
# Cost observer was notified with the same usage + the configured model.
|
|
assert notified == [{"model": "qwen/qwen3-coder", "usage": completes[0].usage}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_usage_none_when_unreported(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""No usage chunk → TurnComplete.usage is None and the observer isn't called."""
|
|
notified: list[dict] = []
|
|
monkeypatch.setattr(
|
|
"omnigent.inner.qwen_executor._notify_usage_from_dict",
|
|
lambda model, usage: notified.append({"model": model, "usage": usage}),
|
|
)
|
|
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-no-usage"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
session_id = executor._session_id
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
await executor._queue.put(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "session/update",
|
|
"params": {
|
|
"sessionId": session_id,
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": "hi"},
|
|
},
|
|
},
|
|
}
|
|
)
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": msg["id"], "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
events = [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "")]
|
|
completes = [e for e in events if isinstance(e, TurnComplete)]
|
|
assert len(completes) == 1
|
|
assert completes[0].usage is None
|
|
assert notified == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_drains_all_chunks_before_completing() -> None:
|
|
"""All buffered chunks are yielded even if the future resolves first.
|
|
|
|
The reader can enqueue several chunks AND resolve the prompt future before
|
|
run_turn drains the queue. Completion is gated on an empty queue, so no
|
|
chunk is lost to a premature return.
|
|
"""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-drain"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
session_id = executor._session_id
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
for piece in ("a", "b", "c"):
|
|
await executor._queue.put(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "session/update",
|
|
"params": {
|
|
"sessionId": session_id,
|
|
"update": {
|
|
"sessionUpdate": "agent_message_chunk",
|
|
"content": {"type": "text", "text": piece},
|
|
},
|
|
},
|
|
}
|
|
)
|
|
# Resolve immediately — chunks are still buffered in the queue.
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": msg["id"], "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
events = [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "")]
|
|
text = "".join(e.text for e in events if isinstance(e, TextChunk))
|
|
completes = [e for e in events if isinstance(e, TurnComplete)]
|
|
assert text == "abc"
|
|
assert len(completes) == 1
|
|
assert completes[0].response == "abc"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_approval_does_not_count_against_idle_timeout(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A slow human approval must not trip the response timeout.
|
|
|
|
With a tiny timeout and an elicitation handler slower than it, the turn must
|
|
still complete: the idle deadline resets after the approval round-trip.
|
|
"""
|
|
monkeypatch.setattr("omnigent.inner.qwen_executor._PROMPT_TIMEOUT_SECONDS", 0.05)
|
|
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-slow"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
async def slow_handler(tool_name: str, tool_input: dict) -> bool:
|
|
await asyncio.sleep(0.25) # 5x the timeout
|
|
return True
|
|
|
|
executor._elicitation_handler = slow_handler # type: ignore[assignment]
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
# qwen asks permission before finishing the turn.
|
|
await executor._queue.put(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 999,
|
|
"method": "session/request_permission",
|
|
"params": {
|
|
"toolCall": {"_meta": {"toolName": "run_shell_command"}},
|
|
"options": [{"kind": "allow_once", "optionId": "ok"}],
|
|
},
|
|
}
|
|
)
|
|
elif "result" in msg and msg.get("id") == 999:
|
|
# Our approval reply went out — now qwen completes the prompt.
|
|
rid = executor._rpc_id
|
|
fut = executor._pending.get(rid)
|
|
if fut and not fut.done():
|
|
fut.set_result({"jsonrpc": "2.0", "id": rid, "result": {"stopReason": "end_turn"}})
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
events = [e async for e in executor.run_turn([{"role": "user", "content": "rm it"}], [], "")]
|
|
assert any(isinstance(e, TurnComplete) for e in events)
|
|
assert not any(isinstance(e, ExecutorError) for e in events)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_turn — ACP error response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_yields_executor_error_on_acp_error() -> None:
|
|
"""run_turn yields ExecutorError when session/prompt returns an error."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-err"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": msg["id"],
|
|
"error": {"code": -32603, "message": "Something went wrong"},
|
|
}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
messages = [{"role": "user", "content": "hi"}]
|
|
events = []
|
|
async for event in executor.run_turn(messages, [], ""):
|
|
events.append(event)
|
|
|
|
assert len(events) == 1
|
|
assert isinstance(events[0], ExecutorError)
|
|
assert "Something went wrong" in events[0].message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_resets_session_on_not_found_error() -> None:
|
|
"""run_turn clears _session_id when ACP reports Session not found."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "stale-sess"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
fut = executor._pending.get(msg["id"])
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": msg["id"],
|
|
"error": {
|
|
"code": -32603,
|
|
"message": "Session not found: stale-sess",
|
|
},
|
|
}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
async for _ in executor.run_turn([{"role": "user", "content": "hi"}], [], ""):
|
|
pass
|
|
|
|
# Session id should be reset so next turn creates a fresh session.
|
|
assert executor._session_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Harness registry / alias wiring
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_qwen_in_harness_registry() -> None:
|
|
"""'qwen' must be in the _HARNESS_MODULES dispatch table."""
|
|
from omnigent.runtime.harnesses import _HARNESS_MODULES
|
|
|
|
assert "qwen" in _HARNESS_MODULES
|
|
|
|
|
|
def test_qwen_in_harness_allowlist() -> None:
|
|
"""'qwen' must be in OMNIGENT_HARNESSES."""
|
|
from omnigent.spec._omnigent_compat import OMNIGENT_HARNESSES
|
|
|
|
assert "qwen" in OMNIGENT_HARNESSES
|
|
|
|
|
|
def test_qwen_code_alias_resolves_to_qwen() -> None:
|
|
"""'qwen-code' alias maps to the canonical 'qwen' harness id."""
|
|
from omnigent.harness_aliases import canonicalize_harness
|
|
|
|
assert canonicalize_harness("qwen-code") == "qwen"
|
|
|
|
|
|
def test_qwen_code_in_harness_aliases() -> None:
|
|
"""'qwen-code' must be in OMNIGENT_HARNESS_ALIASES."""
|
|
from omnigent.spec._omnigent_compat import OMNIGENT_HARNESS_ALIASES
|
|
|
|
assert "qwen-code" in OMNIGENT_HARNESS_ALIASES
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FastAPI app shape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_qwen_harness_creates_fastapi_app() -> None:
|
|
"""create_app() returns a FastAPI app with at least a /health route."""
|
|
from omnigent.inner.qwen_harness import create_app
|
|
|
|
app = create_app()
|
|
assert app is not None
|
|
assert hasattr(app, "routes")
|
|
health_routes = [r for r in app.routes if hasattr(r, "path") and "/health" in r.path]
|
|
assert len(health_routes) > 0
|
|
|
|
|
|
def test_qwen_harness_module_importable() -> None:
|
|
"""qwen_harness can be imported and exposes create_app."""
|
|
from omnigent.inner import qwen_harness
|
|
|
|
assert hasattr(qwen_harness, "create_app")
|
|
|
|
|
|
def test_wrap_passes_gateway_env_to_executor(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""_build_qwen_executor threads HARNESS_QWEN_GATEWAY_* into the executor."""
|
|
from omnigent.inner import qwen_harness
|
|
|
|
monkeypatch.setenv("HARNESS_QWEN_MODEL", "qwen/qwen3-coder")
|
|
monkeypatch.setenv("HARNESS_QWEN_GATEWAY_BASE_URL", "https://gw.example/v1")
|
|
monkeypatch.setenv("HARNESS_QWEN_GATEWAY_AUTH_COMMAND", "printf '%s' sk-x")
|
|
|
|
executor = qwen_harness._build_qwen_executor()
|
|
assert isinstance(executor, QwenExecutor)
|
|
assert executor._gateway_base_url == "https://gw.example/v1"
|
|
assert executor._gateway_auth_command == "printf '%s' sk-x"
|
|
|
|
|
|
def test_wrap_gateway_env_absent_leaves_executor_ungated(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Without the gateway env vars, the executor has no gateway config."""
|
|
from omnigent.inner import qwen_harness
|
|
|
|
monkeypatch.delenv("HARNESS_QWEN_GATEWAY_BASE_URL", raising=False)
|
|
monkeypatch.delenv("HARNESS_QWEN_GATEWAY_AUTH_COMMAND", raising=False)
|
|
|
|
executor = qwen_harness._build_qwen_executor()
|
|
assert isinstance(executor, QwenExecutor)
|
|
assert executor._gateway_base_url is None
|
|
assert executor._gateway_auth_command is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# close_session is a no-op (sessions are per-process)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_session_is_noop() -> None:
|
|
"""close_session() does nothing and does not raise."""
|
|
executor = QwenExecutor()
|
|
await executor.close_session("some-key") # must not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# system_prompt folded into the first turn (ACP has no system field)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_prepends_system_prompt_on_first_turn() -> None:
|
|
"""The system prompt is prepended to the first user turn's text.
|
|
|
|
ACP has no dedicated system-prompt field, so the spec ``prompt:`` would
|
|
otherwise never reach qwen. The second turn must NOT repeat it (the
|
|
session retains context).
|
|
"""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-sp"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
loop = asyncio.get_event_loop()
|
|
|
|
sent_prompts: list[str] = []
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
sent_prompts.append(msg["params"]["prompt"][0]["text"])
|
|
req_id = msg["id"]
|
|
|
|
def _resolve() -> None:
|
|
fut = executor._pending.get(req_id)
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
loop.call_soon(_resolve)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
messages = [{"role": "user", "content": "first"}]
|
|
async for _ in executor.run_turn(messages, [], "SYSTEM RULES"):
|
|
pass
|
|
async for _ in executor.run_turn([{"role": "user", "content": "second"}], [], "SYSTEM RULES"):
|
|
pass
|
|
|
|
assert sent_prompts[0] == "SYSTEM RULES\n\nfirst"
|
|
assert sent_prompts[1] == "second" # not repeated
|
|
assert executor._system_prompt_sent is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_resends_system_prompt_after_session_reset() -> None:
|
|
"""After a 'Session not found' reset, the next turn re-folds the system prompt.
|
|
|
|
Losing the session clears ``_system_prompt_sent`` so the fresh session
|
|
receives the spec prompt again (qwen no longer holds the earlier context).
|
|
"""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-1"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
|
|
sent_prompts: list[str] = []
|
|
fail_next = {"flag": True} # turn 1's prompt is rejected with "Session not found"
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
method = msg.get("method")
|
|
fut = executor._pending.get(msg["id"])
|
|
if method == "session/new":
|
|
# Fresh session created after the reset.
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": msg["id"], "result": {"sessionId": "sess-2"}}
|
|
)
|
|
elif method == "session/prompt":
|
|
sent_prompts.append(msg["params"]["prompt"][0]["text"])
|
|
if fut and not fut.done():
|
|
if fail_next["flag"]:
|
|
fail_next["flag"] = False
|
|
fut.set_result(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": msg["id"],
|
|
"error": {"code": -32603, "message": "Session not found: sess-1"},
|
|
}
|
|
)
|
|
else:
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": msg["id"], "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
# Turn 1: folds the prompt, then errors with "Session not found" → reset.
|
|
async for _ in executor.run_turn([{"role": "user", "content": "first"}], [], "SYSTEM"):
|
|
pass
|
|
# Turn 2: fresh session — the system prompt must be re-folded.
|
|
async for _ in executor.run_turn([{"role": "user", "content": "second"}], [], "SYSTEM"):
|
|
pass
|
|
|
|
assert sent_prompts[0] == "SYSTEM\n\nfirst" # turn 1 folded
|
|
assert sent_prompts[1] == "SYSTEM\n\nsecond" # re-folded into the new session
|
|
assert executor._session_id == "sess-2"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# History replay on a fresh session (model switch / reset doesn't drop context)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_history_prefix_serializes_prior_turns() -> None:
|
|
"""_history_prefix renders prior turns as labeled role: content lines."""
|
|
prior = [
|
|
{"role": "user", "content": "what is 2+2"},
|
|
{"role": "assistant", "content": "4"},
|
|
{"role": "user", "content": [{"type": "input_text", "text": "and 3+3"}]},
|
|
{"role": "assistant", "content": [{"type": "output_text", "text": "6"}]},
|
|
]
|
|
out = QwenExecutor._history_prefix(prior)
|
|
assert out.startswith("Conversation so far:")
|
|
assert "user: what is 2+2" in out
|
|
assert "assistant: 4" in out
|
|
assert "user: and 3+3" in out # list content folded via _text_from_blocks
|
|
assert "assistant: 6" in out
|
|
assert out.rstrip().endswith("using the conversation above as context.")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_replays_history_on_fresh_session() -> None:
|
|
"""A fresh session folds prior turns into the prompt (e.g. after /model respawn).
|
|
|
|
qwen normally only sees the latest user turn; on a brand-new subprocess
|
|
(a ``/model`` switch respawns it) that would drop everything before the
|
|
switch. The first turn must replay the transcript so context survives.
|
|
"""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-fresh" # session exists, but no prior turns sent
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
loop = asyncio.get_event_loop()
|
|
|
|
sent_prompts: list[str] = []
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
sent_prompts.append(msg["params"]["prompt"][0]["text"])
|
|
req_id = msg["id"]
|
|
|
|
def _resolve() -> None:
|
|
fut = executor._pending.get(req_id)
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
loop.call_soon(_resolve)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
messages = [
|
|
{"role": "user", "content": "remember the number 42"},
|
|
{"role": "assistant", "content": "Got it, 42."},
|
|
{"role": "user", "content": "what number did I say?"},
|
|
]
|
|
async for _ in executor.run_turn(messages, [], "SYS"):
|
|
pass
|
|
|
|
prompt = sent_prompts[0]
|
|
# System prompt first, then replayed history, then the latest turn.
|
|
assert prompt.startswith("SYS\n\n")
|
|
assert "Conversation so far:" in prompt
|
|
assert "user: remember the number 42" in prompt
|
|
assert "assistant: Got it, 42." in prompt
|
|
assert prompt.rstrip().endswith("user: what number did I say?")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_no_replay_on_continuing_session() -> None:
|
|
"""A continuing session sends only the latest turn (qwen retains context)."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-cont"
|
|
executor._system_prompt_sent = True # not a fresh session
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
loop = asyncio.get_event_loop()
|
|
|
|
sent_prompts: list[str] = []
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
sent_prompts.append(msg["params"]["prompt"][0]["text"])
|
|
req_id = msg["id"]
|
|
|
|
def _resolve() -> None:
|
|
fut = executor._pending.get(req_id)
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
loop.call_soon(_resolve)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
messages = [
|
|
{"role": "user", "content": "earlier turn"},
|
|
{"role": "assistant", "content": "ok"},
|
|
{"role": "user", "content": "latest turn"},
|
|
]
|
|
async for _ in executor.run_turn(messages, [], "SYS"):
|
|
pass
|
|
|
|
# No history prefix, no system-prompt re-fold — just the latest message.
|
|
assert sent_prompts[0] == "latest turn"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_no_replay_on_genuine_first_turn() -> None:
|
|
"""A brand-new conversation (single user turn) has nothing to replay."""
|
|
executor = QwenExecutor()
|
|
executor._initialized = True
|
|
executor._session_id = "sess-first"
|
|
executor._proc = MagicMock()
|
|
executor._proc.returncode = None
|
|
loop = asyncio.get_event_loop()
|
|
|
|
sent_prompts: list[str] = []
|
|
|
|
async def fake_send(msg: dict) -> None:
|
|
if msg.get("method") == "session/prompt":
|
|
sent_prompts.append(msg["params"]["prompt"][0]["text"])
|
|
req_id = msg["id"]
|
|
|
|
def _resolve() -> None:
|
|
fut = executor._pending.get(req_id)
|
|
if fut and not fut.done():
|
|
fut.set_result(
|
|
{"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
|
|
)
|
|
|
|
loop.call_soon(_resolve)
|
|
|
|
executor._send = fake_send # type: ignore[method-assign]
|
|
|
|
async for _ in executor.run_turn([{"role": "user", "content": "hello"}], [], "SYS"):
|
|
pass
|
|
|
|
assert sent_prompts[0] == "SYS\n\nhello" # system prompt only, no replay
|
|
assert "Conversation so far:" not in sent_prompts[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Server-initiated requests: permission, unsupported methods (incl. fs/*)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_to_fs_read_text_file_unsupported_without_os_env() -> None:
|
|
"""fs/* is method-not-found when fs delegation isn't advertised.
|
|
|
|
With no os_env configured, ``_fs_delegation`` is False and we never
|
|
advertise ``clientCapabilities.fs``, so qwen uses its own file tools. A
|
|
stray ``fs/read_text_file`` must get a JSON-RPC method-not-found error
|
|
rather than a fabricated (and dangerous) empty/real-file response.
|
|
"""
|
|
executor = QwenExecutor() # no os_env → delegation off
|
|
assert executor._fs_delegation is False
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 7, "method": "fs/read_text_file", "params": {"path": "/x"}}
|
|
)
|
|
|
|
assert sent[0]["id"] == 7
|
|
assert sent[0]["error"]["code"] == -32601
|
|
assert "result" not in sent[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Filesystem delegation (fs/read_text_file, fs/write_text_file)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeOSEnv:
|
|
"""Minimal OSEnvironment stand-in capturing read/write calls."""
|
|
|
|
def __init__(self, read_result: dict | None = None, write_result: dict | None = None) -> None:
|
|
self._read_result = read_result if read_result is not None else {}
|
|
self._write_result = write_result if write_result is not None else {}
|
|
self.read_calls: list[tuple] = []
|
|
self.write_calls: list[tuple] = []
|
|
self.closed = False
|
|
|
|
async def read(self, path: str, offset: int = 1, limit: int | None = None) -> dict:
|
|
self.read_calls.append((path, offset, limit))
|
|
return self._read_result
|
|
|
|
async def write(self, path: str, content: str) -> dict:
|
|
self.write_calls.append((path, content))
|
|
return self._write_result
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def test_fs_delegation_flag_tracks_os_env() -> None:
|
|
"""Delegation is on with an os_env, off without one or for a fork env."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
assert QwenExecutor()._fs_delegation is False # no os_env
|
|
assert QwenExecutor(os_env=OSEnvSpec(type="caller_process"))._fs_delegation is True
|
|
# A fork env operates on a copied tree → path would diverge from the qwen
|
|
# subprocess cwd, so delegation must stay off.
|
|
assert QwenExecutor(os_env=OSEnvSpec(type="caller_process", fork=True))._fs_delegation is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_advertises_fs_capability_per_delegation() -> None:
|
|
"""initialize advertises clientCapabilities.fs matching the delegation flag."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
init_result = {"result": {"agentCapabilities": {"promptCapabilities": {}}}}
|
|
|
|
# Delegation ON (os_env configured).
|
|
on = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
on._rpc = AsyncMock(return_value=init_result) # type: ignore[method-assign]
|
|
await on._ensure_initialized()
|
|
caps = on._rpc.call_args.args[1]["clientCapabilities"]["fs"]
|
|
assert caps == {"readTextFile": True, "writeTextFile": True}
|
|
|
|
# Delegation OFF (no os_env).
|
|
off = QwenExecutor()
|
|
off._rpc = AsyncMock(return_value=init_result) # type: ignore[method-assign]
|
|
await off._ensure_initialized()
|
|
caps = off._rpc.call_args.args[1]["clientCapabilities"]["fs"]
|
|
assert caps == {"readTextFile": False, "writeTextFile": False}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_read_returns_content_and_maps_window() -> None:
|
|
"""fs/read_text_file reads through the OSEnvironment; line/limit → offset/limit."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
fake = _FakeOSEnv(read_result={"content": "hello\nworld\n", "encoding": "utf-8"})
|
|
executor._os_environment = fake # type: ignore[assignment]
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 3,
|
|
"method": "fs/read_text_file",
|
|
"params": {"path": "a.txt", "line": 2, "limit": 5},
|
|
}
|
|
)
|
|
|
|
assert sent[0]["result"] == {"content": "hello\nworld\n"}
|
|
assert fake.read_calls == [("a.txt", 2, 5)] # line→offset, limit→limit
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_read_whole_file_when_no_window() -> None:
|
|
"""Absent line/limit reads the whole file (offset=1, limit=None)."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
fake = _FakeOSEnv(read_result={"content": "x", "encoding": "utf-8"})
|
|
executor._os_environment = fake # type: ignore[assignment]
|
|
executor._send = AsyncMock() # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 1, "method": "fs/read_text_file", "params": {"path": "a.txt"}}
|
|
)
|
|
|
|
assert fake.read_calls == [("a.txt", 1, None)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_read_missing_file_maps_to_enoent() -> None:
|
|
"""A 'no such file' read error maps to qwen's ENOENT code (-32002)."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
executor._os_environment = _FakeOSEnv( # type: ignore[assignment]
|
|
read_result={"error": "[Errno 2] No such file or directory: 'gone.txt'"}
|
|
)
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 5, "method": "fs/read_text_file", "params": {"path": "gone.txt"}}
|
|
)
|
|
|
|
assert sent[0]["error"]["code"] == -32002
|
|
assert "result" not in sent[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_read_binary_file_is_rejected() -> None:
|
|
"""A non-utf-8 (binary) file is refused rather than returned as bytes."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
executor._os_environment = _FakeOSEnv( # type: ignore[assignment]
|
|
read_result={"content": "AAAA", "encoding": "base64", "total_bytes": 3}
|
|
)
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 6, "method": "fs/read_text_file", "params": {"path": "img.png"}}
|
|
)
|
|
|
|
assert sent[0]["error"]["code"] == -32603
|
|
assert "text file" in sent[0]["error"]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_write_writes_through_os_env() -> None:
|
|
"""fs/write_text_file writes via the OSEnvironment and returns an empty result."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
fake = _FakeOSEnv(write_result={"path": "out.txt", "bytes": 3})
|
|
executor._os_environment = fake # type: ignore[assignment]
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 8,
|
|
"method": "fs/write_text_file",
|
|
"params": {"path": "out.txt", "content": "abc"},
|
|
}
|
|
)
|
|
|
|
assert sent[0]["result"] == {}
|
|
assert fake.write_calls == [("out.txt", "abc")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_write_error_surfaces_as_internal_error() -> None:
|
|
"""A write failure surfaces as a JSON-RPC internal error (-32603)."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
executor._os_environment = _FakeOSEnv( # type: ignore[assignment]
|
|
write_result={"error": "Permission denied"}
|
|
)
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"id": 9,
|
|
"method": "fs/write_text_file",
|
|
"params": {"path": "out.txt", "content": "abc"},
|
|
}
|
|
)
|
|
|
|
assert sent[0]["error"]["code"] == -32603
|
|
assert "Permission denied" in sent[0]["error"]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fs_write_rejects_missing_args() -> None:
|
|
"""Missing path / non-string content is an invalid-params error (-32602)."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
executor._os_environment = _FakeOSEnv() # type: ignore[assignment]
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 10, "method": "fs/write_text_file", "params": {"path": "out.txt"}}
|
|
)
|
|
|
|
assert sent[0]["error"]["code"] == -32602
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_releases_fs_os_environment() -> None:
|
|
"""close() tears down a lazily-created fs-delegation OSEnvironment."""
|
|
from omnigent.inner.datamodel import OSEnvSpec
|
|
|
|
executor = QwenExecutor(os_env=OSEnvSpec(type="caller_process"))
|
|
fake = _FakeOSEnv()
|
|
executor._os_environment = fake # type: ignore[assignment]
|
|
|
|
await executor.close()
|
|
|
|
assert fake.closed is True
|
|
assert executor._os_environment is None
|
|
|
|
|
|
# Realistic qwen session/request_permission payload (from the ACP probe).
|
|
def _perm_request(req_id: int = 9) -> dict:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"method": "session/request_permission",
|
|
"params": {
|
|
"sessionId": "s",
|
|
"options": [
|
|
{"optionId": "proceed_always_project", "kind": "allow_always"},
|
|
{"optionId": "proceed_once", "kind": "allow_once"},
|
|
{"optionId": "cancel", "kind": "reject_once"},
|
|
],
|
|
"toolCall": {
|
|
"kind": "execute",
|
|
"rawInput": {"command": "rm -f victim.txt"},
|
|
"_meta": {"toolName": "run_shell_command"},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_to_permission_allows_when_no_gates_wired() -> None:
|
|
"""With no policy/elicitation bridge wired, permission falls back to allow.
|
|
|
|
Prefers the once-scoped grant (``allow_once``), never ``allow_always``.
|
|
"""
|
|
executor = QwenExecutor() # no _policy_evaluator / _elicitation_handler
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(_perm_request())
|
|
|
|
assert sent[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "proceed_once"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_to_permission_denied_by_policy() -> None:
|
|
"""A POLICY_ACTION_DENY verdict selects a reject option — no elicitation."""
|
|
executor = QwenExecutor()
|
|
executor._policy_evaluator = AsyncMock( # type: ignore[attr-defined]
|
|
return_value=MagicMock(action="POLICY_ACTION_DENY")
|
|
)
|
|
executor._elicitation_handler = AsyncMock(return_value=True) # type: ignore[attr-defined]
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(_perm_request())
|
|
|
|
assert sent[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "cancel"}
|
|
executor._elicitation_handler.assert_not_called() # policy DENY short-circuits
|
|
# Policy saw the real tool name + args extracted from the payload.
|
|
phase, data = executor._policy_evaluator.call_args.args
|
|
assert phase == "PHASE_TOOL_CALL"
|
|
assert data == {"name": "run_shell_command", "arguments": {"command": "rm -f victim.txt"}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_to_permission_elicitation_allow_and_deny() -> None:
|
|
"""With only elicitation wired, the user's accept/deny maps to allow/reject."""
|
|
# Accept → allow_once.
|
|
allow_exec = QwenExecutor()
|
|
allow_exec._elicitation_handler = AsyncMock(return_value=True) # type: ignore[attr-defined]
|
|
sent_a: list[dict] = []
|
|
allow_exec._send = AsyncMock(side_effect=lambda m: sent_a.append(m)) # type: ignore[method-assign]
|
|
await allow_exec._respond_to_agent_request(_perm_request())
|
|
assert sent_a[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "proceed_once"}
|
|
allow_exec._elicitation_handler.assert_awaited_once_with(
|
|
"run_shell_command", {"command": "rm -f victim.txt"}
|
|
)
|
|
|
|
# Deny → reject_once.
|
|
deny_exec = QwenExecutor()
|
|
deny_exec._elicitation_handler = AsyncMock(return_value=False) # type: ignore[attr-defined]
|
|
sent_d: list[dict] = []
|
|
deny_exec._send = AsyncMock(side_effect=lambda m: sent_d.append(m)) # type: ignore[method-assign]
|
|
await deny_exec._respond_to_agent_request(_perm_request())
|
|
assert sent_d[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "cancel"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_to_unknown_method_returns_jsonrpc_error() -> None:
|
|
"""An unsupported server request yields a method-not-found error, not {}."""
|
|
executor = QwenExecutor()
|
|
sent: list[dict] = []
|
|
executor._send = AsyncMock(side_effect=lambda m: sent.append(m)) # type: ignore[method-assign]
|
|
|
|
await executor._respond_to_agent_request(
|
|
{"jsonrpc": "2.0", "id": 11, "method": "terminal/create", "params": {}}
|
|
)
|
|
|
|
assert sent[0]["error"]["code"] == -32601
|
|
assert "result" not in sent[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# stderr is drained so a chatty CLI can't wedge the pipe
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_stderr_drains_until_eof() -> None:
|
|
"""_read_stderr consumes lines and exits cleanly on EOF."""
|
|
executor = QwenExecutor()
|
|
mock_stderr = AsyncMock()
|
|
mock_stderr.readline = AsyncMock(side_effect=[b"warn: something\n", b""])
|
|
mock_proc = MagicMock()
|
|
mock_proc.stderr = mock_stderr
|
|
executor._proc = mock_proc
|
|
|
|
await executor._read_stderr() # must terminate at EOF, not hang
|
|
|
|
assert mock_stderr.readline.await_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Missing-binary path surfaces a clear error on first turn
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_turn_missing_binary_yields_retryable_error() -> None:
|
|
"""A non-existent qwen binary surfaces as an ExecutorError, not a crash."""
|
|
executor = QwenExecutor(qwen_path="/nonexistent/qwen-binary-xyz")
|
|
|
|
events = []
|
|
async for event in executor.run_turn([{"role": "user", "content": "hi"}], [], "be helpful"):
|
|
events.append(event)
|
|
|
|
assert len(events) == 1
|
|
assert isinstance(events[0], ExecutorError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Content-block extraction (file attachments must not drop the message)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_text_from_blocks_recovers_input_text_and_marks_file() -> None:
|
|
"""A message with input_text + input_file keeps the text and notes the file.
|
|
|
|
Regression: the old ``type == "text"`` filter matched neither ``input_text``
|
|
nor ``input_file``, so attaching a file dropped the ENTIRE message.
|
|
"""
|
|
blocks = [
|
|
{"type": "input_text", "text": "review this"},
|
|
{"type": "input_file", "file_id": "f_1", "filename": "foo.py"},
|
|
]
|
|
out = QwenExecutor._text_from_blocks(blocks)
|
|
assert "review this" in out
|
|
assert "[attached file: foo.py]" in out
|
|
|
|
|
|
def test_text_from_blocks_inlines_text_file_data() -> None:
|
|
"""A text input_file with a base64 data URI is inlined into the prompt."""
|
|
import base64
|
|
|
|
payload = base64.b64encode(b"print('hi')").decode()
|
|
blocks = [
|
|
{"type": "input_text", "text": "summarize"},
|
|
{
|
|
"type": "input_file",
|
|
"file_data": f"data:text/x-python;base64,{payload}",
|
|
"filename": "a.py",
|
|
},
|
|
]
|
|
out = QwenExecutor._text_from_blocks(blocks)
|
|
assert "summarize" in out
|
|
assert "print('hi')" in out
|
|
# Content is fenced with a labeled header/footer so weaker models read it as
|
|
# an attachment, not instructions (regression: bare-appended file content
|
|
# derailed qwen3-coder:free into narrating the tool call as prose).
|
|
assert "--- attached file: a.py ---" in out
|
|
assert "--- end of a.py ---" in out
|
|
|
|
|
|
def test_text_from_blocks_skips_image_and_marks_binary_file() -> None:
|
|
"""Images are skipped (deferred); binary files fall back to a name marker."""
|
|
import base64
|
|
|
|
payload = base64.b64encode(b"%PDF-1.4").decode()
|
|
blocks = [
|
|
{"type": "input_image", "file_id": "img"},
|
|
{
|
|
"type": "input_file",
|
|
"file_data": f"data:application/pdf;base64,{payload}",
|
|
"filename": "d.pdf",
|
|
},
|
|
]
|
|
out = QwenExecutor._text_from_blocks(blocks)
|
|
assert out == "[attached file: d.pdf]"
|
|
|
|
|
|
def test_image_blocks_from_content_builds_acp_image_block() -> None:
|
|
"""An input_image with a resolved data URI becomes an ACP image block."""
|
|
import base64
|
|
|
|
payload = base64.b64encode(b"\x89PNG...").decode()
|
|
content = [
|
|
{"type": "input_text", "text": "what is this"},
|
|
{"type": "input_image", "image_url": f"data:image/png;base64,{payload}"},
|
|
]
|
|
blocks = QwenExecutor._image_blocks_from_content(content)
|
|
assert blocks == [{"type": "image", "mimeType": "image/png", "data": payload}]
|
|
|
|
|
|
def test_image_blocks_from_content_skips_non_image_and_external_urls() -> None:
|
|
"""Only inline image data URIs are forwarded; URLs/non-images are skipped."""
|
|
content = [
|
|
{"type": "input_image", "image_url": "https://example.com/cat.png"},
|
|
{"type": "input_file", "file_data": "data:text/plain;base64,aGk="},
|
|
{"type": "input_image", "image_url": "data:application/pdf;base64,JVBE"},
|
|
]
|
|
assert QwenExecutor._image_blocks_from_content(content) == []
|
|
|
|
|
|
def test_image_blocks_from_content_uses_file_data_fallback() -> None:
|
|
"""An input_image carrying its data URI in file_data (not image_url) works."""
|
|
content = [
|
|
{"type": "input_image", "file_data": "data:image/jpeg;base64,/9j/4AAQ"},
|
|
]
|
|
assert QwenExecutor._image_blocks_from_content(content) == [
|
|
{"type": "image", "mimeType": "image/jpeg", "data": "/9j/4AAQ"}
|
|
]
|
|
|
|
|
|
def test_parse_image_data_uri_edge_cases() -> None:
|
|
"""Malformed / non-image data URIs return None rather than raising."""
|
|
from omnigent.inner.qwen_executor import _parse_image_data_uri
|
|
|
|
assert _parse_image_data_uri("data:image/png;base64") is None # no comma
|
|
assert _parse_image_data_uri("data:image/png;base64,") is None # empty payload
|
|
assert _parse_image_data_uri("https://example.com/x.png") is None # not a data URI
|
|
assert _parse_image_data_uri(None) is None
|
|
assert _parse_image_data_uri("data:image/webp;base64,UklGR") == ("image/webp", "UklGR")
|
|
|
|
|
|
def test_text_from_blocks_marks_image_only_when_requested() -> None:
|
|
"""Image markers appear only with emit_image_marker (capability-off path)."""
|
|
content = [
|
|
{"type": "input_text", "text": "what is this"},
|
|
{"type": "input_image", "image_url": "data:image/png;base64,iVBOR", "filename": "p.png"},
|
|
]
|
|
# Default: image handled as a real block elsewhere → no marker here.
|
|
assert QwenExecutor._text_from_blocks(content) == "what is this"
|
|
# Capability off: leave a marker so the image isn't silently dropped.
|
|
marked = QwenExecutor._text_from_blocks(content, emit_image_marker=True)
|
|
assert "what is this" in marked
|
|
assert "[attached image: p.png]" in marked
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_gateway_env_runs_auth_command() -> None:
|
|
"""A wired gateway → OPENAI_* env with the token from the auth command."""
|
|
executor = QwenExecutor(
|
|
model="qwen/qwen3-coder",
|
|
gateway_base_url="https://gw.example/v1",
|
|
gateway_auth_command="printf '%s' sk-tok-123",
|
|
)
|
|
env = await executor._resolve_gateway_env()
|
|
assert env == {
|
|
"OPENAI_BASE_URL": "https://gw.example/v1",
|
|
"OPENAI_API_KEY": "sk-tok-123",
|
|
"OPENAI_MODEL": "qwen/qwen3-coder",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_gateway_env_empty_without_config() -> None:
|
|
"""No gateway configured → no OPENAI_* overrides (ambient auth path)."""
|
|
assert await QwenExecutor(model="m")._resolve_gateway_env() == {}
|
|
# base URL without an auth command is also inert.
|
|
only_url = QwenExecutor(gateway_base_url="https://gw/v1")
|
|
assert await only_url._resolve_gateway_env() == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_gateway_env_raises_on_command_failure() -> None:
|
|
"""A failing auth command surfaces a clear error rather than an empty key."""
|
|
executor = QwenExecutor(
|
|
gateway_base_url="https://gw/v1",
|
|
gateway_auth_command="exit 3",
|
|
)
|
|
with pytest.raises(RuntimeError, match="auth command failed"):
|
|
await executor._resolve_gateway_env()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_gateway_env_raises_on_empty_token() -> None:
|
|
"""An auth command that prints nothing is treated as a failure."""
|
|
executor = QwenExecutor(
|
|
gateway_base_url="https://gw/v1",
|
|
gateway_auth_command="true", # exits 0, no stdout
|
|
)
|
|
with pytest.raises(RuntimeError, match="empty token"):
|
|
await executor._resolve_gateway_env()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_gateway_env_omits_model_when_unset() -> None:
|
|
"""Without a model, only base URL + key are exported (no OPENAI_MODEL)."""
|
|
executor = QwenExecutor(
|
|
gateway_base_url="https://gw/v1",
|
|
gateway_auth_command="printf '%s' k",
|
|
)
|
|
env = await executor._resolve_gateway_env()
|
|
assert env == {"OPENAI_BASE_URL": "https://gw/v1", "OPENAI_API_KEY": "k"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_initialized_captures_image_capability() -> None:
|
|
"""initialize handshake records promptCapabilities.image on the executor."""
|
|
executor = QwenExecutor(model="m")
|
|
executor._rpc = AsyncMock( # type: ignore[method-assign]
|
|
return_value={"result": {"agentCapabilities": {"promptCapabilities": {"image": True}}}}
|
|
)
|
|
await executor._ensure_initialized()
|
|
assert executor._initialized is True
|
|
assert executor._image_supported is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_initialized_image_capability_defaults_false() -> None:
|
|
"""Absent promptCapabilities leaves image support off (degrade to marker)."""
|
|
executor = QwenExecutor(model="m")
|
|
executor._rpc = AsyncMock(return_value={"result": {}}) # type: ignore[method-assign]
|
|
await executor._ensure_initialized()
|
|
assert executor._initialized is True
|
|
assert executor._image_supported is False
|