How to use is_request_valid method in localstack

Best Python code snippet using localstack_python

message_validation.py

Source:message_validation.py Github

copy

Full Screen

1# message.py2from message import Message3class MessageValidation:4 @classmethod5 # ------------------------------------------------------------------------------6 def validate_version_request(self, message: Message) -> bool:7 # --------------------------------------------------------------------------8 """Checks to see if incoming version request message is formatted according9 to our standards so node can handle the request without errors.10 """11 message_keys = {'version': False, 'services': False, 'timestamp': False, 'nonce': False,12 'address_from': False, 'address_receive': False, 'sub_version': False, 'start_accounts_count': False, 'relay': False}13 is_request_valid = True14 if message.type == 'request' and message.flag == 1 and isinstance(message.data, dict):15 for k in message.data:16 if k in message_keys:17 del message_keys[k]18 else:19 is_request_valid = False20 break21 if is_request_valid and len(message_keys) > 0:22 is_request_valid = False23 return is_request_valid24 @classmethod25 # ------------------------------------------------------------------------------26 def validate_version_response(self, message: Message) -> bool:27 # --------------------------------------------------------------------------28 """Checks to see if incoming version reponse message is formatted according29 to our standards so node can handle the request without errors.30 """31 return message.type == 'response' and message.flag == 1 and message.data is None32 @classmethod33 # ------------------------------------------------------------------------------34 def validate_address_request(self, message: Message) -> bool:35 # --------------------------------------------------------------------------36 """Checks to see if incoming peer address request message is formatted according37 to our standards so node can handle the request without errors.38 """39 message_keys = {'address_count': False, 'address_list': False}40 is_request_valid = True41 if message.type == 'request' and message.flag == 2 and isinstance(message.data, dict):42 for k in message.data:43 if k in message_keys:44 del message_keys[k]45 else:46 is_request_valid = False47 break48 if is_request_valid and len(message_keys) > 0:49 is_request_valid = False50 return is_request_valid51 @classmethod52 # ------------------------------------------------------------------------------53 def validate_address_response(self, message: Message) -> bool:54 # --------------------------------------------------------------------------55 """Checks to see if incoming peer address reponse message is formatted according56 to our standards so node can handle the request without errors.57 """58 message_keys = {'address_count': False, 'address_list': False}59 is_request_valid = True60 if message.type == 'response' and message.flag == 2 and isinstance(message.data, dict):61 for k in message.data:62 if k in message_keys:63 del message_keys[k]64 else:65 is_request_valid = False66 break67 if is_request_valid and len(message_keys) > 0:68 is_request_valid = False69 return is_request_valid70 @classmethod71 # ------------------------------------------------------------------------------72 def validate_states_request(self, message: Message) -> bool:73 # --------------------------------------------------------------------------74 """Checks to see if incoming account request message is formatted according75 to our standards so node can handle the request without errors.76 """77 message_keys = {'version': False,78 'account': False, 'best_state': False, }79 is_request_valid = True80 if message.type == 'request' and message.flag == 3 and isinstance(message.data, dict):81 for k in message.data:82 if k in message_keys:83 del message_keys[k]84 else:85 is_request_valid = False86 break87 if is_request_valid and len(message_keys) > 0:88 is_request_valid = False89 return is_request_valid90 @classmethod91 # ------------------------------------------------------------------------------92 def validate_states_response(self, message: Message) -> bool:93 # --------------------------------------------------------------------------94 """Checks to see if incoming states reponse message is formatted according95 to our standards so node can handle the request without errors.96 """97 message_keys = {'inventory_count': False, 'inventory': False, }98 is_request_valid = True99 if message.type == 'response' and message.flag == 4 and isinstance(message.data, dict):100 for k in message.data:101 if k in message_keys:102 del message_keys[k]103 else:104 is_request_valid = False105 break106 if is_request_valid and len(message_keys) > 0:107 is_request_valid = False108 return is_request_valid109 @classmethod110 # ------------------------------------------------------------------------------111 def validate_data_request(self, message: Message) -> bool:112 # --------------------------------------------------------------------------113 """Checks to see if incoming data request message is formatted according114 to our standards so node can handle the request without errors.115 """116 message_keys = {'inventory_count': False, 'inventory': False}117 is_request_valid = True118 if message.type == 'request' and message.flag == 4 and isinstance(message.data, dict):119 for k in message.data:120 if k in message_keys:121 del message_keys[k]122 else:123 is_request_valid = False124 break125 if is_request_valid and len(message_keys) > 0:126 is_request_valid = False127 return is_request_valid...

Full Screen

Full Screen

test_api.py

Source:test_api.py Github

copy

Full Screen

