diff options
Diffstat (limited to 'packages/sdk/python/tests')
| -rw-r--r-- | packages/sdk/python/tests/__init__.py | 1 | ||||
| -rw-r--r-- | packages/sdk/python/tests/test_integration.py | 93 | ||||
| -rw-r--r-- | packages/sdk/python/tests/test_wrapper.py | 116 |
3 files changed, 0 insertions, 210 deletions
diff --git a/packages/sdk/python/tests/__init__.py b/packages/sdk/python/tests/__init__.py deleted file mode 100644 index 1a9709f46..000000000 --- a/packages/sdk/python/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# Makes tests/ a package for import-resolution friendliness diff --git a/packages/sdk/python/tests/test_integration.py b/packages/sdk/python/tests/test_integration.py deleted file mode 100644 index cbb6c3999..000000000 --- a/packages/sdk/python/tests/test_integration.py +++ /dev/null @@ -1,93 +0,0 @@ -import os -import re -import signal -import subprocess -import sys -import time -from pathlib import Path - -import httpx -import pytest - -from opencode_ai import OpenCodeClient - - -def test_integration_live_server_endpoints() -> None: - # Locate repo root by finding sst.config.ts upwards from this file - here = Path(__file__).resolve() - p = here - repo_root = None - for _ in range(8): - if (p / "sst.config.ts").exists(): - repo_root = p - break - if p.parent == p: - break - p = p.parent - assert repo_root is not None, "Could not locate repo root (sst.config.ts)" - - # Start opencode headless server on a random port - pkg_opencode = repo_root / "packages" / "opencode" - cmd = [ - "bun", - "run", - "--conditions=development", - "./src/index.ts", - "serve", - "--port", - "0", - "--hostname", - "127.0.0.1", - ] - - proc = subprocess.Popen( - cmd, - cwd=str(pkg_opencode), - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - bufsize=1, - universal_newlines=True, - ) - - url = None - start = time.time() - assert proc.stdout is not None - while time.time() - start < 15: - line = proc.stdout.readline() - if not line: - time.sleep(0.05) - if proc.poll() is not None: - break - continue - m = re.search(r"opencode server listening on (http://[^\s]+)", line) - if m: - url = m.group(1) - break - assert url, "Server did not report listening URL" - - try: - client = OpenCodeClient(base_url=url) - # Basic endpoints (avoid complex config model parsing issues) - pinfo = client.get_path() - assert pinfo is not None - projects = client.list_projects() - assert projects is not None - - # SSE: should get the initial server.connected event - it = client.subscribe_events() - evt = next(it) - assert isinstance(evt, dict) - assert evt.get("type") == "server.connected" - finally: - # Cleanup server process - try: - if proc.poll() is None: - proc.terminate() - try: - proc.wait(timeout=5) - except subprocess.TimeoutExpired: - proc.kill() - except Exception: - pass diff --git a/packages/sdk/python/tests/test_wrapper.py b/packages/sdk/python/tests/test_wrapper.py deleted file mode 100644 index f2003a536..000000000 --- a/packages/sdk/python/tests/test_wrapper.py +++ /dev/null @@ -1,116 +0,0 @@ -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" |