1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
import json
from typing import Iterator
import httpx
import pytest
from opencode_ai import OpenCodeClient
from opencode_ai.api.default import config_get
from opencode_ai.client import Client
class _State:
def __init__(self):
self.calls = 0
def test_imports_and_methods_available() -> None:
w = OpenCodeClient()
assert hasattr(w, "list_sessions")
assert hasattr(w, "get_config")
assert hasattr(w, "list_agents")
assert hasattr(w, "list_projects")
assert hasattr(w, "current_project")
assert hasattr(w, "file_status")
assert hasattr(w, "get_path")
assert hasattr(w, "subscribe_events")
def test_get_path_with_mock_transport() -> None:
# Arrange a mock transport for GET /path
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/path"
return httpx.Response(
200,
json={
"state": "ok",
"config": "/tmp/config",
"worktree": "/repo",
"directory": "/repo/project",
},
)
transport = httpx.MockTransport(handler)
w = OpenCodeClient(base_url="http://test")
client = httpx.Client(base_url="http://test", transport=transport)
w.client.set_httpx_client(client)
# Act
result = w.get_path()
# Assert
assert result is not None
assert result.directory == "/repo/project"
def test_retry_on_request_error_then_success() -> None:
state = _State()
def handler(request: httpx.Request) -> httpx.Response:
if state.calls == 0:
state.calls += 1
raise httpx.ConnectError("boom", request=request)
return httpx.Response(
200,
json={
"state": "ok",
"config": "/tmp/config",
"worktree": "/repo",
"directory": "/repo/project",
},
)
transport = httpx.MockTransport(handler)
w = OpenCodeClient(base_url="http://test", retries=1, backoff_factor=0)
client = httpx.Client(base_url="http://test", transport=transport)
w.client.set_httpx_client(client)
result = w.get_path()
assert result is not None
assert result.directory == "/repo/project"
def test_generated_config_get_via_mock() -> None:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/config"
return httpx.Response(200, json={})
transport = httpx.MockTransport(handler)
c = Client(base_url="http://test")
c.set_httpx_client(httpx.Client(base_url="http://test", transport=transport))
assert config_get.sync(client=c) is not None
def test_sse_streaming_parses_events() -> None:
# Prepare a simple SSE payload with one event
payload = b'data: {"type":"server.connected"}\n\n'
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/event"
return httpx.Response(
200,
headers={"Content-Type": "text/event-stream"},
content=payload,
)
transport = httpx.MockTransport(handler)
w = OpenCodeClient(base_url="http://test")
client = httpx.Client(base_url="http://test", transport=transport)
w.client.set_httpx_client(client)
it = w.subscribe_events()
first = next(it)
assert isinstance(first, dict)
assert first.get("type") == "server.connected"
|
