coproxy 0.2.0

OpenAI-compatible API proxy backed by GitHub Copilot
Documentation 
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

from __future__ import annotations

from typing import Any, cast

import pytest
from openai import APIStatusError, AsyncOpenAI, BadRequestError, OpenAI
from openai.types.responses import Response


def _assert_not_supported_error(exc: BadRequestError) -> None:
    assert exc.status_code == 400
    assert isinstance(exc.body, dict)
    body = cast(dict[str, Any], exc.body)
    assert body.get("type") == "invalid_request_error"
    assert body.get("code") == "not_supported"
    assert isinstance(body.get("message"), str)


def _assert_response_shape(response: Response, expected_model_prefix: str) -> None:
    assert response.object == "response"
    assert isinstance(response.id, str)
    assert response.id
    assert isinstance(response.created_at, (float, int))
    assert isinstance(response.model, str)
    assert response.model.startswith(expected_model_prefix)


def test_embeddings_not_supported(client: OpenAI) -> None:
    with pytest.raises(BadRequestError) as exc_info:
        client.embeddings.create(
            model="text-embedding-3-small",
            input="The quick brown fox jumped over the lazy dog",
        )

    _assert_not_supported_error(exc_info.value)


def test_responses_create_supported_or_model_gated(
    client: OpenAI, responses_available: bool, responses_model: str
) -> None:
    if not responses_available:
        pytest.skip(
            "Skipping responses success tests: GHCP responses model is unavailable"
        )

    response = client.responses.create(
        model=responses_model,
        input="Reply with pong",
    )
    _assert_response_shape(response, responses_model)


def test_responses_raw_response_create_supported_or_model_gated(
    client: OpenAI, responses_available: bool, responses_model: str
) -> None:
    if not responses_available:
        pytest.skip(
            "Skipping responses success tests: GHCP responses model is unavailable"
        )

    http_response = client.responses.with_raw_response.create(
        model=responses_model,
        input="Reply with pong",
    )

    assert http_response.is_closed is True
    assert http_response.http_request.headers.get("X-Stainless-Lang") == "python"
    parsed = http_response.parse()
    _assert_response_shape(parsed, responses_model)


def test_responses_streaming_response_create_supported_or_model_gated(
    client: OpenAI, responses_available: bool, responses_model: str
) -> None:
    if not responses_available:
        pytest.skip(
            "Skipping responses success tests: GHCP responses model is unavailable"
        )

    with client.responses.with_streaming_response.create(
        model=responses_model,
        input="Reply with pong",
    ) as http_response:
        assert not http_response.is_closed
        assert http_response.http_request.headers.get("X-Stainless-Lang") == "python"
        parsed = http_response.parse()
        _assert_response_shape(parsed, responses_model)

    assert cast(Any, http_response.is_closed) is True


def test_responses_retrieve_not_found_or_not_supported(client: OpenAI) -> None:
    with pytest.raises(APIStatusError) as exc_info:
        client.responses.retrieve("resp_compat_test")

    assert exc_info.value.status_code in {400, 404}


@pytest.mark.asyncio
async def test_async_embeddings_not_supported(async_client: AsyncOpenAI) -> None:
    with pytest.raises(BadRequestError) as exc_info:
        await async_client.embeddings.create(
            model="text-embedding-3-small",
            input="The quick brown fox jumped over the lazy dog",
        )

    _assert_not_supported_error(exc_info.value)


@pytest.mark.asyncio
async def test_async_responses_create_supported_or_model_gated(
    async_client: AsyncOpenAI, responses_available: bool, responses_model: str
) -> None:
    if not responses_available:
        pytest.skip(
            "Skipping responses success tests: GHCP responses model is unavailable"
        )

    response = await async_client.responses.create(
        model=responses_model,
        input="Reply with pong",
    )
    _assert_response_shape(response, responses_model)


@pytest.mark.asyncio
async def test_async_responses_stream_true_supported_or_model_gated(
    async_client: AsyncOpenAI, responses_available: bool, responses_model: str
) -> None:
    if not responses_available:
        pytest.skip(
            "Skipping responses success tests: GHCP responses model is unavailable"
        )

    stream = await async_client.responses.create(
        model=responses_model,
        input="Reply with pong",
        stream=True,
    )
    await stream.response.aclose()


@pytest.mark.asyncio
async def test_async_responses_retrieve_not_supported(
    async_client: AsyncOpenAI,
) -> None:
    with pytest.raises(APIStatusError) as exc_info:
        await async_client.responses.retrieve("resp_compat_test")

    assert exc_info.value.status_code in {400, 404}