1from nose.tools import eq_, ok_, assert_raises2from amazon import AmazonAPI, AmazonAPIError, AmazonAPIResponseError3from config import AWS_KEY_ID, SECRET_KEY, ASSOCIATE_TAG_ID4amz = AmazonAPI(5 aws_access_key=AWS_KEY_ID,6 secret_key=SECRET_KEY,7 associate_tag=ASSOCIATE_TAG_ID8)9# ===============================================================10#11# Item Lookup Unit Tests12#13# ===============================================================14def test_item_lookup_request_validity():15 response = amz.item_lookup(16 host="us",17 IdType="ASIN",18 ItemId="B0041OSCBU",19 ResponseGroup="ItemAttributes,Images"20 )21 convert_str_to_bool = lambda x: x.lower() in ("yes", "true", "t", "1")22 is_request_valid = convert_str_to_bool(response.Items.IsValid.string)23 eq_(is_request_valid,24 True,25 msg="XML tag 'IsValid' is False, Invalid Request")26def test_item_lookup_response_has_ASIN():27 response = amz.item_lookup(28 host="us",29 IdType="ASIN",30 ItemId="B0041OSCBU",31 ResponseGroup="ItemAttributes,Images"32 )33 ASIN = response.Items.Item.ASIN.string34 ok_(len(ASIN) > 0, msg="There is no ASIN in the response")35def test_missing_null_host():36 assert_raises(37 AmazonAPIError,38 amz.item_lookup,39 IdType="ASIN",40 ItemId="B0041OSCBU",41 ResponseGroup="ItemAttributes,Images"42 )43def test_invalid_host():44 assert_raises(45 AmazonAPIError,46 amz.item_lookup,47 host="co",48 IdType="ASIN",49 ItemId="B0041OSCBU",50 ResponseGroup="ItemAttributes,Images"51 )52def test_missing_ItemId():53 assert_raises(54 AmazonAPIResponseError,55 amz.item_lookup,56 host="us",57 IdType="ASIN",58 ResponseGroup="ItemAttributes,Images"59 )60# ===============================================================61#62# Item Search Unit Tests63#64# ===============================================================65def test_item_search_request_validity():66 response = amz.item_search(67 host="us",68 Keywords="Harry Poter",69 SearchIndex="All",70 ResponseGroup="ItemAttributes,Images"71 )72 convert_str_to_bool = lambda x: x.lower() in ("yes", "true", "t", "1")73 is_request_valid = convert_str_to_bool(response.Items.IsValid.string)74 eq_(is_request_valid,75 True,76 msg="XML tag 'IsValid' is False, Invalid Request")77def test_missing_keywords():78 assert_raises(79 AmazonAPIResponseError,80 amz.item_lookup,81 host="us",82 ResponseGroup="ItemAttributes,Images"83 )84# ===============================================================85#86# Item Search Unit Tests87#88# ===============================================================89def test_similarity_lookup_request_validity():90 response = amz.similarity_lookup(91 host="us",92 ItemId="B0011ZK6PC,B000NK8EWI",93 ResponseGroup="ItemAttributes,Images",94 SimilarityType="Intersection",95 Merchant="Amazon"96 )97 convert_str_to_bool = lambda x: x.lower() in ("yes", "true", "t", "1")98 is_request_valid = convert_str_to_bool(response.Items.IsValid.string)99 eq_(is_request_valid,100 True,101 msg="XML tag 'IsValid' is False, Invalid Request")102def test_similarity_lookup_three_ASIN():103 response = amz.similarity_lookup(104 host="ca",105 ItemId="B001ASBBSG,B001AS94XK,B003YL3MI4",106 ResponseGroup="ItemAttributes,Images",107 SimilarityType="Intersection",108 Merchant="Amazon"109 )110 convert_str_to_bool = lambda x: x.lower() in ("yes", "true", "t", "1")111 is_request_valid = convert_str_to_bool(response.Items.IsValid.string)112 eq_(is_request_valid,113 True,114 msg="XML tag 'IsValid' is False, Invalid Request")115def test_similarity_for_no_similarities():116 assert_raises(AmazonAPIResponseError,117 amz.similarity_lookup,118 host="us",119 ItemId="B0011ZK6PC,B00CDIK908",120 ResponseGroup="ItemAttributes,Images",121 SimilarityType="Intersection",122 Merchant="Amazon")123# ===============================================================124#125# BrowseNodeLookup Unit Tests126#127# ===============================================================128def test_browse_node_lookup_request_validity():129 response = amz.node_browse_lookup(host="us", browse_node_id=11091801)130 is_request_valid = response.BrowseNodes.Request.IsValid.string131 eq_(is_request_valid,132 "True",...

Full Screen

Full Screen

middleware.py

Source:middleware.py Github

copy

Full Screen

1"""Security middleware."""2from ipware import get_client_ip3from django.conf import settings4import ipaddress5from utils.utils import get_hashed_mac6from .models import AccessTokens7from django.http import JsonResponse8from django.urls import reverse9from jwt import decode10class AccessVerificationMiddleware:11 """Access verification middleware."""12 def __init__(self, get_response):13 """Initialize the middleware."""14 self.get_response = get_response15 self.permissible_ips = ipaddress.ip_network(settings.DEMO_SECURITY.allowed_ips)16 def __call__(self, request):17 """Verify if the request has access to the APIs."""18 is_request_valid = (19 not settings.DEMO_SECURITY.security_check_required20 or request.path == reverse("pin-verify")21 )22 if not is_request_valid:23 client_ip, is_routable = get_client_ip(request)24 is_ip_safe = ipaddress.ip_address(client_ip) in self.permissible_ips25 if not is_ip_safe and "HTTP_TOKEN" in request.META:26 token = request.META["HTTP_TOKEN"]27 obj = AccessTokens.objects.filter(token__exact=token).first()28 if obj:29 is_request_valid = True30 is_request_valid = is_request_valid or is_ip_safe31 if is_request_valid:32 return self.get_response(request)33 else:34 return JsonResponse({"message": "Invalid request."}, status=401)35class UserAccessMiddleware:36 def __init__(self, get_response):37 self.get_response = get_response38 def __call__(self, request):39 try:40 auth_header = request.META.get("HTTP_AUTHORIZATION", None)41 if auth_header is not None:42 print(auth_header)43 token = auth_header44 payload = decode(45 jwt=token,46 key=settings.SECRET_KEY,47 verify=True,48 algorithms=["HS256"],49 )50 mac = payload.get("hash", None)51 print(mac)52 if mac is not None:53 request.mac_address = mac54 else:55 request.mac_address = None56 except Exception as e:57 # print(e)58 request.mac_address = None59 # print(request.mac_address)60 response = self.get_response(request)...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful