82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from agentci.integrations.opencode.client import OpenCode, OpenCodeError
|
|
|
|
|
|
def client(tmp_path: Path, status: int) -> OpenCode:
|
|
return OpenCode(
|
|
base_url="http://opencode:4096",
|
|
username="opencode",
|
|
password="secret",
|
|
schemas_dir=tmp_path,
|
|
health_directory=tmp_path,
|
|
required_models=(),
|
|
timeout_seconds=60,
|
|
transport=httpx.MockTransport(lambda _request: httpx.Response(status)),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("status", [200, 204, 404, 409])
|
|
async def test_absent_or_inactive_session_is_success(tmp_path: Path, status: int) -> None:
|
|
value = client(tmp_path, status)
|
|
await value.abort("session", tmp_path)
|
|
await value.close()
|
|
|
|
|
|
@pytest.mark.parametrize("status", [400, 429, 500])
|
|
async def test_abort_failure_is_visible_for_retry(tmp_path: Path, status: int) -> None:
|
|
value = client(tmp_path, status)
|
|
with pytest.raises(OpenCodeError):
|
|
await value.abort("session", tmp_path)
|
|
await value.close()
|
|
|
|
|
|
async def test_abort_sends_authenticated_workspace_request(tmp_path: Path) -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
return httpx.Response(204)
|
|
|
|
value = OpenCode(
|
|
base_url="http://opencode:4096/",
|
|
username="opencode",
|
|
password="secret",
|
|
schemas_dir=tmp_path,
|
|
health_directory=tmp_path,
|
|
required_models=(),
|
|
timeout_seconds=60,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await value.abort("ses_123", tmp_path)
|
|
await value.close()
|
|
|
|
assert len(requests) == 1
|
|
request = requests[0]
|
|
assert request.method == "POST"
|
|
assert request.url == httpx.URL("http://opencode:4096/session/ses_123/abort")
|
|
assert request.headers["authorization"].startswith("Basic ")
|
|
assert request.headers["x-opencode-directory"] == str(tmp_path)
|
|
|
|
|
|
async def test_best_effort_abort_suppresses_transport_failure(tmp_path: Path) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("connection refused", request=request)
|
|
|
|
value = OpenCode(
|
|
base_url="http://opencode:4096",
|
|
username="opencode",
|
|
password="secret",
|
|
schemas_dir=tmp_path,
|
|
health_directory=tmp_path,
|
|
required_models=(),
|
|
timeout_seconds=60,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
await value.abort("session", tmp_path, best_effort=True)
|
|
await value.close()
|