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
|
import httpx
import pytest
import respx
import gitlab
@pytest.fixture(scope="module")
def api_url() -> str:
return "https://gitlab.example.com/api/graphql"
@pytest.fixture
def gl_gql() -> gitlab.GraphQL:
return gitlab.GraphQL("https://gitlab.example.com")
@pytest.fixture
def gl_async_gql() -> gitlab.AsyncGraphQL:
return gitlab.AsyncGraphQL("https://gitlab.example.com")
def test_import_error_includes_message(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(gitlab.client, "_GQL_INSTALLED", False)
with pytest.raises(ImportError, match="GraphQL client could not be initialized"):
gitlab.GraphQL()
@pytest.mark.anyio
async def test_async_import_error_includes_message(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(gitlab.client, "_GQL_INSTALLED", False)
with pytest.raises(ImportError, match="GraphQL client could not be initialized"):
gitlab.AsyncGraphQL()
def test_graphql_as_context_manager_exits():
with gitlab.GraphQL() as gl:
assert isinstance(gl, gitlab.GraphQL)
@pytest.mark.anyio
async def test_async_graphql_as_context_manager_aexits():
async with gitlab.AsyncGraphQL() as gl:
assert isinstance(gl, gitlab.AsyncGraphQL)
def test_graphql_retries_on_429_response(
gl_gql: gitlab.GraphQL, respx_mock: respx.MockRouter
):
url = "https://gitlab.example.com/api/graphql"
responses = [
httpx.Response(429, headers={"retry-after": "1"}),
httpx.Response(
200, json={"data": {"currentUser": {"id": "gid://gitlab/User/1"}}}
),
]
respx_mock.post(url).mock(side_effect=responses)
gl_gql.execute("query {currentUser {id}}")
@pytest.mark.anyio
async def test_async_graphql_retries_on_429_response(
api_url: str, gl_async_gql: gitlab.AsyncGraphQL, respx_mock: respx.MockRouter
):
responses = [
httpx.Response(429, headers={"retry-after": "1"}),
httpx.Response(
200, json={"data": {"currentUser": {"id": "gid://gitlab/User/1"}}}
),
]
respx_mock.post(api_url).mock(side_effect=responses)
await gl_async_gql.execute("query {currentUser {id}}")
def test_graphql_raises_when_max_retries_exceeded(
api_url: str, respx_mock: respx.MockRouter
):
responses = [httpx.Response(502), httpx.Response(502), httpx.Response(502)]
respx_mock.post(api_url).mock(side_effect=responses)
gl_gql = gitlab.GraphQL(
"https://gitlab.example.com", max_retries=1, retry_transient_errors=True
)
with pytest.raises(gitlab.GitlabHttpError):
gl_gql.execute("query {currentUser {id}}")
@pytest.mark.anyio
async def test_async_graphql_raises_when_max_retries_exceeded(
api_url: str, respx_mock: respx.MockRouter
):
responses = [httpx.Response(502), httpx.Response(502), httpx.Response(502)]
respx_mock.post(api_url).mock(side_effect=responses)
gl_async_gql = gitlab.AsyncGraphQL(
"https://gitlab.example.com", max_retries=1, retry_transient_errors=True
)
with pytest.raises(gitlab.GitlabHttpError):
await gl_async_gql.execute("query {currentUser {id}}")
def test_graphql_raises_on_401_response(
api_url: str, gl_gql: gitlab.GraphQL, respx_mock: respx.MockRouter
):
respx_mock.post(api_url).mock(return_value=httpx.Response(401))
with pytest.raises(gitlab.GitlabAuthenticationError):
gl_gql.execute("query {currentUser {id}}")
@pytest.mark.anyio
async def test_async_graphql_raises_on_401_response(
api_url: str, gl_async_gql: gitlab.AsyncGraphQL, respx_mock: respx.MockRouter
):
respx_mock.post(api_url).mock(return_value=httpx.Response(401))
with pytest.raises(gitlab.GitlabAuthenticationError):
await gl_async_gql.execute("query {currentUser {id}}")
|
