"""Contract tests for OpenAICompatibleClient. Tests against a mocked compatible server validating all structured-output modes, authentication, retries, metadata capture, schema validation, and credential safety. Requirements: 2.1, 2.2, 2.3, 2.4, 2.5, 2.7, 2.9 """ from __future__ import annotations import json from uuid import uuid4 import httpx import pytest from services.shared.inference.clients.openai_compatible import ( OpenAICompatibleClient, _redact_headers, _resolve_auth_secret, ) from services.shared.inference.models import ( ChatMessage, ErrorCategory, InferenceTarget, ProviderCapabilities, StructuredGenerationRequest, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_target( *, json_schema: bool = True, json_object: bool = False, seed: bool = True, usage: bool = True, auth_secret_ref: str | None = None, auth_scheme: str = "bearer", extra_headers: dict | None = None, extra_body: dict | None = None, max_retries: int = 2, base_url: str = "http://test-vllm:8000", ) -> InferenceTarget: """Build a test InferenceTarget.""" return InferenceTarget( endpoint_id=uuid4(), deployment_id=uuid4(), protocol="openai_chat", base_url=base_url, model="test-model", capabilities=ProviderCapabilities( chat_completions=True, json_schema=json_schema, json_object=json_object, seed=seed, usage=usage, ), auth_secret_ref=auth_secret_ref, auth_scheme=auth_scheme, extra_headers=extra_headers or {}, extra_body=extra_body or {}, max_retries=max_retries, ) def _make_request( *, schema: dict | None = None, temperature: float = 0.0, max_tokens: int = 1024, ) -> StructuredGenerationRequest: """Build a test StructuredGenerationRequest.""" return StructuredGenerationRequest( messages=[ ChatMessage(role="system", content="You are a helpful assistant."), ChatMessage(role="user", content="Extract the data."), ], json_schema=schema, max_output_tokens=max_tokens, temperature=temperature, seed=42, timeout_seconds=30.0, trace_id="test-trace-001", ) _TEST_SCHEMA = { "title": "test_response", "type": "object", "properties": { "answer": {"type": "string"}, "confidence": {"type": "number"}, }, "required": ["answer", "confidence"], } def _valid_response_json() -> str: return json.dumps({"answer": "AAPL beat earnings", "confidence": 0.95}) def _openai_response( content: str, status: int = 200, *, usage: dict | None = None, finish_reason: str = "stop", request_id: str | None = "req-abc-123", ) -> httpx.Response: """Build a fake OpenAI-compatible /v1/chat/completions response.""" body = { "id": "chatcmpl-test", "object": "chat.completion", "choices": [ { "index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": finish_reason, } ], } if usage: body["usage"] = usage headers = {} if request_id: headers["x-request-id"] = request_id return httpx.Response(status, json=body, headers=headers) # =========================================================================== # 11.1: Test /v1/chat/completions using httpx.AsyncClient # =========================================================================== @pytest.mark.asyncio async def test_successful_completion_json_schema(): """Client sends correct payload and parses json_schema response.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["payload"] = json.loads(request.content) captured["headers"] = dict(request.headers) return _openai_response( _valid_response_json(), usage={"prompt_tokens": 50, "completion_tokens": 20}, ) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) request = _make_request(schema=_TEST_SCHEMA) result = await client.generate(request) # Verify URL assert captured["url"] == "http://test-vllm:8000/v1/chat/completions" # Verify payload structure payload = captured["payload"] assert payload["model"] == "test-model" assert payload["temperature"] == 0.0 assert payload["max_tokens"] == 1024 assert payload["seed"] == 42 assert len(payload["messages"]) == 2 assert payload["messages"][0]["role"] == "system" assert payload["messages"][1]["role"] == "user" # Verify response_format for json_schema mode rf = payload["response_format"] assert rf["type"] == "json_schema" assert rf["json_schema"]["name"] == "test_response" assert rf["json_schema"]["strict"] is True assert rf["json_schema"]["schema"] == _TEST_SCHEMA # Verify result assert result.error is None assert result.structured_mode == "json_schema" assert result.parsed == {"answer": "AAPL beat earnings", "confidence": 0.95} assert result.usage.input_tokens == 50 assert result.usage.output_tokens == 20 assert result.request_id == "req-abc-123" assert result.finish_reason == "stop" assert result.schema_valid is True assert result.latency_ms >= 0 await client.close() # =========================================================================== # 11.2: Test Bearer and configurable authentication headers # =========================================================================== @pytest.mark.asyncio async def test_bearer_auth_header(monkeypatch): """Client sends Bearer token from env var.""" monkeypatch.setenv("VLLM_API_KEY", "secret-token-123") captured_headers: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured_headers.update(dict(request.headers)) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(auth_secret_ref="VLLM_API_KEY", auth_scheme="bearer") client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert captured_headers["authorization"] == "Bearer secret-token-123" await client.close() @pytest.mark.asyncio async def test_custom_auth_header(monkeypatch): """Client sends custom auth header scheme.""" monkeypatch.setenv("CUSTOM_KEY", "my-api-key-value") captured_headers: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured_headers.update(dict(request.headers)) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target( auth_secret_ref="CUSTOM_KEY", auth_scheme="X-API-Key" ) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert captured_headers["x-api-key"] == "my-api-key-value" await client.close() @pytest.mark.asyncio async def test_no_auth_when_secret_ref_is_none(): """No Authorization header when auth_secret_ref is None.""" captured_headers: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured_headers.update(dict(request.headers)) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(auth_secret_ref=None) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert "authorization" not in captured_headers await client.close() # =========================================================================== # 11.3: Test standard response_format.json_schema payloads # =========================================================================== @pytest.mark.asyncio async def test_json_schema_payload_structure(): """json_schema mode sends correct response_format structure.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(json_schema=True) client = OpenAICompatibleClient(target, http_client=http) schema = { "title": "extraction", "type": "object", "properties": {"ticker": {"type": "string"}}, "required": ["ticker"], } await client.generate(_make_request(schema=schema)) rf = captured["payload"]["response_format"] assert rf["type"] == "json_schema" assert rf["json_schema"]["name"] == "extraction" assert rf["json_schema"]["strict"] is True assert rf["json_schema"]["schema"] == schema await client.close() # =========================================================================== # 11.4: Test configurable vLLM structured_outputs extra-body payloads # =========================================================================== @pytest.mark.asyncio async def test_vllm_extra_body_inclusion(): """Extra body fields (vLLM structured_outputs) are included in payload.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) extra_body = { "guided_json": {"type": "object", "properties": {"x": {"type": "integer"}}}, "guided_decoding_backend": "outlines", } target = _make_target(extra_body=extra_body) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) payload = captured["payload"] assert payload["guided_json"] == extra_body["guided_json"] assert payload["guided_decoding_backend"] == "outlines" await client.close() # =========================================================================== # 11.5: Test JSON-object and prompt-only fallback policies # =========================================================================== @pytest.mark.asyncio async def test_json_object_fallback(): """Uses json_object mode when target lacks json_schema but has json_object.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(json_schema=False, json_object=True) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert captured["payload"]["response_format"] == {"type": "json_object"} assert result.structured_mode == "json_object" await client.close() @pytest.mark.asyncio async def test_prompt_only_fallback(): """Uses prompt_only mode when target lacks both json_schema and json_object.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(json_schema=False, json_object=False) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) # No response_format in payload assert "response_format" not in captured["payload"] assert result.structured_mode == "prompt_only" await client.close() @pytest.mark.asyncio async def test_no_schema_no_response_format(): """No response_format sent when request has no json_schema.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response("Just a plain response") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(json_schema=True) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=None)) assert "response_format" not in captured["payload"] assert result.structured_mode == "none" assert result.parsed is None await client.close() # =========================================================================== # 11.6: Test metadata capture (request_id, usage, finish_reason, retries) # =========================================================================== @pytest.mark.asyncio async def test_metadata_capture(): """Result captures request_id, usage, finish_reason from response.""" def handler(request: httpx.Request) -> httpx.Response: return _openai_response( _valid_response_json(), usage={"prompt_tokens": 120, "completion_tokens": 45}, finish_reason="length", request_id="req-xyz-789", ) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.request_id == "req-xyz-789" assert result.usage.input_tokens == 120 assert result.usage.output_tokens == 45 assert result.finish_reason == "length" assert result.retries == 0 assert result.error is None assert result.error_category is None await client.close() # =========================================================================== # 11.6 continued: Test retry on 429 rate limit # =========================================================================== @pytest.mark.asyncio async def test_retry_on_429_rate_limit(): """Client retries on 429 and succeeds on subsequent attempt.""" call_count = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal call_count call_count += 1 if call_count == 1: return httpx.Response(429, text="Rate limited") return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=2) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error is None assert result.retries == 1 assert call_count == 2 await client.close() @pytest.mark.asyncio async def test_retry_exhausted_on_500(): """Client returns error after exhausting retries on 500.""" call_count = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal call_count call_count += 1 return httpx.Response(500, text="Internal Server Error") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=2) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error is not None assert result.error_category == ErrorCategory.SERVER_ERROR assert result.retries == 2 assert call_count == 3 # initial + 2 retries await client.close() @pytest.mark.asyncio async def test_retry_on_timeout(): """Client retries on timeout and succeeds on next attempt.""" call_count = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal call_count call_count += 1 if call_count == 1: raise httpx.ReadTimeout("timed out") return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=2) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error is None assert result.retries == 1 await client.close() @pytest.mark.asyncio async def test_timeout_exhausted(): """Client returns timeout error after exhausting retries.""" def handler(request: httpx.Request) -> httpx.Response: raise httpx.ReadTimeout("timed out") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=1) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error_category == ErrorCategory.TIMEOUT assert result.retries == 1 await client.close() # =========================================================================== # 11.7: Test schema validation catches invalid JSON # =========================================================================== @pytest.mark.asyncio async def test_schema_validation_passes_valid_json(): """Schema validation marks valid responses as schema_valid=True.""" def handler(request: httpx.Request) -> httpx.Response: return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.schema_valid is True assert result.parsed is not None await client.close() @pytest.mark.asyncio async def test_schema_validation_catches_invalid_json(): """Schema validation marks responses violating the schema as invalid.""" # Missing required "confidence" field invalid_json = json.dumps({"answer": "test"}) def handler(request: httpx.Request) -> httpx.Response: return _openai_response(invalid_json) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.schema_valid is False assert result.parsed == {"answer": "test"} # Still parsed, but marked invalid await client.close() @pytest.mark.asyncio async def test_unparseable_json_content(): """Non-JSON content in structured mode results in schema_valid=False.""" def handler(request: httpx.Request) -> httpx.Response: return _openai_response("This is not JSON at all") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.schema_valid is False assert result.parsed is None await client.close() # =========================================================================== # 11.8: Additional contract tests # =========================================================================== @pytest.mark.asyncio async def test_authentication_failure_401(): """Client returns auth error on 401 without retrying.""" call_count = 0 def handler(request: httpx.Request) -> httpx.Response: nonlocal call_count call_count += 1 return httpx.Response(401, text="Unauthorized") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=3) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error_category == ErrorCategory.AUTHENTICATION assert call_count == 1 # No retries on auth failure await client.close() @pytest.mark.asyncio async def test_empty_choices_error(): """Client returns error when response has empty choices.""" def handler(request: httpx.Request) -> httpx.Response: body = {"choices": []} return httpx.Response(200, json=body) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error is not None assert "Empty choices" in result.error await client.close() def test_redact_headers(): """Sensitive headers are redacted.""" headers = { "Authorization": "Bearer secret123", "X-API-Key": "key456", "Content-Type": "application/json", } redacted = _redact_headers(headers) assert redacted["Authorization"] == "***REDACTED***" assert redacted["X-API-Key"] == "***REDACTED***" assert redacted["Content-Type"] == "application/json" def test_resolve_auth_secret_from_env(monkeypatch): """Auth secret resolves from environment variable.""" monkeypatch.setenv("MY_SECRET", "resolved-value") assert _resolve_auth_secret("MY_SECRET") == "resolved-value" def test_resolve_auth_secret_returns_none_for_missing(): """Auth secret returns None when env var is not set.""" assert _resolve_auth_secret("NONEXISTENT_VAR_12345") is None def test_resolve_auth_secret_returns_none_for_none_ref(): """Auth secret returns None when ref is None.""" assert _resolve_auth_secret(None) is None @pytest.mark.asyncio async def test_credentials_not_in_repr(): """Client repr does not expose auth secrets.""" target = _make_target(auth_secret_ref="SECRET_KEY") transport = httpx.MockTransport( lambda req: _openai_response(_valid_response_json()) ) http = httpx.AsyncClient(transport=transport) client = OpenAICompatibleClient(target, http_client=http) repr_str = repr(client) assert "SECRET_KEY" not in repr_str assert "secret" not in repr_str.lower() or "auth_secret" not in repr_str await client.close() @pytest.mark.asyncio async def test_connection_error_handling(): """Client returns connection error after exhausting retries.""" def handler(request: httpx.Request) -> httpx.Response: raise httpx.ConnectError("Connection refused") transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(max_retries=1) client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.error_category == ErrorCategory.CONNECTION_ERROR assert result.retries == 1 await client.close() @pytest.mark.asyncio async def test_extra_headers_included(): """Extra headers from target config are included in request.""" captured_headers: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured_headers.update(dict(request.headers)) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(extra_headers={"X-Custom-Header": "custom-value"}) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert captured_headers["x-custom-header"] == "custom-value" await client.close() @pytest.mark.asyncio async def test_seed_included_when_supported(): """Seed is included in payload when target supports it.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(seed=True) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert captured["payload"]["seed"] == 42 await client.close() @pytest.mark.asyncio async def test_seed_excluded_when_not_supported(): """Seed is excluded from payload when target doesn't support it.""" captured: dict = {} def handler(request: httpx.Request) -> httpx.Response: captured["payload"] = json.loads(request.content) return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target(seed=False) client = OpenAICompatibleClient(target, http_client=http) await client.generate(_make_request(schema=_TEST_SCHEMA)) assert "seed" not in captured["payload"] await client.close() @pytest.mark.asyncio async def test_target_stored_in_result(): """InferenceResult includes the target used for the request.""" def handler(request: httpx.Request) -> httpx.Response: return _openai_response(_valid_response_json()) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.endpoint_id == target.endpoint_id assert result.model == "test-model" await client.close() @pytest.mark.asyncio async def test_no_request_id_header(): """Result has None request_id when server doesn't send x-request-id.""" def handler(request: httpx.Request) -> httpx.Response: return _openai_response( _valid_response_json(), request_id=None ) transport = httpx.MockTransport(handler) http = httpx.AsyncClient(transport=transport) target = _make_target() client = OpenAICompatibleClient(target, http_client=http) result = await client.generate(_make_request(schema=_TEST_SCHEMA)) assert result.request_id is None await client.close()