1# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
 2
 3from __future__ import annotations
 4
 5import os
 6import logging
 7from typing import TYPE_CHECKING, Iterator, AsyncIterator
 8
 9import httpx
10import pytest
11from pytest_asyncio import is_async_test
12
13from openai import OpenAI, AsyncOpenAI, DefaultAioHttpClient
14from openai._utils import is_dict
15
16if TYPE_CHECKING:
17    from _pytest.fixtures import FixtureRequest  # pyright: ignore[reportPrivateImportUsage]
18
19pytest.register_assert_rewrite("tests.utils")
20
21logging.getLogger("openai").setLevel(logging.DEBUG)
22
23
24# automatically add `pytest.mark.asyncio()` to all of our async tests
25# so we don't have to add that boilerplate everywhere
26def pytest_collection_modifyitems(items: list[pytest.Function]) -> None:
27    pytest_asyncio_tests = (item for item in items if is_async_test(item))
28    session_scope_marker = pytest.mark.asyncio(loop_scope="session")
29    for async_test in pytest_asyncio_tests:
30        async_test.add_marker(session_scope_marker, append=False)
31
32    # We skip tests that use both the aiohttp client and respx_mock as respx_mock
33    # doesn't support custom transports.
34    for item in items:
35        if "async_client" not in item.fixturenames or "respx_mock" not in item.fixturenames:
36            continue
37
38        if not hasattr(item, "callspec"):
39            continue
40
41        async_client_param = item.callspec.params.get("async_client")
42        if is_dict(async_client_param) and async_client_param.get("http_client") == "aiohttp":
43            item.add_marker(pytest.mark.skip(reason="aiohttp client is not compatible with respx_mock"))
44
45
46base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
47
48api_key = "My API Key"
49
50
51@pytest.fixture(scope="session")
52def client(request: FixtureRequest) -> Iterator[OpenAI]:
53    strict = getattr(request, "param", True)
54    if not isinstance(strict, bool):
55        raise TypeError(f"Unexpected fixture parameter type {type(strict)}, expected {bool}")
56
57    with OpenAI(base_url=base_url, api_key=api_key, _strict_response_validation=strict) as client:
58        yield client
59
60
61@pytest.fixture(scope="session")
62async def async_client(request: FixtureRequest) -> AsyncIterator[AsyncOpenAI]:
63    param = getattr(request, "param", True)
64
65    # defaults
66    strict = True
67    http_client: None | httpx.AsyncClient = None
68
69    if isinstance(param, bool):
70        strict = param
71    elif is_dict(param):
72        strict = param.get("strict", True)
73        assert isinstance(strict, bool)
74
75        http_client_type = param.get("http_client", "httpx")
76        if http_client_type == "aiohttp":
77            http_client = DefaultAioHttpClient()
78    else:
79        raise TypeError(f"Unexpected fixture parameter type {type(param)}, expected bool or dict")
80
81    async with AsyncOpenAI(
82        base_url=base_url, api_key=api_key, _strict_response_validation=strict, http_client=http_client
83    ) as client:
84        yield client