test(core): port general behavior tests from TypeScript SDK

.github/workflows/test.yml

@@ -0,0 +1,28 @@
+name: Tests
+
+on:
+  push:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          version: "0.9.5"
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.9"
+
+      - name: Install dependencies
+        run: uv sync --group dev
+
+      - name: Run tests
+        run: uv run pytest tests/ -v
+

.speakeasy/gen.yaml

@@ -30,7 +30,10 @@ generation:
 python:
   version: 0.2.0
   additionalDependencies:
-    dev: {}
+    dev:
+      pytest: ">=8.0.0"
+      respx: ">=0.21.0"
+      inline-snapshot: ">=0.13.0"
     main: {}
   allowedRedefinedBuiltins:
     - id

AGENTS.md

@@ -0,0 +1,15 @@
+# Testing Setup
+
+Tests use pytest with `respx` for HTTP mocking and `inline-snapshot` for assertions.
+
+**MockContext helper** (`tests/conftest.py`):
+- `ctx = MockContext(api_key="...", integration_id="...")` - creates SDK instance
+- `ctx.mock_endpoint(method, path, response)` - mocks HTTP endpoints
+- `ctx.get_last_request()` - captures request details (path, headers, body)
+- `ctx.clear()` - resets mocks between calls
+
+**Running tests**:
+- `mise run test` or `uv run pytest tests/ -v`
+- `mise run test:update` - update snapshots
+- Use `snapshot(...)` from `inline-snapshot` for assertions
+

mise.toml

@@ -1,6 +1,16 @@
 [tools]
+python = "3.9"
 "aqua:speakeasy-api/speakeasy" = "1"
+uv = "0.9.5"
 
 [tasks."dev"]
 description = "Build the SDK without publishing"
 run = 'OPENAPI_DOC_AUTH_TOKEN="$(doppler secrets get SDK_SPEC_ENDPOINT_TOKEN --project kombo-engine --config dev --plain)" speakeasy run --minimal --skip-upload-spec --skip-versioning --skip-compile --target kombo-python'
+
+[tasks."test"]
+description = "Run tests"
+run = "uv run pytest tests/ -v"
+
+[tasks."test:update"]
+description = "Run tests and update inline snapshots"
+run = "uv run pytest tests/ --inline-snapshot=fix -v"

pyproject.toml

@@ -18,6 +18,9 @@ dev = [
     "mypy ==1.15.0",
     "pylint ==3.2.3",
     "pyright ==1.1.398",
+    "pytest >=8.0.0",
+    "respx >=0.21.0",
+    "inline-snapshot >=0.13.0",
 ]
 
 [tool.setuptools.packages.find]

tests/__init__.py

@@ -0,0 +1 @@
+"""Test suite for Kombo Python SDK."""

tests/conftest.py

