1#-------------------------------------------------------------------------- 2# 3# Copyright (c) Microsoft Corporation. All rights reserved. 4# 5# The MIT License (MIT) 6# 7# Permission is hereby granted, free of charge, to any person obtaining a copy 8# of this software and associated documentation files (the ""Software""), to deal 9# in the Software without restriction, including without limitation the rights 10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 11# copies of the Software, and to permit persons to whom the Software is 12# furnished to do so, subject to the following conditions: 13# 14# The above copyright notice and this permission notice shall be included in 15# all copies or substantial portions of the Software. 16# 17# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 23# THE SOFTWARE. 24# 25#-------------------------------------------------------------------------- 26 27import json 28import httpretty 29try: 30 from http.server import( 31 HTTPServer, 32 BaseHTTPRequestHandler) 33except ImportError: 34 from BaseHTTPServer import HTTPServer 35 from BaseHTTPServer import BaseHTTPRequestHandler 36import os 37import requests 38import re 39import unittest 40try: 41 from unittest import mock 42except ImportError: 43 import mock 44 45from msrest.authentication import ( 46 Authentication, 47 OAuthTokenAuthentication) 48from msrest.universal_http import ( 49 ClientRequest 50) 51from msrest import ( 52 ServiceClient, 53 Configuration) 54from msrest.exceptions import ( 55 TokenExpiredError, 56 ClientRequestError) 57 58import pytest 59 60class TestRuntime(unittest.TestCase): 61 62 @httpretty.activate 63 def test_credential_headers(self): 64 65 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 66 body='[{"title": "Test Data"}]', 67 content_type="application/json") 68 69 token = { 70 'access_token': 'eswfld123kjhn1v5423', 71 'refresh_token': 'asdfkljh23490sdf', 72 'token_type': 'Bearer', 73 'expires_in': '3600', 74 } 75 76 cfg = Configuration("https://my_service.com") 77 cfg.credentials = OAuthTokenAuthentication("client_id", token) 78 79 client = ServiceClient(None, cfg) 80 81 url = client.format_url("/get_endpoint") 82 request = client.get(url, {'check':True}) 83 response = client.send(request) 84 assert 'Authorization' in response.request.headers 85 assert response.request.headers['Authorization'] == 'Bearer eswfld123kjhn1v5423' 86 httpretty.has_request() 87 assert response.json() == [{"title": "Test Data"}] 88 89 # Expiration test 90 91 token['expires_in'] = '-30' 92 cfg.credentials = OAuthTokenAuthentication("client_id", token) 93 client = ServiceClient(None, cfg) 94 url = client.format_url("/get_endpoint") 95 request = client.get(url, {'check':True}) 96 97 with pytest.raises(TokenExpiredError): 98 response = client.send(request) 99 100 @mock.patch.object(requests, 'Session') 101 def test_request_fail(self, mock_requests): 102 103 mock_requests.return_value.request.return_value = mock.Mock(text="text") 104 105 cfg = Configuration("https://my_service.com") 106 cfg.credentials = Authentication() 107 108 client = ServiceClient(None, cfg) 109 url = client.format_url("/get_endpoint") 110 request = client.get(url, {'check':True}) 111 response = client.send(request) 112 113 assert response.text == "text" 114 115 mock_requests.return_value.request.side_effect = requests.RequestException 116 with self.assertRaises(ClientRequestError): 117 client.send(request) 118 119 @httpretty.activate 120 def test_request_proxy(self): 121 # Note that this test requires requests >= 2.8.0 to accept host on proxy 122 123 cfg = Configuration("http://my_service.com") 124 cfg.proxies.add("http://my_service.com", 'http://localhost:57979') 125 cfg.credentials = Authentication() 126 127 httpretty.register_uri(httpretty.GET, "http://localhost:57979/get_endpoint?check=True", 128 body='"Mocked body"', 129 content_type="application/json", 130 status=200) 131 132 client = ServiceClient(None, cfg) 133 url = client.format_url("/get_endpoint") 134 request = client.get(url, {'check':True}) 135 response = client.send(request) 136 assert response.json() == "Mocked body" 137 138 with mock.patch.dict('os.environ', {'HTTP_PROXY': "http://localhost:1987"}): 139 httpretty.register_uri(httpretty.GET, "http://localhost:1987/get_endpoint?check=True", 140 body='"Mocked body"', 141 content_type="application/json", 142 status=200) 143 144 cfg = Configuration("http://my_service.com") 145 client = ServiceClient(None, cfg) 146 url = client.format_url("/get_endpoint") 147 request = client.get(url, {'check':True}) 148 response = client.send(request) 149 assert response.json() == "Mocked body" 150 151 152class TestRedirect(unittest.TestCase): 153 154 def setUp(self): 155 156 cfg = Configuration("https://my_service.com") 157 cfg.retry_policy.backoff_factor=0 158 cfg.redirect_policy.max_redirects=2 159 cfg.credentials = Authentication() 160 161 self.client = ServiceClient(None, cfg) 162 163 return super(TestRedirect, self).setUp() 164 165 @httpretty.activate 166 def test_request_redirect_post(self): 167 168 url = self.client.format_url("/get_endpoint") 169 request = self.client.post(url, {'check':True}) 170 171 httpretty.register_uri(httpretty.GET, 'https://my_service.com/http/success/get/200', status=200) 172 httpretty.register_uri(httpretty.POST, "https://my_service.com/get_endpoint", 173 responses=[ 174 httpretty.Response(body="", status=303, method='POST', location='/http/success/get/200'), 175 ]) 176 177 178 response = self.client.send(request) 179 assert response.status_code == 200, "Should redirect with GET on 303 with location header" 180 assert response.request.method == 'GET' 181 182 assert response.history[0].status_code == 303 183 assert response.history[0].is_redirect 184 185 httpretty.reset() 186 httpretty.register_uri(httpretty.POST, "https://my_service.com/get_endpoint", 187 responses=[ 188 httpretty.Response(body="", status=303, method='POST'), 189 ]) 190 191 response = self.client.send(request) 192 assert response.status_code == 303, "Should not redirect on 303 without location header" 193 assert response.history == [] 194 assert not response.is_redirect 195 196 @httpretty.activate 197 def test_request_redirect_head(self): 198 199 url = self.client.format_url("/get_endpoint") 200 request = self.client.head(url, {'check':True}) 201 202 httpretty.register_uri(httpretty.HEAD, 'https://my_service.com/http/success/200', status=200) 203 httpretty.register_uri(httpretty.HEAD, "https://my_service.com/get_endpoint", 204 responses=[ 205 httpretty.Response(body="", status=307, method='HEAD', location='/http/success/200'), 206 ]) 207 208 209 response = self.client.send(request) 210 assert response.status_code == 200, "Should redirect on 307 with location header" 211 assert response.request.method == 'HEAD' 212 213 assert response.history[0].status_code == 307 214 assert response.history[0].is_redirect 215 216 httpretty.reset() 217 httpretty.register_uri(httpretty.HEAD, "https://my_service.com/get_endpoint", 218 responses=[ 219 httpretty.Response(body="", status=307, method='HEAD'), 220 ]) 221 222 response = self.client.send(request) 223 assert response.status_code == 307, "Should not redirect on 307 without location header" 224 assert response.history == [] 225 assert not response.is_redirect 226 227 @httpretty.activate 228 def test_request_redirect_delete(self): 229 230 url = self.client.format_url("/get_endpoint") 231 request = self.client.delete(url, {'check':True}) 232 233 httpretty.register_uri(httpretty.DELETE, 'https://my_service.com/http/success/200', status=200) 234 httpretty.register_uri(httpretty.DELETE, "https://my_service.com/get_endpoint", 235 responses=[ 236 httpretty.Response(body="", status=307, method='DELETE', location='/http/success/200'), 237 ]) 238 239 240 response = self.client.send(request) 241 assert response.status_code == 200, "Should redirect on 307 with location header" 242 assert response.request.method == 'DELETE' 243 244 assert response.history[0].status_code == 307 245 assert response.history[0].is_redirect 246 247 httpretty.reset() 248 httpretty.register_uri(httpretty.DELETE, "https://my_service.com/get_endpoint", 249 responses=[ 250 httpretty.Response(body="", status=307, method='DELETE'), 251 ]) 252 253 response = self.client.send(request) 254 assert response.status_code == 307, "Should not redirect on 307 without location header" 255 assert response.history == [] 256 assert not response.is_redirect 257 258 @httpretty.activate 259 def test_request_redirect_put(self): 260 261 url = self.client.format_url("/get_endpoint") 262 request = self.client.put(url, {'check':True}) 263 264 httpretty.register_uri(httpretty.PUT, "https://my_service.com/get_endpoint", 265 responses=[ 266 httpretty.Response(body="", status=305, method='PUT', location='/http/success/200'), 267 ]) 268 269 response = self.client.send(request) 270 assert response.status_code == 305, "Should not redirect on 305" 271 assert response.history == [] 272 assert not response.is_redirect 273 274 @httpretty.activate 275 def test_request_redirect_get(self): 276 277 url = self.client.format_url("/get_endpoint") 278 request = self.client.get(url, {'check':True}) 279 280 httpretty.register_uri(httpretty.GET, "https://my_service.com/http/finished", 281 responses=[ 282 httpretty.Response(body="", status=200, method='GET'), 283 ]) 284 285 httpretty.register_uri(httpretty.GET, "https://my_service.com/http/redirect3", 286 responses=[ 287 httpretty.Response(body="", status=307, method='GET', location='/http/finished'), 288 ]) 289 290 httpretty.register_uri(httpretty.GET, "https://my_service.com/http/redirect2", 291 responses=[ 292 httpretty.Response(body="", status=307, method='GET', location='/http/redirect3'), 293 ]) 294 295 httpretty.register_uri(httpretty.GET, "https://my_service.com/http/redirect1", 296 responses=[ 297 httpretty.Response(body="", status=307, method='GET', location='/http/redirect2'), 298 ]) 299 300 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 301 responses=[ 302 httpretty.Response(body="", status=307, method='GET', location='/http/redirect1'), 303 ]) 304 305 with self.assertRaises(ClientRequestError, msg="Should exceed maximum redirects"): 306 self.client.send(request) 307 308 309 310class TestRuntimeRetry(unittest.TestCase): 311 312 def setUp(self): 313 cfg = Configuration("https://my_service.com") 314 cfg.retry_policy.backoff_factor=0 315 creds = Authentication() 316 317 self.client = ServiceClient(creds, cfg) 318 url = self.client.format_url("/get_endpoint") 319 self.request = self.client.get(url, {'check':True}) 320 return super(TestRuntimeRetry, self).setUp() 321 322 @httpretty.activate 323 def test_request_retry_502(self): 324 325 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 326 responses=[ 327 httpretty.Response(body="retry response", status=502), 328 httpretty.Response(body='success response', status=202), 329 ]) 330 331 332 response = self.client.send(self.request) 333 self.assertEqual(response.status_code, 202, msg="Should retry on 502") 334 335 @httpretty.activate 336 def test_request_retry_408(self): 337 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 338 responses=[ 339 httpretty.Response(body="retry response", status=408), 340 httpretty.Response(body='success response', status=202), 341 ]) 342 response = self.client.send(self.request) 343 self.assertEqual(response.status_code, 202, msg="Should retry on 408") 344 345 @httpretty.activate 346 def test_request_retry_3_times(self): 347 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 348 responses=[ 349 httpretty.Response(body="retry response", status=502), 350 httpretty.Response(body="retry response", status=502), 351 httpretty.Response(body="retry response", status=502), 352 httpretty.Response(body='success response', status=202), 353 ]) 354 355 response = self.client.send(self.request) 356 self.assertEqual(response.status_code, 202, msg="Should retry 3 times") 357 358 @httpretty.activate 359 def test_request_retry_max(self): 360 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 361 responses=[ 362 httpretty.Response(body="retry response", status=502), 363 httpretty.Response(body="retry response", status=502), 364 httpretty.Response(body="retry response", status=502), 365 httpretty.Response(body="retry response", status=502), 366 ]) 367 368 with self.assertRaises(ClientRequestError, msg="Max retries reached"): 369 self.client.send(self.request) 370 371 @httpretty.activate 372 def test_request_retry_404(self): 373 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 374 responses=[ 375 httpretty.Response(body="retry response", status=404), 376 httpretty.Response(body='success response', status=202), 377 ]) 378 379 response = self.client.send(self.request) 380 self.assertEqual(response.status_code, 404, msg="Shouldn't retry on 404") 381 382 @httpretty.activate 383 def test_request_retry_501(self): 384 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 385 responses=[ 386 httpretty.Response(body="retry response", status=501), 387 httpretty.Response(body='success response', status=202), 388 ]) 389 390 response = self.client.send(self.request) 391 self.assertEqual(response.status_code, 501, msg="Shouldn't retry on 501") 392 393 @httpretty.activate 394 def test_request_retry_505(self): 395 httpretty.register_uri(httpretty.GET, "https://my_service.com/get_endpoint", 396 responses=[ 397 httpretty.Response(body="retry response", status=505), 398 httpretty.Response(body='success response', status=202), 399 ]) 400 401 response = self.client.send(self.request) 402 self.assertEqual(response.status_code, 505, msg="Shouldn't retry on 505") 403 404 405if __name__ == '__main__': 406 unittest.main()