Repository navigation
Expand file tree
/
Copy pathconftest.py
More file actions
84 lines (60 loc) · 2.9 KB
/
Copy pathconftest.py
File metadata and controls
84 lines (60 loc) · 2.9 KB
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
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
import logging
from typing import TYPE_CHECKING, Iterator, AsyncIterator
import httpx
import pytest
from pytest_asyncio import is_async_test
from unlayer import Unlayer, AsyncUnlayer, DefaultAioHttpClient
from unlayer._utils import is_dict
if TYPE_CHECKING:
from _pytest.fixtures import FixtureRequest # pyright: ignore[reportPrivateImportUsage]
pytest.register_assert_rewrite("tests.utils")
logging.getLogger("unlayer").setLevel(logging.DEBUG)
# automatically add `pytest.mark.asyncio()` to all of our async tests
# so we don't have to add that boilerplate everywhere
def pytest_collection_modifyitems(items: list[pytest.Function]) -> None:
pytest_asyncio_tests = (item for item in items if is_async_test(item))
session_scope_marker = pytest.mark.asyncio(loop_scope="session")
for async_test in pytest_asyncio_tests:
async_test.add_marker(session_scope_marker, append=False)
# We skip tests that use both the aiohttp client and respx_mock as respx_mock
# doesn't support custom transports.
for item in items:
if "async_client" not in item.fixturenames or "respx_mock" not in item.fixturenames:
continue
if not hasattr(item, "callspec"):
continue
async_client_param = item.callspec.params.get("async_client")
if is_dict(async_client_param) and async_client_param.get("http_client") == "aiohttp":
item.add_marker(pytest.mark.skip(reason="aiohttp client is not compatible with respx_mock"))
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
api_key = "My API Key"
@pytest.fixture(scope="session")
def client(request: FixtureRequest) -> Iterator[Unlayer]:
strict = getattr(request, "param", True)
if not isinstance(strict, bool):
raise TypeError(f"Unexpected fixture parameter type {type(strict)}, expected {bool}")
with Unlayer(base_url=base_url, api_key=api_key, _strict_response_validation=strict) as client:
yield client
@pytest.fixture(scope="session")
async def async_client(request: FixtureRequest) -> AsyncIterator[AsyncUnlayer]:
param = getattr(request, "param", True)
# defaults
strict = True
http_client: None | httpx.AsyncClient = None
if isinstance(param, bool):
strict = param
elif is_dict(param):
strict = param.get("strict", True)
assert isinstance(strict, bool)
http_client_type = param.get("http_client", "httpx")
if http_client_type == "aiohttp":
http_client = DefaultAioHttpClient()
else:
raise TypeError(f"Unexpected fixture parameter type {type(param)}, expected bool or dict")
async with AsyncUnlayer(
base_url=base_url, api_key=api_key, _strict_response_validation=strict, http_client=http_client
) as client:
yield client