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

from __future__ import annotations

from typing import Any, cast

import pytest
from openai import APIStatusError, AsyncOpenAI, OpenAI
from openai.types.chat import ChatCompletion, ChatCompletionChunk


def _basic_messages() -> list[dict[str, str]]:
    return [{"role": "user", "content": "Reply with exactly: pong"}]


def _assert_completion_shape(completion: ChatCompletion) -> None:
    assert completion.object == "chat.completion"
    assert isinstance(completion.id, str)
    assert isinstance(completion.created, int)
    assert isinstance(completion.model, str)
    assert completion.choices
    assert completion.choices[0].message.role == "assistant"


def test_chat_completion_create_basic(
    client: OpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    completion = client.chat.completions.create(
        model=compat_model,
        messages=_basic_messages(),
    )
    _assert_completion_shape(completion)


def test_chat_completion_raw_response(
    client: OpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    response = client.chat.completions.with_raw_response.create(
        model=compat_model,
        messages=_basic_messages(),
    )
    assert response.is_closed is True
    assert response.http_request.headers.get("X-Stainless-Lang") == "python"
    completion = response.parse()
    _assert_completion_shape(completion)


def test_chat_completion_streaming_response_wrapper(
    client: OpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    with client.chat.completions.with_streaming_response.create(
        model=compat_model,
        messages=_basic_messages(),
    ) as response:
        assert not response.is_closed
        assert response.http_request.headers.get("X-Stainless-Lang") == "python"
        completion = response.parse()
        _assert_completion_shape(completion)

    assert cast(Any, response.is_closed) is True


def test_chat_completion_stream_true(
    client: OpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    stream = client.chat.completions.create(
        model=compat_model,
        messages=_basic_messages(),
        stream=True,
    )

    saw_chunk = False
    for chunk in stream:
        assert isinstance(chunk, ChatCompletionChunk)
        saw_chunk = True
    assert saw_chunk


def test_chat_completion_missing_messages_is_bad_request(
    client: OpenAI, compat_model: str
) -> None:
    with pytest.raises(APIStatusError) as exc_info:
        client.chat.completions.create(model=compat_model, messages=[])

    err = exc_info.value
    assert err.status_code == 400
    assert isinstance(err.body, dict)
    body = cast(dict[str, Any], err.body)
    assert body.get("type") == "invalid_request_error"
    assert isinstance(body.get("message"), str)


@pytest.mark.asyncio
async def test_async_chat_completion_create_basic(
    async_client: AsyncOpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    completion = await async_client.chat.completions.create(
        model=compat_model,
        messages=_basic_messages(),
    )
    _assert_completion_shape(completion)


@pytest.mark.asyncio
async def test_async_chat_completion_stream_true(
    async_client: AsyncOpenAI, chat_available: bool, compat_model: str
) -> None:
    if not chat_available:
        pytest.skip("Skipping chat success tests: GHCP auth is unavailable")

    stream = await async_client.chat.completions.create(
        model=compat_model,
        messages=_basic_messages(),
        stream=True,
    )

    saw_chunk = False
    async for chunk in stream:
        assert isinstance(chunk, ChatCompletionChunk)
        saw_chunk = True
    assert saw_chunk