wehub-resource-sync
1340 lines
43 KiB
Python
1340 lines
43 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from opensquilla.cli.agent_cmd import (
|
|
AgentRunResult,
|
|
_to_benchmark_transcript,
|
|
run_agent_command,
|
|
run_agent_once,
|
|
)
|
|
from opensquilla.engine.types import ArtifactEvent, DoneEvent
|
|
from opensquilla.gateway.config import AgentEntryConfig, GatewayConfig, PermissionsConfig
|
|
from opensquilla.sandbox.config import SandboxSettings
|
|
from opensquilla.tools.types import CallerKind, InteractionMode
|
|
|
|
|
|
class _FakeSessionManager:
|
|
def __init__(self) -> None:
|
|
self.messages: list[tuple[str, str, str]] = []
|
|
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main") -> object:
|
|
return SimpleNamespace(session_key=session_key, agent_id=agent_id)
|
|
|
|
async def append_message(self, session_key: str, role: str, content: str) -> None:
|
|
self.messages.append((session_key, role, content))
|
|
return None
|
|
|
|
|
|
class _FakeServices:
|
|
def __init__(self, config: GatewayConfig) -> None:
|
|
self.memory_sync_managers = {"main": object()}
|
|
self.memory_retrievers = {"main": object()}
|
|
self.turn_capture_services = {"main": object()}
|
|
self.flush_service = object()
|
|
self.model_catalog = object()
|
|
self.provider_selector = object()
|
|
self.tool_registry = None
|
|
self.session_manager = _FakeSessionManager()
|
|
self.skill_loader = None
|
|
self.usage_tracker = None
|
|
self.config = config
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
|
|
def test_benchmark_transcript_preserves_tool_result_execution_status() -> None:
|
|
status = {
|
|
"version": 1,
|
|
"status": "timeout",
|
|
"exit_code": None,
|
|
"timed_out": True,
|
|
"truncated": False,
|
|
"reason": "runtime_timeout",
|
|
"source": "tool_runtime",
|
|
"preservation_class": "diagnostic",
|
|
}
|
|
entry = SimpleNamespace(
|
|
role="assistant",
|
|
content="",
|
|
created_at=None,
|
|
tool_calls=[
|
|
{
|
|
"type": "tool_use",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"input": {"cmd": "sleep"},
|
|
},
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"result": "timeout",
|
|
"is_error": True,
|
|
"execution_status": status,
|
|
},
|
|
],
|
|
)
|
|
|
|
transcript = _to_benchmark_transcript([entry])
|
|
|
|
assert transcript[1]["message"]["isError"] is True
|
|
assert transcript[1]["message"]["executionStatus"] == status
|
|
|
|
|
|
def test_benchmark_transcript_preserves_result_truncated() -> None:
|
|
entry = SimpleNamespace(
|
|
role="assistant",
|
|
content="",
|
|
created_at=None,
|
|
tool_calls=[
|
|
{
|
|
"type": "tool_use",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"input": {"cmd": "cat big.log"},
|
|
},
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"result": "head of output ...",
|
|
"is_error": False,
|
|
"result_truncated": True,
|
|
"result_original_chars": 48210,
|
|
},
|
|
],
|
|
)
|
|
|
|
transcript = _to_benchmark_transcript([entry])
|
|
|
|
assert transcript[1]["message"]["resultTruncated"] is True
|
|
assert transcript[1]["message"]["resultOriginalChars"] == 48210
|
|
|
|
|
|
def test_benchmark_transcript_omits_truncation_fields_when_absent() -> None:
|
|
entry = SimpleNamespace(
|
|
role="assistant",
|
|
content="",
|
|
created_at=None,
|
|
tool_calls=[
|
|
{
|
|
"type": "tool_use",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"input": {"cmd": "true"},
|
|
},
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "call-1",
|
|
"name": "exec_command",
|
|
"result": "",
|
|
"is_error": False,
|
|
},
|
|
],
|
|
)
|
|
|
|
transcript = _to_benchmark_transcript([entry])
|
|
|
|
assert "resultTruncated" not in transcript[1]["message"]
|
|
assert "resultOriginalChars" not in transcript[1]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_agent_registry_model_when_model_not_explicit(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
captured["runner_config_model"] = kwargs["config"].llm.model
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["run_model"] = kwargs.get("model")
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
captured["service_config_model"] = config.llm.model
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
root = tmp_path / "root"
|
|
agent_workspace = tmp_path / "ops-workspace"
|
|
cfg = GatewayConfig(
|
|
workspace_dir=str(root),
|
|
agents=[
|
|
AgentEntryConfig(
|
|
id="ops",
|
|
model="agent/default",
|
|
workspace=str(agent_workspace),
|
|
)
|
|
],
|
|
)
|
|
result = await run_agent_once(message="hello", agent_id="ops", config=cfg)
|
|
|
|
assert result.text == "ok"
|
|
assert captured["service_config_model"] == "agent/default"
|
|
assert captured["runner_config_model"] == "agent/default"
|
|
assert captured["run_model"] == "agent/default"
|
|
assert captured["tool_context"].workspace_dir == str(agent_workspace)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_configured_full_host_run_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
cfg = GatewayConfig(
|
|
workspace_dir=str(tmp_path),
|
|
sandbox=SandboxSettings(run_mode="full"),
|
|
)
|
|
|
|
result = await run_agent_once(message="hello", config=cfg)
|
|
|
|
assert result.text == "ok"
|
|
assert captured["tool_context"].run_mode == "full"
|
|
assert captured["tool_context"].elevated == "full"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_collects_artifact_events(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
artifact = {
|
|
"id": "art-cli",
|
|
"kind": "artifact_ref",
|
|
"name": "report.txt",
|
|
"mime": "text/plain",
|
|
"size": 4,
|
|
"sha256": "d" * 64,
|
|
"session_id": "session-1",
|
|
"session_key": "agent:main:main",
|
|
"source": "publish_artifact",
|
|
"created_at": "2026-05-06T12:00:00Z",
|
|
"download_url": "/api/v1/artifacts/art-cli?sessionKey=agent%3Amain%3Amain",
|
|
"store": "artifacts",
|
|
}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
yield ArtifactEvent(**artifact)
|
|
yield DoneEvent(text="ok")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
result = await run_agent_once(message="hello", config=GatewayConfig())
|
|
|
|
assert result.artifacts == [
|
|
{
|
|
**{key: value for key, value in artifact.items() if key != "session_key"},
|
|
"download_url": "/api/v1/artifacts/art-cli",
|
|
}
|
|
]
|
|
assert result.artifacts[0]["download_url"] == "/api/v1/artifacts/art-cli"
|
|
|
|
|
|
def test_run_agent_command_json_includes_artifacts(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
artifact = {
|
|
"id": "art-cli",
|
|
"kind": "artifact_ref",
|
|
"name": "report.txt",
|
|
"mime": "text/plain",
|
|
"size": 4,
|
|
"sha256": "d" * 64,
|
|
"session_id": "session-1",
|
|
"session_key": "agent:main:main",
|
|
"source": "publish_artifact",
|
|
"created_at": "2026-05-06T12:00:00Z",
|
|
"download_url": "/api/v1/artifacts/art-cli?sessionKey=agent%3Amain%3Amain",
|
|
"store": "artifacts",
|
|
}
|
|
|
|
async def fake_run_agent_once(**kwargs: Any) -> AgentRunResult:
|
|
return AgentRunResult(
|
|
status="ok",
|
|
agent_id="main",
|
|
session_key="agent:main:main",
|
|
text="ok",
|
|
usage={},
|
|
errors=[],
|
|
artifacts=[artifact],
|
|
)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.agent_cmd.run_agent_once", fake_run_agent_once)
|
|
|
|
run_agent_command(message="hello", json_output=True)
|
|
|
|
output_artifact = json.loads(capsys.readouterr().out)["artifacts"][0]
|
|
assert "session_key" not in output_artifact
|
|
assert "sessionKey" not in json.dumps(output_artifact)
|
|
assert output_artifact["download_url"] == "/api/v1/artifacts/art-cli"
|
|
|
|
|
|
def test_run_agent_command_direct_call_normalizes_typer_defaults(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_run_agent_once(**kwargs: Any) -> AgentRunResult:
|
|
captured.update(kwargs)
|
|
return AgentRunResult(
|
|
status="ok",
|
|
agent_id="main",
|
|
session_key="agent:main:main",
|
|
text="ok",
|
|
usage={},
|
|
errors=[],
|
|
)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.agent_cmd.run_agent_once", fake_run_agent_once)
|
|
|
|
run_agent_command(message="hello")
|
|
|
|
assert captured["agent_id"] == "main"
|
|
assert captured["session_id"] == ""
|
|
assert captured["model"] is None
|
|
assert captured["workspace"] is None
|
|
assert captured["workspace_strict"] is None
|
|
assert captured["workspace_lockdown"] is False
|
|
assert captured["workspace_write_deny_globs"] == []
|
|
assert captured["scratch_dir"] is None
|
|
assert captured["timeout"] is None
|
|
assert captured["max_iterations"] is None
|
|
assert captured["iteration_timeout"] is None
|
|
assert captured["tool_timeout"] is None
|
|
assert captured["request_timeout"] is None
|
|
assert captured["max_provider_retries"] is None
|
|
assert captured["length_capped_continuations"] is None
|
|
assert captured["thinking"] is None
|
|
assert captured["transcript_path"] is None
|
|
assert captured["usage_path"] is None
|
|
assert captured["session_db_path"] == ":memory:"
|
|
assert captured["no_memory_capture"] is False
|
|
assert captured["attachment_paths"] == []
|
|
assert captured["unattended"] is True
|
|
assert captured["stateless"] is False
|
|
assert captured["stateless_keep_project_rules"] is False
|
|
assert captured["permissions"] is None
|
|
|
|
|
|
def test_run_agent_command_json_includes_routing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
async def fake_run_agent_once(**kwargs: Any) -> AgentRunResult:
|
|
result = AgentRunResult(
|
|
status="ok",
|
|
agent_id="main",
|
|
session_key="agent:main:main",
|
|
text="ok",
|
|
usage={},
|
|
errors=[],
|
|
)
|
|
result.routing = { # type: ignore[attr-defined]
|
|
"routed_tier": "c2",
|
|
"routing_source": "v4_phase3",
|
|
"routing_confidence": 0.91,
|
|
"baseline_model": "openrouter/heavy",
|
|
"routed_model": "openrouter/light",
|
|
}
|
|
return result
|
|
|
|
monkeypatch.setattr("opensquilla.cli.agent_cmd.run_agent_once", fake_run_agent_once)
|
|
|
|
run_agent_command(message="hello", json_output=True)
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["routing"] == {
|
|
"routed_tier": "c2",
|
|
"routing_source": "v4_phase3",
|
|
"routing_confidence": 0.91,
|
|
"baseline_model": "openrouter/heavy",
|
|
"routed_model": "openrouter/light",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_collects_done_routing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
yield DoneEvent(
|
|
text="ok",
|
|
routed_tier="c2",
|
|
routing_source="v4_phase3",
|
|
routing_confidence=0.91,
|
|
baseline_model="openrouter/heavy",
|
|
routed_model="openrouter/light",
|
|
)
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
result = await run_agent_once(message="hello", config=GatewayConfig())
|
|
|
|
assert result.routing == { # type: ignore[attr-defined]
|
|
"routed_tier": "c2",
|
|
"routing_source": "v4_phase3",
|
|
"routing_confidence": 0.91,
|
|
"baseline_model": "openrouter/heavy",
|
|
"routed_model": "openrouter/light",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_explicit_model_overrides_agent_registry_model(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
captured["runner_config_model"] = kwargs["config"].llm.model
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["run_model"] = kwargs.get("model")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
captured["service_config_model"] = config.llm.model
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
cfg = GatewayConfig(agents=[AgentEntryConfig(id="ops", model="agent/default")])
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="ops",
|
|
model="explicit/model",
|
|
config=cfg,
|
|
)
|
|
|
|
assert captured["service_config_model"] == "explicit/model"
|
|
assert captured["runner_config_model"] == "explicit/model"
|
|
assert captured["run_model"] == "explicit/model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_configured_agent_workspace_without_global_workspace(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
agent_workspace = tmp_path / "ops-workspace"
|
|
cfg = GatewayConfig(
|
|
agents=[
|
|
AgentEntryConfig(
|
|
id="ops",
|
|
workspace=str(agent_workspace),
|
|
)
|
|
],
|
|
)
|
|
|
|
result = await run_agent_once(message="hello", agent_id="ops", config=cfg)
|
|
|
|
assert captured["tool_context"].workspace_dir == str(agent_workspace)
|
|
assert result.workspace == str(agent_workspace)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_wires_memory_services_into_turnrunner(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
services: _FakeServices | None = None
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
nonlocal services
|
|
services = _FakeServices(config)
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(message="hello", agent_id="main", config=GatewayConfig())
|
|
|
|
assert services is not None
|
|
assert captured["memory_sync_managers"] is services.memory_sync_managers
|
|
assert captured["memory_retrievers"] is services.memory_retrievers
|
|
assert captured["turn_capture_services"] is services.turn_capture_services
|
|
assert captured["session_flush_service"] is services.flush_service
|
|
assert captured["model_catalog"] is services.model_catalog
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_forwards_max_iterations(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["max_iterations"] = kwargs.get("max_iterations")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
max_iterations=321,
|
|
)
|
|
|
|
assert captured["max_iterations"] == 321
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_forwards_zero_max_iterations(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["max_iterations"] = kwargs.get("max_iterations")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
max_iterations=0,
|
|
)
|
|
|
|
assert captured["max_iterations"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_defaults_to_unattended_interaction_contract(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
captured["bootstrap_context_mode"] = kwargs.get("bootstrap_context_mode")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(message="hello", agent_id="main", config=GatewayConfig())
|
|
|
|
ctx = captured["tool_context"]
|
|
assert ctx.caller_kind is CallerKind.CLI
|
|
assert ctx.interaction_mode is InteractionMode.UNATTENDED
|
|
assert ctx.run_mode == "full"
|
|
assert ctx.elevated == "full"
|
|
assert captured["bootstrap_context_mode"] == "unattended"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_passes_bypass_permissions_to_tool_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
permissions="bypass",
|
|
)
|
|
|
|
ctx = captured["tool_context"]
|
|
assert ctx.interaction_mode is InteractionMode.UNATTENDED
|
|
assert ctx.elevated == "bypass"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_permissions_environment_default(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setenv("OPENSQUILLA_AGENT_PERMISSIONS", "full")
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(message="hello", agent_id="main", config=GatewayConfig())
|
|
|
|
assert captured["tool_context"].elevated == "full"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_default_full_host_run_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.delenv("OPENSQUILLA_AGENT_PERMISSIONS", raising=False)
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
)
|
|
|
|
assert captured["tool_context"].run_mode == "full"
|
|
assert captured["tool_context"].elevated == "full"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_uses_configured_permissions_default(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.delenv("OPENSQUILLA_AGENT_PERMISSIONS", raising=False)
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(permissions=PermissionsConfig(default_mode="bypass")),
|
|
)
|
|
|
|
assert captured["tool_context"].elevated == "bypass"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_rejects_invalid_permissions() -> None:
|
|
with pytest.raises(ValueError, match="permissions must be one of"):
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
permissions="benchmark",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_can_opt_into_interactive_single_shot(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
captured["bootstrap_context_mode"] = kwargs.get("bootstrap_context_mode")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
unattended=False,
|
|
)
|
|
|
|
assert captured["tool_context"].interaction_mode is InteractionMode.INTERACTIVE
|
|
assert captured["bootstrap_context_mode"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_can_opt_into_stateless_bootstrap(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["bootstrap_context_mode"] = kwargs.get("bootstrap_context_mode")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(message="hello", config=GatewayConfig(), stateless=True)
|
|
|
|
assert captured["bootstrap_context_mode"] == "stateless"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_can_keep_project_rules_in_stateless_bootstrap(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["bootstrap_context_mode"] = kwargs.get("bootstrap_context_mode")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(),
|
|
stateless=True,
|
|
stateless_keep_project_rules=True,
|
|
)
|
|
|
|
assert captured["bootstrap_context_mode"] == "stateless_keep_project_rules"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("stateless", "stateless_keep_project_rules"),
|
|
[
|
|
(True, False),
|
|
(False, True),
|
|
],
|
|
)
|
|
async def test_run_agent_once_disables_workspace_template_seeding_for_stateless(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
stateless: bool,
|
|
stateless_keep_project_rules: bool,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
captured.update(kwargs)
|
|
captured["config"] = config
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(),
|
|
workspace=str(tmp_path),
|
|
stateless=stateless,
|
|
stateless_keep_project_rules=stateless_keep_project_rules,
|
|
no_memory_capture=True,
|
|
)
|
|
|
|
assert captured["seed_agent_workspaces"] is False
|
|
assert captured["config"].memory.source == "state"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_keeps_workspace_template_seeding_for_normal_runs(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
captured.update(kwargs)
|
|
captured["config"] = config
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(),
|
|
workspace=str(tmp_path),
|
|
)
|
|
|
|
assert captured["seed_agent_workspaces"] is True
|
|
assert captured["config"].memory.source == "workspace"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_passes_scratch_and_lockdown_to_tool_context(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
scratch_dir = tmp_path / "scratch"
|
|
await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(),
|
|
scratch_dir=str(scratch_dir),
|
|
workspace_lockdown=True,
|
|
workspace_write_deny_globs=["reports/*.txt, tmp/**", "generated/**"],
|
|
)
|
|
|
|
ctx = captured["tool_context"]
|
|
assert ctx.scratch_dir == str(scratch_dir)
|
|
assert ctx.workspace_lockdown is True
|
|
assert ctx.workspace_write_deny_globs == ["reports/*.txt", "tmp/**", "generated/**"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_merges_config_workspace_write_deny_globs(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeRegistry:
|
|
def list_names(self) -> list[str]:
|
|
return ["write_file", "exec_command", "apply_patch"]
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["tool_context"] = kwargs.get("tool_context")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
services = _FakeServices(config)
|
|
services.tool_registry = FakeRegistry()
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
scratch_dir = tmp_path / "scratch"
|
|
result = await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(
|
|
tools={
|
|
"workspace_write_deny_globs": [
|
|
"**/test/**",
|
|
"**/*Test.java",
|
|
]
|
|
}
|
|
),
|
|
scratch_dir=str(scratch_dir),
|
|
workspace_lockdown=True,
|
|
workspace_write_deny_globs=["reports/*.txt", "**/test/**"],
|
|
)
|
|
|
|
ctx = captured["tool_context"]
|
|
assert ctx.workspace_write_deny_globs[0] == "reports/*.txt"
|
|
assert set(ctx.workspace_write_deny_globs) == {
|
|
"reports/*.txt",
|
|
"**/test/**",
|
|
"**/*Test.java",
|
|
}
|
|
assert len(ctx.workspace_write_deny_globs) == 3
|
|
assert set(result.workspace_write_deny_globs or []) == {
|
|
"reports/*.txt",
|
|
"**/test/**",
|
|
"**/*Test.java",
|
|
}
|
|
|
|
|
|
def test_run_agent_command_forwards_workspace_lockdown_deny_paths(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_run_agent_once(**kwargs: Any) -> AgentRunResult:
|
|
captured.update(kwargs)
|
|
return AgentRunResult(
|
|
status="ok",
|
|
agent_id="main",
|
|
session_key="agent:main:main",
|
|
text="ok",
|
|
usage={},
|
|
errors=[],
|
|
)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.agent_cmd.run_agent_once", fake_run_agent_once)
|
|
|
|
run_agent_command(
|
|
message="hello",
|
|
workspace_lockdown_deny_paths=["reports/*.txt, tmp/**", "generated/**"],
|
|
)
|
|
|
|
assert captured["workspace_write_deny_globs"] == [
|
|
"reports/*.txt",
|
|
"tmp/**",
|
|
"generated/**",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_rejects_workspace_lockdown_without_allowed_roots() -> None:
|
|
with pytest.raises(ValueError, match="workspace_lockdown requires"):
|
|
await run_agent_once(
|
|
message="hello",
|
|
config=GatewayConfig(workspace_dir=None),
|
|
workspace_lockdown=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_rejects_invalid_max_iterations() -> None:
|
|
with pytest.raises(ValueError, match="max_iterations"):
|
|
await run_agent_once(
|
|
message="hello",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
max_iterations=-1,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_forwards_inline_attachments(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["attachments"] = kwargs.get("attachments")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
payload = base64.b64encode(b"hello").decode("ascii")
|
|
await run_agent_once(
|
|
message="summarize",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
attachments=[{"type": "text/plain", "data": payload, "name": "note.txt"}],
|
|
)
|
|
|
|
assert captured["attachments"] == [
|
|
{"type": "text/plain", "data": payload, "name": "note.txt"}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_builds_multiple_file_attachments(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs: Any):
|
|
captured["attachments"] = kwargs.get("attachments")
|
|
yield DoneEvent(text="ok", model=kwargs.get("model") or "")
|
|
|
|
async def fake_build_services(*, config: GatewayConfig, **kwargs: Any) -> _FakeServices:
|
|
return _FakeServices(config)
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
|
|
note = tmp_path / "note.txt"
|
|
note.write_bytes(b"hello")
|
|
data = tmp_path / "data.csv"
|
|
data.write_bytes(b"a,b\n1,2\n")
|
|
|
|
await run_agent_once(
|
|
message="compare",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
attachment_paths=[str(note), str(data)],
|
|
)
|
|
|
|
attachments = captured["attachments"]
|
|
assert len(attachments) == 2
|
|
assert [item["type"] for item in attachments] == ["text/plain", "text/csv"]
|
|
assert base64.b64decode(attachments[0]["data"]) == b"hello"
|
|
assert base64.b64decode(attachments[1]["data"]) == b"a,b\n1,2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_rejects_agent_file_requiring_upload_bridge(tmp_path) -> None:
|
|
big_pdf = tmp_path / "big.pdf"
|
|
big_pdf.write_bytes(b"%PDF-1.4\n" + b"a" * 2_000_001)
|
|
|
|
with pytest.raises(ValueError, match="gateway bridge upload"):
|
|
await run_agent_once(
|
|
message="read",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
attachment_paths=[str(big_pdf)],
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_once_requires_staging_for_large_text_file(tmp_path) -> None:
|
|
# Text above the inline threshold is stageable now; the one-shot path has
|
|
# no gateway upload bridge, so it must fail with the staging hint rather
|
|
# than a text-family dead end.
|
|
big_csv = tmp_path / "big.csv"
|
|
big_csv.write_bytes(b"a" * 2_000_001)
|
|
|
|
with pytest.raises(ValueError, match=r"upload is required|too large to inline"):
|
|
await run_agent_once(
|
|
message="read",
|
|
agent_id="main",
|
|
config=GatewayConfig(),
|
|
attachment_paths=[str(big_csv)],
|
|
)
|
|
|
|
|
|
def test_top_level_agent_command_accepts_repeated_file_options(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_run_agent_command(**kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.main.run_agent_command", fake_run_agent_command)
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"compare",
|
|
"--file",
|
|
"a.txt",
|
|
"--file",
|
|
"b.csv",
|
|
"--json",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["message"] == "compare"
|
|
assert captured["file_paths"] == ["a.txt", "b.csv"]
|
|
assert captured["json_output"] is True
|
|
assert captured["unattended"] is True
|
|
assert captured["permissions"] is None
|
|
|
|
|
|
def test_top_level_agent_command_accepts_interactive_option(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_run_agent_command(**kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.main.run_agent_command", fake_run_agent_command)
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"hello",
|
|
"--interactive",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["unattended"] is False
|
|
|
|
|
|
def test_top_level_agent_command_accepts_permissions_option(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_run_agent_command(**kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.main.run_agent_command", fake_run_agent_command)
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"hello",
|
|
"--permissions",
|
|
"bypass",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["permissions"] == "bypass"
|
|
|
|
|
|
def test_top_level_agent_command_accepts_automation_options(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_run_agent_command(**kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.main.run_agent_command", fake_run_agent_command)
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"hello",
|
|
"--clean-room",
|
|
"--stateless-keep-project-rules",
|
|
"--scratch-dir",
|
|
"scratch",
|
|
"--workspace-lockdown",
|
|
"--workspace-lockdown-deny-paths",
|
|
"reports/*.txt, tmp/**",
|
|
"--workspace-lockdown-deny-paths",
|
|
"generated/**",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["clean_room"] is True
|
|
assert captured["stateless_keep_project_rules"] is True
|
|
assert captured["scratch_dir"] == "scratch"
|
|
assert captured["workspace_lockdown"] is True
|
|
assert captured["workspace_lockdown_deny_paths"] == [
|
|
"reports/*.txt, tmp/**",
|
|
"generated/**",
|
|
]
|
|
|
|
|
|
def test_top_level_agent_command_accepts_length_capped_continuations(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_run_agent_command(**kwargs: Any) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr("opensquilla.cli.main.run_agent_command", fake_run_agent_command)
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"hello",
|
|
"--length-capped-continuations",
|
|
"3",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
assert captured["length_capped_continuations"] == 3
|
|
|
|
|
|
def test_top_level_agent_command_rejects_invalid_length_capped_continuations() -> None:
|
|
from opensquilla.cli.main import app
|
|
|
|
result = CliRunner().invoke(
|
|
app,
|
|
[
|
|
"agent",
|
|
"--message",
|
|
"hello",
|
|
"--length-capped-continuations",
|
|
"0",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
output = re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", result.output)
|
|
assert "Invalid value" in output
|
|
assert "--length-capped-continuations" in output
|
|
assert "x>=1" in output
|