KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use requests_response method in localstack
Best Python code snippet using localstack_python
connection.py
Source:connection.py 
1# coding: utf82# Copyright 2014-2015 Vincent Jacques <vincent@vincent-jacques.net>3import datetime4import hashlib5import hmac6import json7import urlparse8import time9import requests10import LowVoltage as _lv11import LowVoltage.testing as _tst12import LowVoltage.exceptions as _exn13from . import retry_policies14class Connection(object):15 """16 The main entry point of the package.17 :param region: the identifier of the AWS region you want to connect to. Like ``"us-west-2"`` or ``"eu-west-1"``.18 :param credentials: a credentials providers. See :mod:`.credentials`.19 :param endpoint:20 the HTTP endpoint you want to connect to.21 Typically useful to connect to `DynamoDB Local <http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Tools.DynamoDBLocal.html>`__22 with ``endpoint="http://localhost:8000/"``.23 If left ``None``, it will be computed from the region.24 :param retry_policy: a retry policy. See :mod:`.retry_policies`. If left ``None``, the :obj:`~.retry_policies.DEFAULT` retry policy will be used.25 :param requests_session: a ``Session`` object from the `python-requests <http://python-requests.org>`__ library. Typically not used. Leave it to ``None`` and one will be created for you.26 """27 def __init__(self, region, credentials, endpoint=None, retry_policy=None, requests_session=None):28 if endpoint is None:29 endpoint = "https://dynamodb.{}.amazonaws.com/".format(region)30 if retry_policy is None:31 retry_policy = retry_policies.DEFAULT32 if requests_session is None:33 requests_session = requests.Session()34 self.__region = region35 self.__credentials = credentials36 self.__endpoint = endpoint37 self.__host = urlparse.urlparse(self.__endpoint).hostname38 self.__retry_policy = retry_policy39 self.__session = requests_session40 # Dependency injection through monkey-patching41 self.__signer = Signer(self.__region, self.__host)42 self.__responder = Responder()43 self.__now = datetime.datetime.utcnow44 def __call__(self, action):45 """46 Send requests and return responses.47 """48 errors = []49 while True:50 try:51 return self.__request_once(action)52 except _exn.Error as e:53 if e.retryable:54 errors.append(e)55 delay = self.__retry_policy.retry(action, errors)56 if delay is None:57 raise58 else:59 time.sleep(delay)60 else:61 raise62 def __request_once(self, action):63 key, secret, token = self.__credentials.get()64 payload = json.dumps(action.payload)65 headers = self.__signer(key, secret, self.__now(), action.name, payload)66 if token is not None:67 headers["X-Amz-Security-Token"] = token68 try:69 r = self.__session.post(self.__endpoint, data=payload, headers=headers)70 except requests.exceptions.RequestException as e:71 raise _exn.NetworkError(e)72 except Exception as e:73 raise _exn.UnknownError(e)74 return self.__responder(action.response_class, r)75class ConnectionUnitTests(_tst.UnitTestsWithMocks):76 def setUp(self):77 super(ConnectionUnitTests, self).setUp()78 self.credentials = self.mocks.create("credentials")79 self.retry_policy = self.mocks.create("retry_policy")80 self.session = self.mocks.create("session")81 self.connection = Connection(82 region="us-west-2",83 credentials=self.credentials.object,84 endpoint="http://endpoint.com:8000/",85 retry_policy=self.retry_policy.object,86 requests_session=self.session.object,87 )88 self.now = self.mocks.replace("self.connection._Connection__now")89 self.signer = self.mocks.replace("self.connection._Connection__signer")90 self.responder = self.mocks.replace("self.connection._Connection__responder")91 self.action = self.mocks.create("action")92 def test_identification_with_token(self):93 self.credentials.expect.get().andReturn(("a", "b", "t"))94 self.action.expect.payload.andReturn({"d": "e"})95 self.now.expect().andReturn("f")96 self.action.expect.name.andReturn("c")97 self.signer.expect("a", "b", "f", "c", '{"d": "e"}').andReturn({"g": "h"})98 self.session.expect.post("http://endpoint.com:8000/", data='{"d": "e"}', headers={"g": "h", "X-Amz-Security-Token": "t"}).andReturn("i")99 self.action.expect.response_class.andReturn("j")100 self.responder.expect("j", "i").andReturn("k")101 self.assertEqual(self.connection(self.action.object), "k")102 def __expect_post(self):103 self.credentials.expect.get().andReturn(("a", "b", None))104 self.action.expect.payload.andReturn({"d": "e"})105 self.now.expect().andReturn("f")106 self.action.expect.name.andReturn("c")107 self.signer.expect("a", "b", "f", "c", '{"d": "e"}').andReturn({"g": "h"})108 return self.session.expect.post("http://endpoint.com:8000/", data='{"d": "e"}', headers={"g": "h"})109 def test_success_on_first_try(self):110 self.__expect_post().andReturn("i")111 self.action.expect.response_class.andReturn("j")112 self.responder.expect("j", "i").andReturn("k")113 self.assertEqual(self.connection(self.action.object), "k")114 def test_success_on_fourth_try(self):115 self.__expect_post().andReturn("i")116 self.action.expect.response_class.andReturn("j")117 exception1 = _exn.ProvisionedThroughputExceededException()118 self.responder.expect("j", "i").andRaise(exception1)119 self.retry_policy.expect.retry(self.action.object, [exception1]).andReturn(0)120 self.__expect_post().andReturn("k")121 self.action.expect.response_class.andReturn("l")122 exception2 = _exn.ProvisionedThroughputExceededException()123 self.responder.expect("l", "k").andRaise(exception2)124 self.retry_policy.expect.retry(self.action.object, [exception1, exception2]).andReturn(0)125 self.__expect_post().andReturn("m")126 self.action.expect.response_class.andReturn("n")127 exception3 = _exn.ProvisionedThroughputExceededException()128 self.responder.expect("n", "m").andRaise(exception3)129 self.retry_policy.expect.retry(self.action.object, [exception1, exception2, exception3]).andReturn(0)130 self.__expect_post().andReturn("o")131 self.action.expect.response_class.andReturn("p")132 self.responder.expect("p", "o").andReturn("q")133 self.assertEqual(self.connection(self.action.object), "q")134 def test_failure_on_second_try(self):135 self.__expect_post().andReturn("i")136 self.action.expect.response_class.andReturn("j")137 exception1 = _exn.ProvisionedThroughputExceededException()138 self.responder.expect("j", "i").andRaise(exception1)139 self.retry_policy.expect.retry(self.action.object, [exception1]).andReturn(0)140 self.__expect_post().andReturn("k")141 self.action.expect.response_class.andReturn("l")142 exception2 = _exn.UnknownClientError()143 self.responder.expect("l", "k").andRaise(exception2)144 with self.assertRaises(_exn.UnknownClientError) as catcher:145 self.connection(self.action.object)146 self.assertIs(catcher.exception, exception2)147 def test_give_up_after_third_try(self):148 self.__expect_post().andReturn("i")149 self.action.expect.response_class.andReturn("j")150 exception1 = _exn.ProvisionedThroughputExceededException()151 self.responder.expect("j", "i").andRaise(exception1)152 self.retry_policy.expect.retry(self.action.object, [exception1]).andReturn(0)153 self.__expect_post().andReturn("k")154 self.action.expect.response_class.andReturn("l")155 exception2 = _exn.ProvisionedThroughputExceededException()156 self.responder.expect("l", "k").andRaise(exception2)157 self.retry_policy.expect.retry(self.action.object, [exception1, exception2]).andReturn(0)158 self.__expect_post().andReturn("m")159 self.action.expect.response_class.andReturn("n")160 exception3 = _exn.ProvisionedThroughputExceededException()161 self.responder.expect("n", "m").andRaise(exception3)162 self.retry_policy.expect.retry(self.action.object, [exception1, exception2, exception3]).andReturn(None)163 with self.assertRaises(_exn.ProvisionedThroughputExceededException) as catcher:164 self.connection(self.action.object)165 self.assertIs(catcher.exception, exception3)166 def test_success_after_network_error(self):167 exception = requests.exceptions.RequestException()168 self.__expect_post().andRaise(exception)169 self.retry_policy.expect.retry.withArguments(lambda args, kwds: args[0] is self.action.object and isinstance(args[1][0], _exn.NetworkError)).andReturn(0)170 self.__expect_post().andReturn("k")171 self.action.expect.response_class.andReturn("l")172 exception2 = _exn.ProvisionedThroughputExceededException()173 self.responder.expect("l", "k").andReturn("m")174 self.assertEqual(self.connection(self.action.object), "m")175 def test_failure_on_unkown_exception_raised_by_requests(self):176 exception = Exception()177 self.__expect_post().andRaise(exception)178 with self.assertRaises(_exn.UnknownError) as catcher:179 self.connection(self.action.object)180 self.assertEqual(catcher.exception.args, (exception,))181 def test_success_after_network_error_during_credentials(self):182 exception = _exn.NetworkError()183 self.credentials.expect.get().andRaise(exception)184 self.retry_policy.expect.retry(self.action.object, [exception]).andReturn(0)185 self.__expect_post().andReturn("k")186 self.action.expect.response_class.andReturn("l")187 exception2 = _exn.ProvisionedThroughputExceededException()188 self.responder.expect("l", "k").andReturn("m")189 self.assertEqual(self.connection(self.action.object), "m")190class Signer(object):191 def __init__(self, region, host):192 self.__host = host193 self.__region = region194 def __call__(self, key, secret, now, action, payload):195 # http://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html196 timestamp = now.strftime("%Y%m%dT%H%M%SZ")197 datestamp = now.strftime("%Y%m%d")198 headers = {199 "Content-Type": "application/x-amz-json-1.0",200 "X-Amz-Date": timestamp,201 "X-Amz-Target": "DynamoDB_20120810.{}".format(action),202 "Host": self.__host,203 }204 header_names = ";".join(k.lower() for k in sorted(headers.keys()))205 request = "POST\n/\n\n{}\n{}\n{}".format(206 "".join("{}:{}\n".format(k.lower(), v) for k, v in sorted(headers.iteritems())),207 header_names,208 hashlib.sha256(payload.encode("utf-8")).hexdigest(),209 )210 credentials = "{}/{}/dynamodb/aws4_request".format(datestamp, self.__region)211 to_sign = "AWS4-HMAC-SHA256\n{}\n{}\n{}".format(timestamp, credentials, hashlib.sha256(request.encode("utf-8")).hexdigest())212 hmac_key = hmac.new(213 hmac.new(214 hmac.new(215 hmac.new(216 "AWS4{}".format(secret).encode("utf-8"),217 datestamp.encode("utf-8"),218 hashlib.sha256219 ).digest(),220 self.__region.encode("utf-8"),221 hashlib.sha256222 ).digest(),223 "dynamodb".encode("utf-8"),224 hashlib.sha256225 ).digest(),226 "aws4_request".encode("utf-8"),227 hashlib.sha256228 ).digest()229 headers["Authorization"] = "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}".format(230 key,231 credentials,232 header_names,233 hmac.new(hmac_key, to_sign.encode("utf-8"), hashlib.sha256).hexdigest(),234 )235 return headers236class SignerUnitTests(_tst.UnitTests):237 def test(self):238 signer = Signer("us-west-2", "localhost")239 self.assertEqual(240 signer("DummyKey", "DummySecret", datetime.datetime(2014, 10, 4, 6, 33, 2), "Operation", '{"Payload": "Value"}'),241 {242 "Host": "localhost",243 "Content-Type": "application/x-amz-json-1.0",244 "Authorization": "AWS4-HMAC-SHA256 Credential=DummyKey/20141004/us-west-2/dynamodb/aws4_request, SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=f47b4025d95692c1623d01bd7db6d53e68f7a8a28264c1ab3393477f0dae520a",245 "X-Amz-Date": "20141004T063302Z",246 "X-Amz-Target": "DynamoDB_20120810.Operation",247 }248 )249class Responder(object):250 def __call__(self, response_class, r):251 status_code = r.status_code252 if status_code == 200:253 try:254 data = r.json()255 except ValueError:256 raise _exn.ServerError(200, r.text)257 return response_class(**data)258 else:259 self.__raise(status_code, r)260 def __raise(self, status_code, r):261 try:262 data = r.json()263 except ValueError:264 data = r.text265 if isinstance(data, dict):266 typ = data.get("__type")267 else:268 typ = None269 if 400 <= status_code < 500:270 if typ is not None:271 for suffix, cls in _exn.client_errors:272 if typ.endswith(suffix):273 raise cls(data)274 raise _exn.UnknownClientError(status_code, data)275 elif 500 <= status_code < 600:276 raise _exn.ServerError(status_code, data)277 else:278 raise _exn.UnknownError(status_code, data)279class ResponderUnitTests(_tst.UnitTestsWithMocks):280 def setUp(self):281 super(ResponderUnitTests, self).setUp()282 self.response_class = self.mocks.create("response_class")283 self.response_instance = object()284 self.requests_response = self.mocks.create("requests_response")285 self.json = {"a": 0}286 self.responder = Responder()287 def test_good_response(self):288 self.requests_response.expect.status_code.andReturn(200)289 self.requests_response.expect.json().andReturn(self.json)290 self.response_class.expect(a=0).andReturn(self.response_instance)291 self.assertIs(self.responder(self.response_class.object, self.requests_response.object), self.response_instance)292 def test_non_json_response_with_good_status(self):293 self.requests_response.expect.status_code.andReturn(200)294 self.requests_response.expect.json().andRaise(ValueError)295 self.requests_response.expect.text.andReturn("foobar")296 with self.assertRaises(_exn.ServerError) as catcher:297 self.responder(self.response_class.object, self.requests_response.object)298 self.assertEqual(catcher.exception.args, (200, "foobar"))299 def test_unknown_client_error_with_correct_json(self):300 self.requests_response.expect.status_code.andReturn(400)301 self.requests_response.expect.json().andReturn({"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"})302 with self.assertRaises(_exn.UnknownClientError) as catcher:303 self.responder(self.response_class.object, self.requests_response.object)304 self.assertEqual(catcher.exception.args, (400, {"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"}))305 def test_unknown_client_error_with_json_without_type(self):306 self.requests_response.expect.status_code.andReturn(400)307 self.requests_response.expect.json().andReturn({"Message": "tralala"})308 with self.assertRaises(_exn.UnknownClientError) as catcher:309 self.responder(self.response_class.object, self.requests_response.object)310 self.assertEqual(catcher.exception.args, (400, {"Message": "tralala"}))311 def test_unknown_client_error_with_non_dict_json(self):312 self.requests_response.expect.status_code.andReturn(400)313 self.requests_response.expect.json().andReturn(["Message", "tralala"])314 with self.assertRaises(_exn.UnknownClientError) as catcher:315 self.responder(self.response_class.object, self.requests_response.object)316 self.assertEqual(catcher.exception.args, (400, ["Message", "tralala"]))317 def test_unknown_client_error_without_json(self):318 self.requests_response.expect.status_code.andReturn(400)319 self.requests_response.expect.json().andRaise(ValueError)320 self.requests_response.expect.text.andReturn("Message: tralala")321 with self.assertRaises(_exn.UnknownClientError) as catcher:322 self.responder(self.response_class.object, self.requests_response.object)323 self.assertEqual(catcher.exception.args, (400, "Message: tralala"))324 def test_server_error_with_correct_json(self):325 self.requests_response.expect.status_code.andReturn(500)326 self.requests_response.expect.json().andReturn({"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"})327 with self.assertRaises(_exn.ServerError) as catcher:328 self.responder(self.response_class.object, self.requests_response.object)329 self.assertEqual(catcher.exception.args, (500, {"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"}))330 def test_server_error_with_json_without_type(self):331 self.requests_response.expect.status_code.andReturn(500)332 self.requests_response.expect.json().andReturn({"Message": "tralala"})333 with self.assertRaises(_exn.ServerError) as catcher:334 self.responder(self.response_class.object, self.requests_response.object)335 self.assertEqual(catcher.exception.args, (500, {"Message": "tralala"}))336 def test_server_error_with_non_dict_json(self):337 self.requests_response.expect.status_code.andReturn(500)338 self.requests_response.expect.json().andReturn(["Message", "tralala"])339 with self.assertRaises(_exn.ServerError) as catcher:340 self.responder(self.response_class.object, self.requests_response.object)341 self.assertEqual(catcher.exception.args, (500, ["Message", "tralala"]))342 def test_server_error_without_json(self):343 self.requests_response.expect.status_code.andReturn(500)344 self.requests_response.expect.json().andRaise(ValueError)345 self.requests_response.expect.text.andReturn("Message: tralala")346 with self.assertRaises(_exn.ServerError) as catcher:347 self.responder(self.response_class.object, self.requests_response.object)348 self.assertEqual(catcher.exception.args, (500, "Message: tralala"))349 def test_unknown_error_with_correct_json(self):350 self.requests_response.expect.status_code.andReturn(750)351 self.requests_response.expect.json().andReturn({"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"})352 with self.assertRaises(_exn.UnknownError) as catcher:353 self.responder(self.response_class.object, self.requests_response.object)354 self.assertEqual(catcher.exception.args, (750, {"__type": "NobodyKnewThisCouldHappen", "Message": "tralala"}))355 def test_unknown_error_with_json_without_type(self):356 self.requests_response.expect.status_code.andReturn(750)357 self.requests_response.expect.json().andReturn({"Message": "tralala"})358 with self.assertRaises(_exn.UnknownError) as catcher:359 self.responder(self.response_class.object, self.requests_response.object)360 self.assertEqual(catcher.exception.args, (750, {"Message": "tralala"}))361 def test_unknown_error_with_non_dict_json(self):362 self.requests_response.expect.status_code.andReturn(750)363 self.requests_response.expect.json().andReturn(["Message", "tralala"])364 with self.assertRaises(_exn.UnknownError) as catcher:365 self.responder(self.response_class.object, self.requests_response.object)366 self.assertEqual(catcher.exception.args, (750, ["Message", "tralala"]))367 def test_unknown_error_without_json(self):368 self.requests_response.expect.status_code.andReturn(750)369 self.requests_response.expect.json().andRaise(ValueError)370 self.requests_response.expect.text.andReturn("Message: tralala")371 with self.assertRaises(_exn.UnknownError) as catcher:372 self.responder(self.response_class.object, self.requests_response.object)373 self.assertEqual(catcher.exception.args, (750, "Message: tralala"))374 def test_known_client_errors(self):375 for type_name, cls in [376 ("xxx.AccessDeniedException", _exn.AccessDeniedException),377 ("xxx.ConditionalCheckFailedException", _exn.ConditionalCheckFailedException),378 ("xxx.IncompleteSignature", _exn.IncompleteSignature),379 ("xxx.InvalidAction", _exn.InvalidAction),380 ("xxx.InvalidClientTokenId", _exn.InvalidClientTokenId),381 ("xxx.InvalidParameterCombination", _exn.InvalidParameterCombination),382 ("xxx.InvalidParameterValue", _exn.InvalidParameterValue),383 ("xxx.InvalidQueryParameter", _exn.InvalidQueryParameter),384 ("xxx.InvalidSignatureException", _exn.InvalidSignatureException),385 ("xxx.ItemCollectionSizeLimitExceededException", _exn.ItemCollectionSizeLimitExceededException),386 ("xxx.LimitExceededException", _exn.LimitExceededException),387 ("xxx.MalformedQueryString", _exn.MalformedQueryString),388 ("xxx.MissingAction", _exn.MissingAction),389 ("xxx.MissingAuthenticationToken", _exn.MissingAuthenticationToken),390 ("xxx.MissingParameter", _exn.MissingParameter),391 ("xxx.OptInRequired", _exn.OptInRequired),392 ("xxx.ProvisionedThroughputExceededException", _exn.ProvisionedThroughputExceededException),393 ("xxx.RequestExpired", _exn.RequestExpired),394 ("xxx.ResourceInUseException", _exn.ResourceInUseException),395 ("xxx.ResourceNotFoundException", _exn.ResourceNotFoundException),396 ("xxx.Throttling", _exn.Throttling),397 ("xxx.ValidationError", _exn.ValidationError),398 ("xxx.ValidationException", _exn.ValidationException),399 ]:400 self.requests_response.expect.status_code.andReturn(400)401 self.requests_response.expect.json().andReturn({"__type": type_name, "Message": "tralala"})402 with self.assertRaises(cls) as catcher:403 self.responder(self.response_class.object, self.requests_response.object)404 def test_different_statuses(self):405 self.requests_response.expect.status_code.andReturn(400)406 self.requests_response.expect.json().andReturn({})407 with self.assertRaises(_exn.ClientError):408 self.responder(self.response_class.object, self.requests_response.object)409 self.requests_response.expect.status_code.andReturn(453)410 self.requests_response.expect.json().andReturn({})411 with self.assertRaises(_exn.ClientError):412 self.responder(self.response_class.object, self.requests_response.object)413 self.requests_response.expect.status_code.andReturn(499)414 self.requests_response.expect.json().andReturn({})415 with self.assertRaises(_exn.ClientError):416 self.responder(self.response_class.object, self.requests_response.object)417 self.requests_response.expect.status_code.andReturn(500)418 self.requests_response.expect.json().andReturn({})419 with self.assertRaises(_exn.ServerError):420 self.responder(self.response_class.object, self.requests_response.object)421 self.requests_response.expect.status_code.andReturn(547)422 self.requests_response.expect.json().andReturn({})423 with self.assertRaises(_exn.ServerError):424 self.responder(self.response_class.object, self.requests_response.object)425 self.requests_response.expect.status_code.andReturn(599)426 self.requests_response.expect.json().andReturn({})427 with self.assertRaises(_exn.ServerError):428 self.responder(self.response_class.object, self.requests_response.object)429 self.requests_response.expect.status_code.andReturn(600)430 self.requests_response.expect.json().andReturn({})431 with self.assertRaises(_exn.UnknownError):432 self.responder(self.response_class.object, self.requests_response.object)433 self.requests_response.expect.status_code.andReturn(612)434 self.requests_response.expect.json().andReturn({})435 with self.assertRaises(_exn.UnknownError):436 self.responder(self.response_class.object, self.requests_response.object)437 self.requests_response.expect.status_code.andReturn(9999)438 self.requests_response.expect.json().andReturn({})439 with self.assertRaises(_exn.UnknownError):...
wCurl.py
Source:wCurl.py
...112 '''113 if resp_from_requests is None:114 response = Response(req_url=req_url)115 else:116 response = from_requests_response(resp_from_requests,req_url)117 return response...
WSCurl.py
Source:WSCurl.py
...63 def _make_response(self, resp_from_requests, req_url):64 if resp_from_requests is None:65 response = WSResponse(req_url=req_url)66 else:67 response = from_requests_response(resp_from_requests, req_url)68 return response69if __name__ == '__main__':70 print("curl test")71 curl = wCurl()72 urlStr = "https://www.paypalm.cn"73 res = curl.get(urlStr)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