@@ -0,0 +1,176 @@
+"""Test fixtures and helpers for Kombo SDK tests."""
+
+from typing import Optional, Dict, Any, List
+import json
+import re
+import pytest
+import respx
+import httpx
+from kombo import Kombo
+
+
+# Sentinel value to distinguish between "not provided" and "explicitly None"
+_UNSET = object()
+
+
+class CapturedRequest:
+    """Represents a captured HTTP request."""
+
+    def __init__(self, method: str, path: str, headers: Dict[str, str], body: Optional[Any] = None):
+        self.method = method
+        self.path = path
+        self.headers = headers
+        self.body = body
+
+
+class MockContext:
+    """Test context for mocking HTTP requests and capturing request details."""
+
+    def __init__(self, api_key: Optional[str] = None, integration_id: Any = _UNSET):
+        """
+        Initialize test context.
+
+        :param api_key: API key to use (defaults to "test-api-key")
+        :param integration_id: Integration ID to use. Defaults to "test-integration-id" if not provided.
+                               Pass None explicitly to omit the integration_id from SDK initialization.
+        """
+        self._api_key = api_key or "test-api-key"
+
+        # Determine integration_id value
+        if integration_id is _UNSET:
+            # Not provided, use default
+            id_to_use = "test-integration-id"
+        else:
+            # Use provided value (could be None, a string, etc.)
+            id_to_use = integration_id
+
+        self._integration_id = id_to_use
+        self._captured_requests: List[CapturedRequest] = []
+
+        # Initialize SDK
+        if id_to_use is None:
+            self.kombo = Kombo(api_key=self._api_key)
+        else:
+            self.kombo = Kombo(api_key=self._api_key, integration_id=id_to_use)
+
+    def mock_endpoint(
+        self,
+        method: str,
+        path: str,
+        response: Dict[str, Any],
+    ) -> None:
+        """
+        Mock an HTTP endpoint.
+
+        :param method: HTTP method (GET, POST, PUT, DELETE, PATCH)
+        :param path: URL path (e.g., "/v1/ats/jobs")
+        :param response: Response dict with 'body', optional 'statusCode', and optional 'headers'
+        """
+        status_code = response.get("statusCode", 200)
+        body = response.get("body")
+        response_headers = response.get("headers", {})
+
+        # Set up Content-Type header for JSON responses if not provided
+        if isinstance(body, dict) and "Content-Type" not in response_headers:
+            response_headers = {**response_headers, "Content-Type": "application/json"}
+
+        # Prepare response content
+        if isinstance(body, (dict, list)):
+            content = json.dumps(body).encode()
+        elif isinstance(body, str):
+            content = body.encode()
+        else:
+            content = b""
+
+        # Create response function that captures request
+        def create_response(request: httpx.Request) -> httpx.Response:
+            # Capture request details
+            query_string = request.url.query.decode() if request.url.query else ""
+            full_path = request.url.path + (f"?{query_string}" if query_string else "")
+
+            # Read body for non-GET requests
+            request_body = None
+            if request.method != "GET":
+                try:
+                    # Try to get content from request
+                    if hasattr(request, "_content"):
+                        body_bytes = request._content
+                    elif hasattr(request, "content"):
+                        body_bytes = request.content
+                    else:
+                        # Try reading from stream
+                        body_bytes = request.read()
+
+                    if body_bytes:
+                        try:
+                            if isinstance(body_bytes, bytes):
+                                request_body = json.loads(body_bytes.decode())
+                            else:
+                                request_body = json.loads(body_bytes)
+                        except (json.JSONDecodeError, UnicodeDecodeError, TypeError):
+                            if isinstance(body_bytes, bytes):
+                                request_body = body_bytes.decode()
+                            else:
+                                request_body = body_bytes
+                except Exception:
+                    # If we can't read the body, that's okay
+                    pass
+
+            captured = CapturedRequest(
+                method=request.method,
+                path=full_path,
+                headers=dict(request.headers),
+                body=request_body,
+            )
+            self._captured_requests.append(captured)
+
+            return httpx.Response(
+                status_code=status_code,
+                headers=response_headers,
+                content=content,
+            )
+
+        # Create the mock route
+        # For GET requests, match any query parameters using regex
+        # For other methods, match exact path
+        base_url = f"https://api.kombo.dev{path}"
+        if method == "GET":
+            # Match the base path, allowing any query parameters
+            route = respx.request(
+                method=method,
+                url__regex=re.compile(f"^{re.escape(base_url)}(\\?.*)?$"),
+            )
+        else:
+            route = respx.request(
+                method=method,
+                url=base_url,
+            )
+
+        route.mock(side_effect=create_response)
+
+    def get_requests(self) -> List[CapturedRequest]:
+        """Get all captured requests."""
+        return list(self._captured_requests)
+
+    def get_last_request(self) -> CapturedRequest:
+        """Get the last captured request."""
+        if not self._captured_requests:
+            raise RuntimeError("No requests captured!")
+        return self._captured_requests[-1]
+
+    def clear(self) -> None:
+        """Clear captured requests and reset mocks."""
+        self._captured_requests.clear()
+        respx.stop()
+        respx.clear()
+        respx.start()  # Restart respx so new mocks can be registered
+
+
+@pytest.fixture(autouse=True)
+def reset_respx():
+    """Reset respx mocks before and after each test."""
+    respx.start()
+    yield
+    respx.stop()
+    respx.clear()
+