-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathtest_rest_client.py
More file actions
161 lines (128 loc) · 6.23 KB
/
Copy pathtest_rest_client.py
File metadata and controls
161 lines (128 loc) · 6.23 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
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
151
152
153
154
155
156
157
158
159
160
161
from unittest.mock import Mock, patch
from graphene_linked_events.rest_client import LinkedEventsApiClient
class TestLinkedEventsApiClientTimeout:
"""Tests for LinkedEventsApiClient timeout configuration."""
def get_base_config(self, timeout=None):
"""Return a base config dict, optionally with timeout."""
config = {
"ROOT": "https://api.example.com/v1/",
"API_KEY": "test-api-key",
"DATA_SOURCE": "test-source",
}
if timeout is not None:
config["TIMEOUT"] = timeout
return config
def test_timeout_from_config(self):
"""Client should use timeout value from config."""
config = self.get_base_config(timeout=90)
client = LinkedEventsApiClient(config)
assert client.timeout == 90
def test_timeout_default_when_not_in_config(self):
"""Client should use default timeout (60s) when not in config."""
config = self.get_base_config()
client = LinkedEventsApiClient(config)
assert client.timeout == 60
def test_timeout_zero_is_allowed(self):
"""Client should accept zero timeout from config."""
config = self.get_base_config(timeout=0)
client = LinkedEventsApiClient(config)
assert client.timeout == 0
@patch("graphene_linked_events.rest_client.requests.request")
def test_retrieve_uses_configured_timeout(self, mock_request):
"""Retrieve method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=200)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.retrieve("event", "123")
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_list_uses_configured_timeout(self, mock_request):
"""List method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=200)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.list("event")
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_create_uses_configured_timeout(self, mock_request):
"""Create method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=201)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.create("event", '{"name": "Test"}')
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_update_uses_configured_timeout(self, mock_request):
"""Update method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=200)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.update("event", "123", '{"name": "Updated"}')
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_delete_uses_configured_timeout(self, mock_request):
"""Delete method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=204)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.delete("event", "123")
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_search_uses_configured_timeout(self, mock_request):
"""Search method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=200)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
# Mock get_actions to return a valid search action
# (search is a special action that doesn't use resource param)
with patch.object(
client,
"get_actions",
return_value={"search": {"method": "GET", "url": client.root + "search/"}},
):
client.search({"q": "test"})
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
@patch("graphene_linked_events.rest_client.requests.request")
def test_upload_uses_configured_timeout(self, mock_request):
"""Upload method should pass configured timeout to requests."""
mock_request.return_value = Mock(status_code=201)
config = self.get_base_config(timeout=45)
client = LinkedEventsApiClient(config)
client.upload("image", {"name": "test"}, {"file": b"data"})
mock_request.assert_called_once()
assert mock_request.call_args.kwargs["timeout"] == 45
class TestLinkedEventsApiClientConfig:
"""Tests for LinkedEventsApiClient configuration."""
def test_client_stores_root_url(self):
"""Client should store root URL from config."""
config = {
"ROOT": "https://api.example.com/v1/",
"API_KEY": "test-key",
"DATA_SOURCE": "test-source",
}
client = LinkedEventsApiClient(config)
assert client.root == "https://api.example.com/v1/"
def test_client_stores_api_key(self):
"""Client should store API key from config."""
config = {
"ROOT": "https://api.example.com/v1/",
"API_KEY": "my-secret-key",
"DATA_SOURCE": "test-source",
}
client = LinkedEventsApiClient(config)
assert client.api_key == "my-secret-key"
def test_client_stores_data_source(self):
"""Client should store data source from config."""
config = {
"ROOT": "https://api.example.com/v1/",
"API_KEY": "test-key",
"DATA_SOURCE": "palvelutarjotin",
}
client = LinkedEventsApiClient(config)
assert client.data_source == "palvelutarjotin"