How to use cors_config method in localstack
Best Python code snippet using localstack_python
app.py
Source:app.py 
1import json2import logging3from botocore.vendored import requests4from datetime import date, datetime5from chalice import Chalice, CORSConfig6#####7# VARS8#####9cors_config = CORSConfig(10 allow_origin='*',11 allow_headers=['Accept'],12 max_age=60013 )14apiEndpoint = 'https://owner-api.teslamotors.com/api/1/'15res = {}16#####17# HELPERS18#####19def json_serial(obj):20 logger.debug('in json_serial()...')21 if isinstance(obj, (datetime, date)):22 return obj.isoformat()23 raise TypeError("Type %s not serializable" % type(obj))24def return_json(statusCode, body, headers={}):25 logger.debug('in return_json()......')26 headers['Content-Type'] = 'application/json'27 headers['Access-Control-Allow-Headers'] = 'Content-Type,X-Amz-Date,Authorization,X-Api-Key'28 headers['Access-Control-Allow-Methods'] = 'POST, GET, PUT, DELETE'29 headers['Access-Control-Allow-Origin'] = '*'30 return {31 'statusCode': statusCode,32 'body': json.loads(json.dumps(body, default=json_serial)),33 'headers': headers}34#####35# LOGGING36#####37logger = logging.getLogger()38logger.setLevel(logging.DEBUG)39#####40# CHALICE41#####42app = Chalice(app_name="root")43#####44# ROUTES45#####46@app.route('/command/{vehicleId}/sentry_mode_off', methods=['POST'], cors=cors_config)47def vehicles_command_sentry_mode_off(vehicleId):48 global res49 logger.info('in /command/{vehicleId}/sentry_mode_off POST')50 try:51 res = command(vehicleId, app.current_request.json_body, 'set_sentry_mode', {'on': False})52 return return_json(200, res)53 except Exception as e:54 logger.error(str(e))55 res['error'] = f'Failed to turn off sentry mode.'56 return return_json(500, res)57@app.route('/command/{vehicleId}/sentry_mode_on', methods=['POST'], cors=cors_config)58def vehicles_command_sentry_mode_on(vehicleId):59 global res60 logger.info('in /command/{vehicleId}/sentry_mode_on POST')61 try:62 res = command(vehicleId, app.current_request.json_body, 'set_sentry_mode', {'on': 'on'})63 return return_json(200, res)64 except Exception as e:65 logger.error(str(e))66 res['error'] = f'Failed to turn on sentry mode.'67 return return_json(500, res)68@app.route('/command/{vehicleId}/remote_steering_wheel_heater_off', methods=['POST'], cors=cors_config)69def vehicles_command_remote_steering_wheel_heater_off(vehicleId):70 global res71 logger.info('in /command/{vehicleId}/remote_steering_wheel_heater_off POST')72 try:73 res = command(vehicleId, app.current_request.json_body, 'remote_steering_wheel_heater_request', {'on': False})74 return return_json(200, res)75 except Exception as e:76 logger.error(str(e))77 res['error'] = f'Failed to turn off steering wheel heater.'78 return return_json(500, res)79@app.route('/command/{vehicleId}/remote_steering_wheel_heater_on', methods=['POST'], cors=cors_config)80def vehicles_command_remote_steering_wheel_heater_on(vehicleId):81 global res82 logger.info('in /command/{vehicleId}/remote_steering_wheel_heater_on POST')83 try:84 res = command(vehicleId, app.current_request.json_body, 'remote_steering_wheel_heater_request', {'on': True})85 return return_json(200, res)86 except Exception as e:87 logger.error(str(e))88 res['error'] = f'Failed to turn on steering wheel heater.'89 return return_json(500, res)90@app.route('/command/{vehicleId}/remote_seat_heater_request/{seats}/{level}', methods=['POST'], cors=cors_config)91def vehicles_seat_heater_request(vehicleId, seats, level):92 global res93 seatsDict = {94 'driver': 0,95 'passenger': 1,96 'left': 2,97 'center': 4,98 'right': 5}99 levelDict = {100 'off': 0,101 'low': 1,102 'medium': 2,103 'high': 3}104 logger.info('in /command/{vehicleId}/remote_seat_heater_request/{seats}/{level} POST')105 try:106 res = command(vehicleId, app.current_request.json_body, 'remote_seat_heater_request', {'heater': seatsDict[seats], 'level': levelDict[level]})107 return return_json(200, res)108 except Exception as e:109 logger.error(str(e))110 res['error'] = f'Failed to set seat heat temperature.'111 return return_json(500, res)112@app.route('/command/{vehicleId}/set_temps/passenger/{temp}', methods=['POST'], cors=cors_config)113def vehicles_command_set_temp_passenger(vehicleId, temp):114 global res115 logger.info('in /command/{vehicleId}/set_temps/passenger/{temp} POST')116 try:117 res = command(vehicleId, app.current_request.json_body, 'set_temps', {'driver_temp': 22.2, 'passenger_temp': temp})118 return return_json(200, res)119 except Exception as e:120 logger.error(str(e))121 res['error'] = f'Failed to set passenger temprature for HVAC.'122 return return_json(500, res)123@app.route('/command/{vehicleId}/set_temps/driver/{temp}', methods=['POST'], cors=cors_config)124def vehicles_command_set_temp_driver(vehicleId, temp):125 global res126 logger.info('in /command/{vehicleId}/set_temps/driver/{temp} POST')127 try:128 res = command(vehicleId, app.current_request.json_body, 'set_temps', {'driver_temp': temp, 'passenger_temp': 22.2})129 return return_json(200, res)130 except Exception as e:131 logger.error(str(e))132 res['error'] = f'Failed to set driver temprature for HVAC.'133 return return_json(500, res)134@app.route('/command/{vehicleId}/set_temps/{temp}', methods=['POST'], cors=cors_config)135def vehicles_command_set_temp(vehicleId, temp):136 global res137 logger.info('in /command/{vehicleId}/set_temps/{temp} POST')138 try:139 res = command(vehicleId, app.current_request.json_body, 'set_temps', {'driver_temp': temp, 'passenger_temp': temp})140 return return_json(200, res)141 except Exception as e:142 logger.error(str(e))143 res['error'] = f'Failed to set temprature for HVAC.'144 return return_json(500, res)145@app.route('/command/{vehicleId}/auto_conditioning_start', methods=['POST'], cors=cors_config)146def vehicles_command_auto_conditioning_start(vehicleId):147 global res148 logger.info('in /command/{vehicleId}/auto_conditioning_start POST')149 try:150 res = command(vehicleId, app.current_request.json_body, 'auto_conditioning_start')151 return return_json(200, res)152 except Exception as e:153 logger.error(str(e))154 res['error'] = f'Failed to turn on HVAC.'155 return return_json(500, res)156@app.route('/command/{vehicleId}/auto_conditioning_stop', methods=['POST'], cors=cors_config)157def vehicles_command_auto_conditioning_stop(vehicleId):158 global res159 logger.info('in /command/{vehicleId}/auto_conditioning_stop POST')160 try:161 res = command(vehicleId, app.current_request.json_body, 'auto_conditioning_stop')162 return return_json(200, res)163 except Exception as e:164 logger.error(str(e))165 res['error'] = f'Failed to turn off HVAC.'166 return return_json(500, res)167@app.route('/command/{vehicleId}/set_charge_limit/{chargeLimit}', methods=['POST'], cors=cors_config)168def vehicles_command_set_charge_limit(vehicleId, chargeLimit):169 global res170 logger.info('in /command/{vehicleId}/set_charge_limit/{chargeLimit} POST')171 try:172 res = command(vehicleId, app.current_request.json_body, 'set_charge_limit', {'percent': chargeLimit})173 return return_json(200, res)174 except Exception as e:175 logger.error(str(e))176 res['error'] = f'Failed to set charge limit to {chargeLimit}.'177 return return_json(500, res)178@app.route('/command/{vehicleId}/charge_max_range', methods=['POST'], cors=cors_config)179def vehicles_command_charge_max_range(vehicleId):180 global res181 logger.info('in /command/{vehicleId}/charge_max_range POST')182 try:183 res = command(vehicleId, app.current_request.json_body, 'charge_max_range')184 return return_json(200, res)185 except Exception as e:186 logger.error(str(e))187 res['error'] = 'Failed to set charge standard to max.'188 return return_json(500, res)189@app.route('/command/{vehicleId}/charge_standard', methods=['POST'], cors=cors_config)190def vehicles_command_charge_standard(vehicleId):191 global res192 logger.info('in /command/{vehicleId}/charge_standard POST')193 try:194 res = command(vehicleId, app.current_request.json_body, 'charge_standard')195 return return_json(200, res)196 except Exception as e:197 logger.error(str(e))198 res['error'] = 'Failed to set charge standard.'199 return return_json(500, res)200@app.route('/command/{vehicleId}/charge_stop', methods=['POST'], cors=cors_config)201def vehicles_command_charge_stop(vehicleId):202 global res203 logger.info('in /command/{vehicleId}/charge_stop POST')204 try:205 res = command(vehicleId, app.current_request.json_body, 'charge_stop')206 return return_json(200, res)207 except Exception as e:208 logger.error(str(e))209 res['error'] = 'Failed to stop charging.'210 return return_json(500, res)211@app.route('/command/{vehicleId}/charge_start', methods=['POST'], cors=cors_config)212def vehicles_command_charge_start(vehicleId):213 global res214 logger.info('in /command/{vehicleId}/charge_start POST')215 try:216 res = command(vehicleId, app.current_request.json_body, 'charge_start')217 return return_json(200, res)218 except Exception as e:219 logger.error(str(e))220 res['error'] = 'Failed to start charging.'221 return return_json(500, res)222@app.route('/command/{vehicleId}/charge_port_door_open', methods=['POST'], cors=cors_config)223def vehicles_command_charge_port_door_open(vehicleId):224 global res225 logger.info('in /command/{vehicleId}/charge_port_door_open POST')226 try:227 res = command(vehicleId, app.current_request.json_body, 'charge_port_door_open')228 return return_json(200, res)229 except Exception as e:230 logger.error(str(e))231 res['error'] = 'Failed to open charge port door.'232 return return_json(500, res)233@app.route('/command/{vehicleId}/charge_port_door_close', methods=['POST'], cors=cors_config)234def vehicles_command_charge_port_door_close(vehicleId):235 global res236 logger.info('in /command/{vehicleId}/charge_port_door_close POST')237 try:238 res = command(vehicleId, app.current_request.json_body, 'charge_port_door_close')239 return return_json(200, res)240 except Exception as e:241 logger.error(str(e))242 res['error'] = 'Failed to close charge port door.'243 return return_json(500, res)244@app.route('/command/{vehicleId}/open_frunk', methods=['POST'], cors=cors_config)245def vehicles_command_open_frunk(vehicleId):246 global res247 logger.info('in /command/{vehicleId}/open_frunk POST')248 try:249 res = command(vehicleId, app.current_request.json_body, 'actuate_trunk', {"which_trunk": 'front'})250 return return_json(200, res)251 except Exception as e:252 logger.error(str(e))253 res['error'] = 'Failed to open frunk.'254 return return_json(500, res)255@app.route('/command/{vehicleId}/open_trunk', methods=['POST'], cors=cors_config)256def vehicles_command_open_trunk(vehicleId):257 global res258 logger.info('in /command/{vehicleId}/open_trunk POST')259 try:260 res = command(vehicleId, app.current_request.json_body, 'actuate_trunk', {"which_trunk": 'rear'})261 return return_json(200, res)262 except Exception as e:263 logger.error(str(e))264 res['error'] = 'Failed to open trunk.'265 return return_json(500, res)266@app.route('/command/{vehicleId}/honk_horn', methods=['POST'], cors=cors_config)267def vehicles_command_honk_horn(vehicleId):268 global res269 logger.info('in /command/{vehicleId}/honk_horn POST')270 try:271 res = command(vehicleId, app.current_request.json_body, 'honk_horn')272 return return_json(200, res)273 except Exception as e:274 logger.error(str(e))275 res['error'] = 'Failed to honk horn.'276 return return_json(500, res)277@app.route('/command/{vehicleId}/flash_lights', methods=['POST'], cors=cors_config)278def vehicles_command_flash_lights(vehicleId):279 global res280 logger.info('in /command/{vehicleId}/flash_lights POST')281 try:282 res = command(vehicleId, app.current_request.json_body, 'flash_lights')283 return return_json(200, res)284 except Exception as e:285 logger.error(str(e))286 res['error'] = 'Failed to flash lights.'287 return return_json(500, res)288@app.route('/command/{vehicleId}/door_lock', methods=['POST'], cors=cors_config)289def vehicles_command_door_lock(vehicleId):290 global res291 logger.info('in /command/{vehicleId}/door_lock POST')292 try:293 res = command(vehicleId, app.current_request.json_body, 'door_lock')294 return return_json(200, res)295 except Exception as e:296 logger.error(str(e))297 res['error'] = 'Failed to lock door.'298 return return_json(500, res)299@app.route('/command/{vehicleId}/door_unlock', methods=['POST'], cors=cors_config)300def vehicles_command_door_unlock(vehicleId):301 global res302 logger.info('in /command/{vehicleId}/door_unlock POST')303 try:304 res = command(vehicleId, app.current_request.json_body, 'door_unlock')305 return return_json(200, res)306 except Exception as e:307 logger.error(str(e))308 res['error'] = 'Failed to unlock door.'309 return return_json(500, res)310@app.route('/command/{vehicleId}/wake_up', methods=['POST'], cors=cors_config)311def vehicles_command_wake_up(vehicleId):312 global res313 logger.info('in /command/{vehicleId}/wake_up POST')314 try:315 res = wake(vehicleId, app.current_request.json_body)316 return return_json(200, res)317 except Exception as e:318 logger.error(str(e))319 res['error'] = 'Failed to wake up car.'320 return return_json(500, res)321@app.route('/vehicles', methods=['POST'], cors=cors_config)322def vehicles_list():323 global res324 logger.info('in /vehicles POST')325 try:326 res = postVehicles(app.current_request.json_body)327 return return_json(200, res)328 except Exception as e:329 logger.error(str(e))330 res['error'] = 'Failed to list vehicles'331 return return_json(500, res)332def wake(vehicleId, Params):333 bearerToken = Params['bearer_token']334 ret = requests.post(335 f'{apiEndpoint}vehicles/{vehicleId}/wake_up',336 headers={337 'Authorization': f'bearer {bearerToken}'})338 if ret.status_code == 200:339 return f'wake_up on {vehicleId} successful.'340 else:341 raise Exception(f'vehicle_id:{vehicleId}, problem with command:wake_up')342def command(vehicleId, Params, command, data={}):343 bearerToken = Params['bearer_token']344 ret = requests.post(345 f'{apiEndpoint}vehicles/{vehicleId}/command/{command}',346 data=data,347 headers={348 'Authorization': f'bearer {bearerToken}'})349 if ret.status_code == 200:350 return f'{command} on {vehicleId} successful.'351 else:352 raise Exception(f'vehicle_id:{vehicleId} problem with command:{command}')353def postVehicles(Params):354 bearerToken = Params['bearer_token']355 response = requests.get(356 f'{apiEndpoint}vehicles',357 headers={358 'Authorization': f'bearer {bearerToken}'})359 vehicles_json = response.json()360 vehicles = []361 for vehicle in vehicles_json['response']:362 vehicles.append(str(vehicle['id']))...
test_cors.py
Source:test_cors.py
1# Copyright 2021 Ben Kehoe2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# https://www.apache.org/licenses/LICENSE-2.08#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import pytest14import secrets15import dataclasses16import datetime17from aws_lambda_api_event_utils import *18from tests.events import *19rand_str = lambda: secrets.token_hex(4)20# TODO:21# - header dedup22# - * for method and headers23def test_cors_is_preflight_request():24 for integration_type in IntegrationType:25 base_event = Event(26 integration_type=integration_type,27 method="OPTIONS",28 path=Path(stage="live", path="/foo", resource="/foo"),29 body=None,30 )31 assert not CORSConfig.is_preflight_request(base_event.get_event())32 assert CORSConfig.is_preflight_request(33 base_event.with_(34 headers={"Access-Control-Request-Method": "GET"}35 ).get_event()36 )37 assert not CORSConfig.is_preflight_request(38 base_event.with_(39 method="GET", headers={"Access-Control-Request-Method": "GET"}40 ).get_event()41 )42def test_cors_make_preflight_response():43 for integration_type in IntegrationType:44 base_cors_config = CORSConfig(45 allow_origin="https://example.com", allow_methods="GET"46 )47 response = base_cors_config.make_preflight_response(48 format_version=integration_type.format_version49 )50 assert response["statusCode"] == 20451 assert not response.get("body")52 assert (53 response["headers"]["Access-Control-Allow-Origin"] == "https://example.com"54 )55 assert response["headers"]["Access-Control-Allow-Methods"] == "OPTIONS, GET"56 assert "Access-Control-Allow-Headers" not in response["headers"]57 assert "Access-Control-Max-Age" not in response["headers"]58 assert "Access-Control-Credentials" not in response["headers"]59 assert "Access-Control-Expose-Headers" not in response["headers"]60 cors_config = dataclasses.replace(61 base_cors_config, allow_headers=["foo", "bar"]62 )63 response = cors_config.make_preflight_response(64 format_version=integration_type.format_version65 )66 assert response["headers"]["Access-Control-Allow-Headers"] == "foo, bar"67 cors_config = dataclasses.replace(base_cors_config, allow_headers="foo")68 response = cors_config.make_preflight_response(69 format_version=integration_type.format_version70 )71 assert response["headers"]["Access-Control-Allow-Headers"] == "foo"72 cors_config = dataclasses.replace(base_cors_config, max_age=300)73 response = cors_config.make_preflight_response(74 format_version=integration_type.format_version75 )76 assert response["headers"]["Access-Control-Max-Age"] == "300"77 cors_config = dataclasses.replace(78 base_cors_config, max_age=datetime.timedelta(minutes=5)79 )80 response = cors_config.make_preflight_response(81 format_version=integration_type.format_version82 )83 assert response["headers"]["Access-Control-Max-Age"] == "300"84 cors_config = dataclasses.replace(base_cors_config, allow_credentials=True)85 response = cors_config.make_preflight_response(86 format_version=integration_type.format_version87 )88 assert response["headers"]["Access-Control-Allow-Credentials"] == "true"89 cors_config = dataclasses.replace(90 base_cors_config, expose_headers=["foo", "bar"]91 )92 response = cors_config.make_preflight_response(93 format_version=integration_type.format_version94 )95 assert "Access-Control-Expose-Headers" not in response["headers"]96def test_get_headers():97 base_cors_config = CORSConfig(98 allow_origin="https://example.com", allow_methods="GET"99 )100 headers = base_cors_config.get_headers()101 assert headers["Access-Control-Allow-Origin"] == "https://example.com"102 assert "Access-Control-Allow-Methods" not in headers103 assert "Access-Control-Allow-Headers" not in headers104 assert "Access-Control-Max-Age" not in headers105 assert "Access-Control-Credentials" not in headers106 assert "Access-Control-Expose-Headers" not in headers107 cors_config = dataclasses.replace(base_cors_config, allow_headers=["foo", "bar"])108 headers = cors_config.get_headers()109 assert "Access-Control-Allow-Headers" not in headers110 cors_config = dataclasses.replace(111 base_cors_config, max_age=datetime.timedelta(minutes=5)112 )113 headers = cors_config.get_headers()114 assert "Access-Control-Max-Age" not in headers115 cors_config = dataclasses.replace(base_cors_config, allow_credentials=True)116 headers = cors_config.get_headers()117 assert headers["Access-Control-Allow-Credentials"] == "true"118 cors_config = dataclasses.replace(base_cors_config, expose_headers=["foo", "bar"])119 headers = cors_config.get_headers()120 assert headers["Access-Control-Expose-Headers"] == "foo, bar"121 cors_config = dataclasses.replace(base_cors_config, expose_headers="foo")122 headers = cors_config.get_headers()123 assert headers["Access-Control-Expose-Headers"] == "foo"124def test_dedup():125 cors_config = CORSConfig(126 allow_origin="https://example.com",127 allow_methods="GET",128 allow_headers="Accept",129 expose_headers="Content-Type",130 )131 assert cors_config.allow_methods == ("OPTIONS", "GET")132 assert cors_config.allow_headers == ("Accept",)133 assert cors_config.expose_headers == ("Content-Type",)134 cors_config = CORSConfig(135 allow_origin="https://example.com", allow_methods=["GET", "OPTIONS"]136 )137 assert cors_config.allow_methods == ("GET", "OPTIONS")138 cors_config = CORSConfig(139 allow_origin="https://example.com",140 allow_methods="GET",141 allow_headers=["Accept", "Content-Type", "Accept"],142 )143 assert cors_config.allow_headers == ("Accept", "Content-Type")144 cors_config = CORSConfig(145 allow_origin="https://example.com",146 allow_methods="GET",147 allow_headers=["Accept", "*", "Accept"],148 )149 assert cors_config.allow_headers == ("*",)150 cors_config = CORSConfig(151 allow_origin="https://example.com", allow_methods=["*", "GET"]152 )...
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?
