|
24 | 24 | from google.auth import _helpers |
25 | 25 | from google.auth import environment_vars |
26 | 26 | from google.auth import exceptions |
| 27 | +from google.auth import transport |
27 | 28 | from google.auth.compute_engine import _metadata |
28 | 29 |
|
29 | 30 | PATH = 'instance/service-accounts/default' |
30 | 31 |
|
31 | 32 |
|
32 | | -@pytest.fixture |
33 | | -def mock_request(): |
34 | | - request_mock = mock.Mock() |
| 33 | +def make_request(data, status=http_client.OK, headers=None): |
| 34 | + response = mock.create_autospec(transport.Response, instance=True) |
| 35 | + response.status = status |
| 36 | + response.data = _helpers.to_bytes(data) |
| 37 | + response.headers = headers or {} |
35 | 38 |
|
36 | | - def set_response(data, status=http_client.OK, headers=None): |
37 | | - response = mock.Mock() |
38 | | - response.status = status |
39 | | - response.data = _helpers.to_bytes(data) |
40 | | - response.headers = headers or {} |
41 | | - request_mock.return_value = response |
42 | | - return request_mock |
| 39 | + request = mock.create_autospec(transport.Request) |
| 40 | + request.return_value = response |
43 | 41 |
|
44 | | - yield set_response |
| 42 | + return request |
45 | 43 |
|
46 | 44 |
|
47 | | -def test_ping_success(mock_request): |
48 | | - request_mock = mock_request('', headers=_metadata._METADATA_HEADERS) |
| 45 | +def test_ping_success(): |
| 46 | + request = make_request('', headers=_metadata._METADATA_HEADERS) |
49 | 47 |
|
50 | | - assert _metadata.ping(request_mock) |
| 48 | + assert _metadata.ping(request) |
51 | 49 |
|
52 | | - request_mock.assert_called_once_with( |
| 50 | + request.assert_called_once_with( |
53 | 51 | method='GET', |
54 | 52 | url=_metadata._METADATA_IP_ROOT, |
55 | 53 | headers=_metadata._METADATA_HEADERS, |
56 | 54 | timeout=_metadata._METADATA_DEFAULT_TIMEOUT) |
57 | 55 |
|
58 | 56 |
|
59 | | -def test_ping_failure_bad_flavor(mock_request): |
60 | | - request_mock = mock_request( |
| 57 | +def test_ping_failure_bad_flavor(): |
| 58 | + request = make_request( |
61 | 59 | '', headers={_metadata._METADATA_FLAVOR_HEADER: 'meep'}) |
62 | 60 |
|
63 | | - assert not _metadata.ping(request_mock) |
| 61 | + assert not _metadata.ping(request) |
64 | 62 |
|
65 | 63 |
|
66 | | -def test_ping_failure_connection_failed(mock_request): |
67 | | - request_mock = mock_request('') |
68 | | - request_mock.side_effect = exceptions.TransportError() |
| 64 | +def test_ping_failure_connection_failed(): |
| 65 | + request = make_request('') |
| 66 | + request.side_effect = exceptions.TransportError() |
69 | 67 |
|
70 | | - assert not _metadata.ping(request_mock) |
| 68 | + assert not _metadata.ping(request) |
71 | 69 |
|
72 | 70 |
|
73 | | -def test_ping_success_custom_root(mock_request): |
74 | | - request_mock = mock_request('', headers=_metadata._METADATA_HEADERS) |
| 71 | +def test_ping_success_custom_root(): |
| 72 | + request = make_request('', headers=_metadata._METADATA_HEADERS) |
75 | 73 |
|
76 | 74 | fake_ip = '1.2.3.4' |
77 | 75 | os.environ[environment_vars.GCE_METADATA_IP] = fake_ip |
78 | 76 | reload_module(_metadata) |
79 | 77 |
|
80 | 78 | try: |
81 | | - assert _metadata.ping(request_mock) |
| 79 | + assert _metadata.ping(request) |
82 | 80 | finally: |
83 | 81 | del os.environ[environment_vars.GCE_METADATA_IP] |
84 | 82 | reload_module(_metadata) |
85 | 83 |
|
86 | | - request_mock.assert_called_once_with( |
| 84 | + request.assert_called_once_with( |
87 | 85 | method='GET', |
88 | 86 | url='http://' + fake_ip, |
89 | 87 | headers=_metadata._METADATA_HEADERS, |
90 | 88 | timeout=_metadata._METADATA_DEFAULT_TIMEOUT) |
91 | 89 |
|
92 | 90 |
|
93 | | -def test_get_success_json(mock_request): |
| 91 | +def test_get_success_json(): |
94 | 92 | key, value = 'foo', 'bar' |
95 | 93 |
|
96 | 94 | data = json.dumps({key: value}) |
97 | | - request_mock = mock_request( |
| 95 | + request = make_request( |
98 | 96 | data, headers={'content-type': 'application/json'}) |
99 | 97 |
|
100 | | - result = _metadata.get(request_mock, PATH) |
| 98 | + result = _metadata.get(request, PATH) |
101 | 99 |
|
102 | | - request_mock.assert_called_once_with( |
| 100 | + request.assert_called_once_with( |
103 | 101 | method='GET', |
104 | 102 | url=_metadata._METADATA_ROOT + PATH, |
105 | 103 | headers=_metadata._METADATA_HEADERS) |
106 | 104 | assert result[key] == value |
107 | 105 |
|
108 | 106 |
|
109 | | -def test_get_success_text(mock_request): |
| 107 | +def test_get_success_text(): |
110 | 108 | data = 'foobar' |
111 | | - request_mock = mock_request(data, headers={'content-type': 'text/plain'}) |
| 109 | + request = make_request(data, headers={'content-type': 'text/plain'}) |
112 | 110 |
|
113 | | - result = _metadata.get(request_mock, PATH) |
| 111 | + result = _metadata.get(request, PATH) |
114 | 112 |
|
115 | | - request_mock.assert_called_once_with( |
| 113 | + request.assert_called_once_with( |
116 | 114 | method='GET', |
117 | 115 | url=_metadata._METADATA_ROOT + PATH, |
118 | 116 | headers=_metadata._METADATA_HEADERS) |
119 | 117 | assert result == data |
120 | 118 |
|
121 | 119 |
|
122 | | -def test_get_success_custom_root(mock_request): |
123 | | - request_mock = mock_request( |
| 120 | +def test_get_success_custom_root(): |
| 121 | + request = make_request( |
124 | 122 | '{}', headers={'content-type': 'application/json'}) |
125 | 123 |
|
126 | 124 | fake_root = 'another.metadata.service' |
127 | 125 | os.environ[environment_vars.GCE_METADATA_ROOT] = fake_root |
128 | 126 | reload_module(_metadata) |
129 | 127 |
|
130 | 128 | try: |
131 | | - _metadata.get(request_mock, PATH) |
| 129 | + _metadata.get(request, PATH) |
132 | 130 | finally: |
133 | 131 | del os.environ[environment_vars.GCE_METADATA_ROOT] |
134 | 132 | reload_module(_metadata) |
135 | 133 |
|
136 | | - request_mock.assert_called_once_with( |
| 134 | + request.assert_called_once_with( |
137 | 135 | method='GET', |
138 | 136 | url='http://{}/computeMetadata/v1/{}'.format(fake_root, PATH), |
139 | 137 | headers=_metadata._METADATA_HEADERS) |
140 | 138 |
|
141 | 139 |
|
142 | | -def test_get_failure(mock_request): |
143 | | - request_mock = mock_request( |
| 140 | +def test_get_failure(): |
| 141 | + request = make_request( |
144 | 142 | 'Metadata error', status=http_client.NOT_FOUND) |
145 | 143 |
|
146 | 144 | with pytest.raises(exceptions.TransportError) as excinfo: |
147 | | - _metadata.get(request_mock, PATH) |
| 145 | + _metadata.get(request, PATH) |
148 | 146 |
|
149 | 147 | assert excinfo.match(r'Metadata error') |
150 | 148 |
|
151 | | - request_mock.assert_called_once_with( |
| 149 | + request.assert_called_once_with( |
152 | 150 | method='GET', |
153 | 151 | url=_metadata._METADATA_ROOT + PATH, |
154 | 152 | headers=_metadata._METADATA_HEADERS) |
155 | 153 |
|
156 | 154 |
|
157 | | -def test_get_failure_bad_json(mock_request): |
158 | | - request_mock = mock_request( |
| 155 | +def test_get_failure_bad_json(): |
| 156 | + request = make_request( |
159 | 157 | '{', headers={'content-type': 'application/json'}) |
160 | 158 |
|
161 | 159 | with pytest.raises(exceptions.TransportError) as excinfo: |
162 | | - _metadata.get(request_mock, PATH) |
| 160 | + _metadata.get(request, PATH) |
163 | 161 |
|
164 | 162 | assert excinfo.match(r'invalid JSON') |
165 | 163 |
|
166 | | - request_mock.assert_called_once_with( |
| 164 | + request.assert_called_once_with( |
167 | 165 | method='GET', |
168 | 166 | url=_metadata._METADATA_ROOT + PATH, |
169 | 167 | headers=_metadata._METADATA_HEADERS) |
170 | 168 |
|
171 | 169 |
|
172 | | -def test_get_project_id(mock_request): |
| 170 | +def test_get_project_id(): |
173 | 171 | project = 'example-project' |
174 | | - request_mock = mock_request( |
| 172 | + request = make_request( |
175 | 173 | project, headers={'content-type': 'text/plain'}) |
176 | 174 |
|
177 | | - project_id = _metadata.get_project_id(request_mock) |
| 175 | + project_id = _metadata.get_project_id(request) |
178 | 176 |
|
179 | | - request_mock.assert_called_once_with( |
| 177 | + request.assert_called_once_with( |
180 | 178 | method='GET', |
181 | 179 | url=_metadata._METADATA_ROOT + 'project/project-id', |
182 | 180 | headers=_metadata._METADATA_HEADERS) |
183 | 181 | assert project_id == project |
184 | 182 |
|
185 | 183 |
|
186 | 184 | @mock.patch('google.auth._helpers.utcnow', return_value=datetime.datetime.min) |
187 | | -def test_get_service_account_token(utcnow, mock_request): |
| 185 | +def test_get_service_account_token(utcnow): |
188 | 186 | ttl = 500 |
189 | | - request_mock = mock_request( |
| 187 | + request = make_request( |
190 | 188 | json.dumps({'access_token': 'token', 'expires_in': ttl}), |
191 | 189 | headers={'content-type': 'application/json'}) |
192 | 190 |
|
193 | | - token, expiry = _metadata.get_service_account_token(request_mock) |
| 191 | + token, expiry = _metadata.get_service_account_token(request) |
194 | 192 |
|
195 | | - request_mock.assert_called_once_with( |
| 193 | + request.assert_called_once_with( |
196 | 194 | method='GET', |
197 | 195 | url=_metadata._METADATA_ROOT + PATH + '/token', |
198 | 196 | headers=_metadata._METADATA_HEADERS) |
199 | 197 | assert token == 'token' |
200 | 198 | assert expiry == utcnow() + datetime.timedelta(seconds=ttl) |
201 | 199 |
|
202 | 200 |
|
203 | | -def test_get_service_account_info(mock_request): |
| 201 | +def test_get_service_account_info(): |
204 | 202 | key, value = 'foo', 'bar' |
205 | | - request_mock = mock_request( |
| 203 | + request = make_request( |
206 | 204 | json.dumps({key: value}), |
207 | 205 | headers={'content-type': 'application/json'}) |
208 | 206 |
|
209 | | - info = _metadata.get_service_account_info(request_mock) |
| 207 | + info = _metadata.get_service_account_info(request) |
210 | 208 |
|
211 | | - request_mock.assert_called_once_with( |
| 209 | + request.assert_called_once_with( |
212 | 210 | method='GET', |
213 | 211 | url=_metadata._METADATA_ROOT + PATH + '/?recursive=true', |
214 | 212 | headers=_metadata._METADATA_HEADERS) |
|
0 commit comments