How to use _validate_response method in localstack
Best Python code snippet using localstack_python
session.py
Source:session.py 
...7from . import urls8import os9import logging10logger = logging.getLogger("Plugin.Tesla") #Can call it whatever I like11def _validate_response(response):12 """ Verify that response is OK """13 if response.status_code == 200:14 return15 raise ResponseError(response.status_code, response.text)16class Error(Exception):17 ''' Verisure session error '''18 pass19class RequestError(Error):20 ''' Wrapped requests.exceptions.RequestException '''21 pass22class LoginError(Error):23 ''' Login failed '''24 pass25class ResponseError(Error):26 ''' Unexpected response '''27 def __init__(self, status_code, text):28 super(ResponseError, self).__init__(29 'Invalid response'30 ', status code: {0} - Data: {1}'.format(31 status_code,32 text))33 self.status_code = status_code34 self.text = text35class Session(object):36 """ Verisure app session37 Args:38 username (str): Username used to login to verisure app39 password (str): Password used to login to verisure app40 """41 def __init__(self, username, password,cookieFileName='.verisure-cookie'):42 self._username = username43 self._password = password44 self._cookieFileName = cookieFileName45 self._vid = None46 self._giid = None47 self.installations = None48 #logger.debug(self._cookieFileName)49 def __enter__(self):50 self.login()51 return self52 def __exit__(self, exc_type, exc_val, exc_tb):53 self.logout()54 """ If of interest, add exception handler55 """56 def login(self):57 """ Login to verisure app api58 Login before calling any read or write commands59 """60 if os.path.exists(self._cookieFileName):61 #logger.debug("Cookie exists")62 with open(self._cookieFileName, 'r') as cookieFile:63 self._vid = cookieFile.read().strip()64 try:65 #logger.debug("Getting installations")66 self._get_installations()67 except ResponseError:68 self._vid = None69 os.remove(self._cookieFileName)70 if self._vid is None:71 try:72 self._create_cookie()73 with open(self._cookieFileName, 'w') as cookieFile:74 cookieFile.write(self._vid)75 self._get_installations()76 except ResponseError: #Failed to login77 self._vid = None78 os.remove(self._cookieFileName)79 return80 except TypeError: #Failed to write VID as login failed81 self._vid = None82 os.remove(self._cookieFileName)83 return84 self._giid = self.installations[0]['giid']85 def _create_cookie(self):86 auth = 'Basic {}'.format(87 base64.b64encode(88 'CPE/{username}:{password}'.format(89 username=self._username,90 password=self._password).encode('utf-8')91 ).decode('utf-8'))92 response = None93 for base_url in urls.BASE_URLS:94 urls.BASE_URL = base_url95 try:96 response = requests.post(97 urls.login(),98 headers={99 'Authorization': auth,100 'Accept': 'application/json,'101 'text/javascript, */*; q=0.01',102 })103 if response.status_code == 200:104 break105 elif response.status_code == 503:106 continue107 elif response.status_code == 401:108 logger.error("Invalid username or password")109 return False110 else:111 raise ResponseError(response.status_code, response.text)112 except requests.exceptions.RequestException as ex:113 raise LoginError(ex)114 _validate_response(response)115 self._vid = json.loads(response.text)['cookie']116 def _get_installations(self):117 """ Get information about installations """118 response = None119 for base_url in urls.BASE_URLS:120 urls.BASE_URL = base_url121 try:122 response = requests.get(123 urls.get_installations(self._username),124 headers={125 'Cookie': 'vid={}'.format(self._vid),126 'Accept': 'application/json,'127 'text/javascript, */*; q=0.01',128 })129 if 2 == response.status_code // 100:130 break131 elif 503 == response.status_code:132 continue133 else:134 raise ResponseError(response.status_code, response.text)135 except requests.exceptions.RequestException as ex:136 raise RequestError(ex)137 _validate_response(response)138 self.installations = json.loads(response.text)139 def set_giid(self, giid):140 """ Set installation giid141 Args:142 giid (str): Installation identifier143 """144 self._giid = giid145 def get_overview(self):146 """ Get overview for installation """147 response = None148 try:149 response = requests.get(150 urls.overview(self._giid),151 headers={152 'Accept': 'application/json, text/javascript, */*; q=0.01',153 'Accept-Encoding': 'gzip, deflate',154 'Content-Type': 'application/json',155 'Cookie': 'vid={}'.format(self._vid)})156 except requests.exceptions.RequestException as ex:157 raise RequestError(ex)158 _validate_response(response)159 return json.loads(response.text)160 def set_smartplug_state(self, device_label, state):161 """ Turn on or off smartplug162 Args:163 device_label (str): Smartplug device label164 state (boolean): new status, 'True' or 'False'165 """166 response = None167 try:168 response = requests.post(169 urls.smartplug(self._giid),170 headers={171 'Content-Type': 'application/json',172 'Cookie': 'vid={}'.format(self._vid)},173 data=json.dumps([{174 "deviceLabel": device_label,175 "state": state}]))176 except requests.exceptions.RequestException as ex:177 raise RequestError(ex)178 _validate_response(response)179 def set_arm_state(self, code, state):180 """ Set alarm state181 Args:182 code (str): Personal alarm code (four or six digits)183 state (str): 'ARMED_HOME', 'ARMED_AWAY' or 'DISARMED'184 """185 response = None186 try:187 response = requests.put(188 urls.set_armstate(self._giid),189 headers={190 'Accept': 'application/json, text/javascript, */*; q=0.01',191 'Content-Type': 'application/json',192 'Cookie': 'vid={}'.format(self._vid)},193 data=json.dumps({"code": str(code), "state": state}))194 except requests.exceptions.RequestException as ex:195 raise RequestError(ex)196 _validate_response(response)197 return json.loads(response.text)198 def get_arm_state_transaction(self, transaction_id):199 """ Get arm state transaction status200 Args:201 transaction_id: Transaction ID received from set_arm_state202 """203 response = None204 try:205 response = requests.get(206 urls.get_armstate_transaction(self._giid, transaction_id),207 headers={208 'Accept': 'application/json, text/javascript, */*; q=0.01',209 'Cookie': 'vid={}'.format(self._vid)})210 except requests.exceptions.RequestException as ex:211 raise RequestError(ex)212 _validate_response(response)213 return json.loads(response.text)214 def get_arm_state(self):215 """ Get arm state """216 response = None217 try:218 response = requests.get(219 urls.get_armstate(self._giid),220 headers={221 'Accept': 'application/json, text/javascript, */*; q=0.01',222 'Cookie': 'vid={}'.format(self._vid)})223 except requests.exceptions.RequestException as ex:224 raise RequestError(ex)225 _validate_response(response)226 return json.loads(response.text)227 def get_history(self, filters=(), pagesize=15, offset=0):228 """ Get recent events229 Args:230 filters (string set): 'ARM', 'DISARM', 'FIRE', 'INTRUSION',231 'TECHNICAL', 'SOS', 'WARNING', 'LOCK',232 'UNLOCK', 'PICTURE', 'CLIMATE'233 pagesize (int): Number of events to display234 offset (int): Skip pagesize * offset first events235 """236 response = None237 try:238 response = requests.get(239 urls.history(self._giid),240 headers={241 'Accept': 'application/json, text/javascript, */*; q=0.01',242 'Cookie': 'vid={}'.format(self._vid)},243 params={244 "offset": int(offset),245 "pagesize": int(pagesize),246 "eventCategories": filters})247 except requests.exceptions.RequestException as ex:248 raise RequestError(ex)249 _validate_response(response)250 return json.loads(response.text)251 def get_climate(self, device_label):252 """ Get climate history253 Args:254 device_label: device label of climate device255 """256 response = None257 try:258 response = requests.get(259 urls.climate(self._giid),260 headers={261 'Accept': 'application/json, text/javascript, */*; q=0.01',262 'Cookie': 'vid={}'.format(self._vid)},263 params={264 "deviceLabel": device_label})265 except requests.exceptions.RequestException as ex:266 raise RequestError(ex)267 _validate_response(response)268 return json.loads(response.text)269 def get_lock_state(self):270 """ Get current lock status """271 response = None272 try:273 response = requests.get(274 urls.get_lockstate(self._giid),275 headers={276 'Accept': 'application/json, text/javascript, */*; q=0.01',277 'Cookie': 'vid={}'.format(self._vid)})278 except requests.exceptions.RequestException as ex:279 raise RequestError(ex)280 _validate_response(response)281 return json.loads(response.text)282 def set_lock_state(self, code, device_label, state):283 """ Lock or unlock284 Args:285 code (str): Lock code286 device_label (str): device label of lock287 state (str): 'lock' or 'unlock'288 """289 response = None290 try:291 response = requests.put(292 urls.set_lockstate(self._giid, device_label, state),293 headers={294 'Accept': 'application/json, text/javascript, */*; q=0.01',295 'Content-Type': 'application/json',296 'Cookie': 'vid={}'.format(self._vid)},297 data=json.dumps({"code": str(code)}))298 except requests.exceptions.RequestException as ex:299 raise RequestError(ex)300 _validate_response(response)301 return json.loads(response.text)302 def get_lock_state_transaction(self, transaction_id):303 """ Get lock state transaction status304 Args:305 transaction_id: Transaction ID received from set_lock_state306 """307 response = None308 try:309 response = requests.get(310 urls.get_lockstate_transaction(self._giid, transaction_id),311 headers={312 'Accept': 'application/json, text/javascript, */*; q=0.01',313 'Cookie': 'vid={}'.format(self._vid)})314 except requests.exceptions.RequestException as ex:315 raise RequestError(ex)316 _validate_response(response)317 return json.loads(response.text)318 def get_lock_config(self, device_label):319 """ Get lock configuration320 Args:321 device_label (str): device label of lock322 """323 response = None324 try:325 response = requests.get(326 urls.lockconfig(self._giid, device_label),327 headers={328 'Accept': 'application/json, text/javascript, */*; q=0.01',329 'Cookie': 'vid={}'.format(self._vid)})330 except requests.exceptions.RequestException as ex:331 raise RequestError(ex)332 _validate_response(response)333 return json.loads(response.text)334 def set_lock_config(self, device_label, volume=None, voice_level=None,335 auto_lock_enabled=None):336 """ Set lock configuration337 Args:338 device_label (str): device label of lock339 volume (str): 'SILENCE', 'LOW' or 'HIGH'340 voice_level (str): 'ESSENTIAL' or 'NORMAL'341 auto_lock_enabled (boolean): auto lock enabled342 """343 response = None344 data = {}345 if volume:346 data['volume'] = volume347 if voice_level:348 data['voiceLevel'] = voice_level349 if auto_lock_enabled is not None:350 data['autoLockEnabled'] = auto_lock_enabled351 try:352 response = requests.put(353 urls.lockconfig(self._giid, device_label),354 headers={355 'Content-Type': 'application/json',356 'Cookie': 'vid={}'.format(self._vid)},357 data=json.dumps(data))358 except requests.exceptions.RequestException as ex:359 raise RequestError(ex)360 _validate_response(response)361 def capture_image(self, device_label):362 """ Capture smartcam image363 Args:364 device_label (str): device label of camera365 """366 response = None367 try:368 response = requests.post(369 urls.imagecapture(self._giid, device_label),370 headers={371 'Content-Type': 'application/json',372 'Cookie': 'vid={}'.format(self._vid)})373 except requests.exceptions.RequestException as ex:374 raise RequestError(ex)375 _validate_response(response)376 def get_camera_imageseries(self, number_of_imageseries=10, offset=0):377 """ Get smartcam image series378 Args:379 number_of_imageseries (int): number of image series to get380 offset (int): skip offset amount of image series381 """382 response = None383 try:384 response = requests.get(385 urls.get_imageseries(self._giid),386 headers={387 'Accept': 'application/json, text/javascript, */*; q=0.01',388 'Cookie': 'vid={}'.format(self._vid)},389 params={390 "numberOfImageSeries": int(number_of_imageseries),391 "offset": int(offset),392 "fromDate": "",393 "toDate": "",394 "onlyNotViewed": "",395 "_": self._giid})396 except requests.exceptions.RequestException as ex:397 raise RequestError(ex)398 _validate_response(response)399 return json.loads(response.text)400 def download_image(self, device_label, image_id, file_name):401 """ Download image taken by a smartcam402 Args:403 device_label (str): device label of camera404 image_id (str): image id from image series405 file_name (str): path to file406 """407 response = None408 try:409 response = requests.get(410 urls.download_image(self._giid, device_label, image_id),411 headers={412 'Cookie': 'vid={}'.format(self._vid)},413 stream=True)414 except requests.exceptions.RequestException as ex:415 raise RequestError(ex)416 _validate_response(response)417 with open(file_name, 'wb') as image_file:418 for chunk in response.iter_content(chunk_size=1024):419 if chunk:420 image_file.write(chunk)421 def get_vacation_mode(self):422 """ Get current vacation mode """423 response = None424 try:425 response = requests.get(426 urls.get_vacationmode(self._giid),427 headers={428 'Accept': 'application/json, text/javascript, */*; q=0.01',429 'Cookie': 'vid={}'.format(self._vid)})430 except requests.exceptions.RequestException as ex:431 raise RequestError(ex)432 _validate_response(response)433 return json.loads(response.text)434 def get_door_window(self):435 """ Get door_window states"""436 response = None437 try:438 response = requests.get(439 urls.door_window(self._giid),440 headers={441 'Accept': 'application/json, text/javascript, */*; q=0.01',442 'Cookie': 'vid={}'.format(self._vid)})443 except requests.exceptions.RequestException as ex:444 raise RequestError(ex)445 _validate_response(response)446 return json.loads(response.text)447 def test_ethernet(self):448 """ Test ethernet status """449 response = None450 try:451 response = requests.post(452 urls.test_ethernet(self._giid),453 headers={454 'Content-Type': 'application/json',455 'Cookie': 'vid={}'.format(self._vid)})456 except requests.exceptions.RequestException as ex:457 raise RequestError(ex)458 _validate_response(response)459 def logout(self):460 """ Logout and remove vid """461 response = None462 try:463 response = requests.delete(464 urls.login(),465 headers={466 'Cookie': 'vid={}'.format(self._vid)})467 except requests.exceptions.RequestException as ex:468 raise RequestError(ex)469 _validate_response(response)470 def get_heat_pump_state(self, device_label):471 """ Get heatpump states"""472 response = None473 try:474 response = requests.get(475 urls.get_heatpump_state(self._giid, device_label),476 headers={477 'Accept': 'application/json, text/javascript, */*; q=0.01',478 'Cookie': 'vid={}'.format(self._vid)})479 except requests.exceptions.RequestException as ex:480 raise RequestError(ex)481 _validate_response(response)482 return json.loads(response.text)483 def set_heat_pump_mode(self, device_label, mode):484 """ Set heatpump mode485 Args:486 mode (str): 'HEAT', 'COOL', 'FAN' or 'AUTO'487 """488 response = None489 try:490 response = requests.put(491 urls.set_heatpump_state(self._giid, device_label),492 headers={493 'Accept': 'application/json',494 'Content-Type': 'application/json',495 'Cookie': 'vid={}'.format(self._vid)},496 data=json.dumps({'mode': mode}))497 except requests.exceptions.RequestException as ex:498 raise RequestError(ex)499 _validate_response(response)500 return json.loads(response.text)501 def set_heat_pump_power(self, device_label, power):502 """ Set heatpump mode503 Args:504 power (str): 'ON', 'OFF'505 """506 response = None507 try:508 response = requests.put(509 urls.set_heatpump_state(self._giid, device_label),510 headers={511 'Accept': 'application/json',512 'Content-Type': 'application/json',513 'Cookie': 'vid={}'.format(self._vid)},514 data=json.dumps({'power': power}))515 except requests.exceptions.RequestException as ex:516 raise RequestError(ex)517 _validate_response(response)518 return json.loads(response.text)519 def set_heat_pump_fan_speed(self, device_label, fan_speed):520 """ Set heatpump mode521 Args:522 fan_speed (str): 'LOW', 'MEDIUM_LOW', 'MEDIUM, 'MEDIUM_HIGH' or 'HIGH'523 """524 response = None525 try:526 response = requests.put(527 urls.set_heatpump_state(self._giid, device_label),528 headers={529 'Accept': 'application/json',530 'Content-Type': 'application/json',531 'Cookie': 'vid={}'.format(self._vid)},532 data=json.dumps({'fanSpeed': fan_speed}))533 except requests.exceptions.RequestException as ex:534 raise RequestError(ex)535 _validate_response(response)536 return json.loads(response.text)537 def set_heat_pump_target_temperature(self, device_label, target_temp):538 """ Set heatpump mode539 Args:540 target_temperature (int): required temperature of the heatpump541 """542 response = None543 try:544 response = requests.put(545 urls.set_heatpump_state(self._giid, device_label),546 headers={547 'Accept': 'application/json',548 'Content-Type': 'application/json',549 'Cookie': 'vid={}'.format(self._vid)},550 data=json.dumps({'targetTemperature': target_temp}))551 except requests.exceptions.RequestException as ex:552 raise RequestError(ex)553 _validate_response(response)554 return json.loads(response.text)555 def set_heat_pump_feature(self, device_label, feature):556 """ Set heatpump mode557 Args:558 feature: 'QUIET', 'ECONAVI', or 'POWERFUL'559 """560 response = None561 try:562 response = requests.put(563 urls.set_heatpump_feature(self._giid, device_label, feature),564 headers={565 'Accept': 'application/json',566 'Content-Type': 'application/json',567 'Cookie': 'vid={}'.format(self._vid)})568 except requests.exceptions.RequestException as ex:569 raise RequestError(ex)570 _validate_response(response)571 return json.loads(response.text)572 def set_heat_pump_airswingdirection(self, device_label, airswingdirection):573 """ Set heatpump mode574 Args:575 airSwingDirection (str): 'AUTO' '0_DEGREES' '30_DEGREES'576 '60_DEGREES' '90_DEGREES'577 """578 response = None579 try:580 response = requests.put(581 urls.set_heatpump_state(self._giid, device_label),582 headers={583 'Accept': 'application/json',584 'Content-Type': 'application/json',585 'Cookie': 'vid={}'.format(self._vid)},586 data=json.dumps({'airSwingDirection':587 {"vertical": airswingdirection}}))588 except requests.exceptions.RequestException as ex:589 raise RequestError(ex)590 _validate_response(response)...
test_captcha.py
Source:test_captcha.py
...18 def test_post_with_proper_data(self, mk):19 mk.post.side_effect = StopIteration20 exp1, exp2 = self.validate_vars21 try:22 self.model_obj._validate_response(exp1, exp2, website=self.website)23 except StopIteration:24 pass25 mk.post.assert_called_once_with(26 self.model_obj.URL,27 data={28 "secret": self.website.recaptcha_key_secret,29 "response": exp1,30 "remoteip": exp2,31 },32 )33 @mock.patch(imp_requests)34 def test_valid(self, mk):35 expect = {"success": True}36 mk.post().json.return_value = expect37 self.assertTrue(38 self.model_obj._validate_response(*self.validate_vars, website=self.website)39 )40 @mock.patch(imp_requests)41 def test_known_error_raises(self, mk):42 expect = {"error-codes": ["missing-input-secret"]}43 mk.post().json.return_value = expect44 with self.assertRaises(ValidationError):45 self.model_obj._validate_response(*self.validate_vars, website=self.website)46 @mock.patch(imp_requests)47 def test_known_error_lookup(self, mk):48 expect = {"error-codes": ["missing-input-secret"]}49 mk.post().json.return_value = expect50 try:51 self.model_obj._validate_response(*self.validate_vars, website=self.website)52 except ValidationError as e:53 self.assertEqual(54 e.name, self.model_obj._get_error_message(expect["error-codes"][0])55 )56 @mock.patch(imp_requests)57 def test_unknown_error_lookup(self, mk):58 expect = {"error-codes": ["derp"]}59 mk.post().json.return_value = expect60 try:61 self.model_obj._validate_response(*self.validate_vars, website=self.website)62 except ValidationError as e:63 self.assertEqual(e.name, self.model_obj._get_error_message())64 @mock.patch(imp_requests)65 def test_no_success_raises(self, mk):66 expect = {"success": False}67 mk.post().json.return_value = expect68 with self.assertRaises(ValidationError):69 self.model_obj._validate_response(*self.validate_vars, website=self.website)70 def test_validate_request_no_value(self):71 request = object()72 req_values = {}73 with self.assertRaises(ValidationError) as err:74 self.model_obj._validate_request(request, req_values)75 self.assertEqual(err.exception.name, "The secret parameter is missing.")76 def test_validate_request_old_value(self):77 request = mock.MagicMock()78 request.g_recaptcha_response = "all good here"79 req_values = {}80 with mock.patch.object(type(self.model_obj), "_validate_response") as mocked:81 self.assertTrue(self.model_obj._validate_request(request, req_values))82 mocked.assert_not_called()83 def test_validate_request_validate_response(self):84 # Ensure that w/ proper conditions `validate_response is called`85 request = mock.MagicMock()86 request.g_recaptcha_response = None87 request.httprequest.environ = {}88 request.httprequest.remote_addr = "1.2.3.4"89 with mock.patch.object(type(self.model_obj), "_validate_response") as mocked:90 self.model_obj._validate_request(91 request, {self.model_obj.RESPONSE_ATTR: "validate_me"}92 )...
slack.py
Source:slack.py
...16 filename=filename,17 initial_comment=comment,18 file=file19 )20 self._validate_response(res)21 def chat_post_message(self, text, channel=None):22 """23 ã¡ãã»ã¼ã¸ãæ稿ãã24 :param text:25 :param channel:26 """27 res = self.slack.api_call(28 'chat.postMessage',29 channel=channel or self.channel,30 text=text31 )32 self._validate_response(res)33 @staticmethod34 def _validate_response(res):35 if not res['ok']:...
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?
