How to use load_json method in tempest

Best Python code snippet using tempest_python

paga_collect.py

Source:paga_collect.py Github

copy

Full Screen

1import requests2import hashlib3import json4from base64 import b64encode5def _get_basic_auth(client_id, password):6 base64_string = b64encode(str.encode(client_id + ":" + password)).decode("ascii")7 return base64_string8def generate_hash(hash_params):9 """10 Generates hash based on passed11 args12 ----------13 hash_params : string14 A concatenated string of parameters be hashed15 """16 return hashlib.sha512(str(hash_params.strip()).encode("utf-8")).hexdigest().strip()17def post_request(headers, json_data, url):18 """19 Posts request to given url with the given payload20 args21 ----------22 headers : dict23 Holds authentication data for the request24 json_data : json25 The request payload26 url : boolean27 The api url28 """29 return requests.request(method="POST", url=url, headers=headers, data=json_data)30def remove_empty_elements(value):31 """32 Recursively remove all None values from dictionaries and lists, and returns33 the result as a new dictionary or list.34 """35 if isinstance(value, list):36 return [remove_empty_elements(x) for x in value if x is not None]37 elif isinstance(value, dict):38 return {39 key: remove_empty_elements(val)40 for key, val in value.items()41 if val is not None42 }43 else:44 return value45class Collect(object):46 """47 Base class for paga Collect api library48 """49 _CONTENT_TYPE = "application/json"50 test_server = "https://beta-collect.paga.com"51 live_Server = "https://collect.paga.com"52 def __init__(self, client_id, password, api_key, is_test_env):53 """54 args55 ----------56 client_id : string57 your public ID gotten from Paga58 password : string59 your account password60 is_test_env : boolean61 indicates whether application is in test or live mode62 """63 self.client_id = client_id64 self.password = password65 self.api_key = api_key66 self.is_test_env = is_test_env67 def build_header(self, hash_params):68 """69 Builds the HTTP request header 70 args71 ----------72 hash_params : string73 A concatenated string of parameters be hashed74 """75 basic_auth = _get_basic_auth(self.client_id, self.password)76 hash_params = hash_params + self.api_key77 hash_data = generate_hash(hash_params)78 headers = {79 "Content-Type": self._CONTENT_TYPE,80 "Authorization": "Basic " + basic_auth,81 "hash": hash_data82 }83 return headers84 def get_url(self, is_test_env):85 """86 Gets the api url 87 args88 ----------89 is_test_env : boolean90 A flag to determine if the requested url is test or live91 """92 if is_test_env:93 return self.test_server94 else:95 return self.live_Server96 def get_history(self, payload):97 """98 Calls Collect API to get the history of transactions done over time 99 args100 ----------101 payload : json102 A request body for the history endpoint of the Collect API103 """104 endpoint = "/history"105 hash_params = ''106 url = self.get_url(self.is_test_env)107 server_url = url + endpoint108 109 data = json.dumps(payload)110 load_json = json.loads(data)111 112 hash_json = {113 "referenceNumber": load_json.get('referenceNumber') or None114 }115 116 request_data = {117 "referenceNumber": load_json.get('referenceNumber') or None,118 "startDateTimeUTC": load_json.get("startDateTimeUTC"),119 "endDateTimeUTC": load_json.get("endDateTimeUTC") or None,120 }121 hash_data = json.dumps(remove_empty_elements(hash_json)) 122 hash_params = ''.join(list(json.loads(hash_data).values()))123 headers = self.build_header(hash_params)124 json_data = json.dumps(remove_empty_elements(request_data))125 126 127 response = post_request(headers, json_data, server_url)128 return response.text129 def payment_request_refund(self, payload):130 """131 Calls Collect API to cancel or initiate a refund if we were unable to fulfill 132 the request for one reason or the other133 ----------134 payload : json135 A request body for the paymentRequestRefund endpoint of the Collect API136 """137 endpoint = "/refund"138 hash_params = ''.join([payload["referenceNumber"], payload["refundAmount"]])139 url = self.get_url(self.is_test_env)140 server_url = url + endpoint141 headers = self.build_header(hash_params)142 data = json.dumps(payload)143 load_json = json.loads(data)144 request_data = {145 "referenceNumber": load_json.get('referenceNumber'),146 "refundAmount": load_json.get("refundAmount"),147 "currency": load_json.get("currency"),148 "reason": load_json.get("reason") or None,149 } 150 json_data = json.dumps(remove_empty_elements(request_data))151 152 153 response = post_request(headers, json_data, server_url)154 return response.text155 156 def delete_persistent_payment_account(self, payload):157 """158 Calls Collect API to delete a persistent payment account. 159 160 ----------161 payload : json162 A request body for the deletePersistentPaymentAccount endpoint of the Collect API163 """164 endpoint = "/deletePersistentPaymentAccount"165 hash_params = ''.join([payload["referenceNumber"], payload["accountIdentifier"]])166 url = self.get_url(self.is_test_env)167 server_url = url + endpoint168 headers = self.build_header(hash_params)169 data = json.dumps(payload)170 load_json = json.loads(data)171 request_data = {172 "referenceNumber": load_json.get('referenceNumber'),173 "accountIdentifier": load_json.get("accountIdentifier"),174 "reason": load_json.get("endDateTimeUTC") or None,175 } 176 json_data = json.dumps(remove_empty_elements(request_data))177 178 179 response = post_request(headers, json_data, server_url)180 return response.text181 def get_persistent_payment_account(self, payload):182 """183 Calls Collect API to to query the properties associated with an existing persistent payment account. 184 185 ----------186 payload : json187 A request body for the getPersistentPaymentAccount endpoint of the Collect API188 """189 endpoint = "/getPersistentPaymentAccount"190 hash_params = ''.join([payload["referenceNumber"], payload["accountIdentifier"]])191 url = self.get_url(self.is_test_env)192 server_url = url + endpoint193 headers = self.build_header(hash_params)194 data = json.dumps(payload)195 load_json = json.loads(data)196 request_data = {197 "referenceNumber": load_json.get('referenceNumber'),198 "accountIdentifier": load_json.get("accountIdentifier"),199 } 200 json_data = json.dumps(request_data)201 202 203 response = post_request(headers, json_data, server_url)204 return response.text205 def update_persistent_payment_account(self, payload):206 """207 This endpoint allows for changing any of the account properties except the accountNumber (NUBAN) and the accounReference properties which cannot be changed. 208 209 ----------210 payload : json211 A request body for the updatePersistentPaymentAccount endpoint of the Collect API212 """213 endpoint = "/updatePersistentPaymentAccount"214 url = self.get_url(self.is_test_env)215 server_url = url + endpoint216 data = json.dumps(payload)217 load_json = json.loads(data)218 request_data = {219 "referenceNumber": load_json.get('referenceNumber'),220 "accountIdentifier": load_json.get("accountIdentifier"),221 "phoneNumber": load_json.get("phoneNumber") or None,222 "firstName": load_json.get("firstName") or None,223 "lastName": load_json.get("endDateTimeUTC") or None,224 "accountName": load_json.get("lastName") or None,225 "financialIdentificationNumber": load_json.get("financialIdentificationNumber") or None,226 "callbackUrl": load_json.get("callbackUrl") or None,227 "creditBankId": load_json.get("creditBankId") or None,228 "creditBankAccountNumber": load_json.get("creditBankAccountNumber") or None,229 } 230 hash_json = {231 "referenceNumber": load_json.get('referenceNumber'),232 "accountIdentifier": load_json.get("accountIdentifier"),233 "financialIdentificationNumber": load_json.get("financialIdentificationNumber") or None,234 "creditBankId": load_json.get("creditBankId") or None,235 "creditBankAccountNumber": load_json.get("creditBankAccountNumber") or None,236 "callbackUrl": load_json.get("callbackUrl") or None,237 }238 hash_data = json.dumps(remove_empty_elements(hash_json))239 hash_params = ''.join(list(json.loads(hash_data).values()))240 print(hash_params)241 headers = self.build_header(hash_params)242 json_data = json.dumps(remove_empty_elements(request_data))243 244 245 response = post_request(headers, json_data, server_url)246 return response.text247 248 def get_status(self, payload):249 """250 Calls Collect API to get the status of a transaction251 args252 ----------253 payload : json254 A request body for the status endpoint of the Collect API255 """256 endpoint = "/status"257 hash_params = ''258 if payload.get("referenceNumber"):259 hash_params += payload["referenceNumber"]260 url = self.get_url(self.is_test_env)261 server_url = url + endpoint262 headers = self.build_header(hash_params)263 json_data = json.dumps(payload)264 response = post_request(headers, json_data, server_url)265 return response.text266 def get_banks(self, payload):267 """268 Calls Collect API to get all banks 269 args270 ----------271 payload : json272 A request body for the getBanks endpoint of the Collect API273 """274 endpoint = "/banks"275 hash_params = payload["referenceNumber"]276 url = self.get_url(self.is_test_env)277 server_url = url + endpoint278 headers = self.build_header(hash_params)279 json_data = json.dumps(payload)280 response = post_request(headers, json_data, server_url)281 return response.text282 def payment_request(self, payload):283 """284 Calls Collect API to make a payment request 285 args286 ----------287 payload : json288 A request body for the paymentRequest endpoint of the Collect API289 """290 endpoint = "/paymentRequest"291 url = self.get_url(self.is_test_env)292 server_url = url + endpoint293 data = json.dumps(payload)294 load_json = json.loads(data)295 payee = {296 "name": load_json.get("payee").get("name"),297 "accountNumber": load_json.get("payee").get("accountNumber") or None,298 "phoneNumber": load_json.get("payee").get("phoneNumber") or None,299 "bankId": load_json.get("payee").get("bankId") or None,300 "bankAccountNumber": load_json.get("payee").get("bankAccountNumber") or None,301 "financialIdentificationNumber": load_json.get("payee").get("financialIdentificationNumber") or None,302 }303 payer = {304 "name": load_json.get("payer").get("name"),305 "phoneNumber": load_json.get("payer").get("phoneNumber") or None,306 "email": load_json.get("payer").get("email") or None,307 "bankId": load_json.get("payee").get("bankId") or None,308 }309 payment_request_payload = {310 "referenceNumber": load_json.get("referenceNumber"),311 "amount": str(load_json.get("amount")),312 "currency": load_json.get("currency"),313 "payer": payer,314 "payee": payee,315 "expiryDateTimeUTC": load_json.get("expiryDateTimeUTC") or None,316 "isSuppressMessages": load_json.get("isSuppressMessages"),317 "payerCollectionFeeShare": load_json.get("payerCollectionFeeShare"),318 "payeeCollectionFeeShare": load_json.get("payeeCollectionFeeShare"),319 "isAllowPartialPayments": load_json.get("isAllowPartialPayments") or None,320 "callBackUrl": load_json.get("callBackUrl"),321 "paymentMethods": load_json.get("paymentMethods"),322 "displayBankDetailToPayer": load_json.get("displayBankDetailToPayer") or None323 }324 request_hash = {325 "referenceNumber": load_json.get("referenceNumber"),326 "amount": str(load_json.get("amount")),327 "currency": load_json.get("currency"),328 }329 payee_hash = {330 "accountNumber": load_json.get("payee").get("accountNumber") or None,331 "phoneNumber": load_json.get("payee").get("phoneNumber") or None,332 "bankId": load_json.get("payee").get("bankId") or None,333 "bankAccountNumber": load_json.get("payee").get("bankAccountNumber") or None,334 }335 payer_hash = {336 "phoneNumber": load_json.get("payer").get("phoneNumber") or None,337 "email": load_json.get("payer").get("email") or None,338 }339 340 payee_hash_data = json.dumps(remove_empty_elements(payee_hash))341 request_hash_data = json.dumps(request_hash)342 payer_hash_data = json.dumps(remove_empty_elements(payer_hash))343 hash_params = ''.join(list(json.loads(request_hash_data).values())) + ''.join(list(json.loads(payer_hash_data).values())) + ''.join(list(json.loads(payee_hash_data).values()))344 print(hash_params)345 headers = self.build_header(hash_params)346 json_data = json.dumps(remove_empty_elements(payment_request_payload))347 348 349 response = post_request(headers, json_data, server_url)350 return response.text351 def register_persistent_payment_account(self, payload):352 """353 Calls Collect API to create persistent payment account354 args355 ----------356 payload : json357 A request body for the register payment persistent account endpoint of the Collect API358 """359 endpoint = "/registerPersistentPaymentAccount"360 url = self.get_url(self.is_test_env)361 server_url = url + endpoint362 data = json.dumps(payload)363 load_json = json.loads(data)364 request_data = {365 "referenceNumber": load_json.get("referenceNumber"),366 "phoneNumber": load_json.get("phoneNumber"),367 "firstName": load_json.get("firstName"),368 "lastName": load_json.get("lastName") or None,369 "accountName": load_json.get("accountName"),370 "email": load_json.get("email"),371 "financialIdentificationNumber": load_json.get("financialIdentificationNumber") or None,372 "accountReference": load_json.get("accountReference"),373 "creditBankId": load_json.get("creditBankId") or None,374 "creditBankAccountNumber": load_json.get("creditBankAccountNumber") or None,375 "callbackUrl": load_json.get("callbackUrl") or None,376 }377 hash_json = {378 "referenceNumber": load_json.get("referenceNumber"),379 "accountReference": load_json.get("accountReference"),380 "financialIdentificationNumber": load_json.get("financialIdentificationNumber") or None,381 "creditBankId": load_json.get("creditBankId") or None,382 "creditBankAccountNumber": load_json.get("creditBankAccountNumber") or None,383 "callbackUrl": load_json.get("callbackUrl") or None,384 }385 hash_data = json.dumps(remove_empty_elements(hash_json))386 hash_params = ''.join(list(json.loads(hash_data).values()))387 headers = self.build_header(hash_params)388 json_data = json.dumps(remove_empty_elements(request_data))389 390 391 response = post_request(headers, json_data, server_url)...

Full Screen

Full Screen

test_pygeojson.py

Source:test_pygeojson.py Github

copy

Full Screen

...17 MultiPolygon,18)19from json import JSONDecodeError20@pytest.fixture21def load_json():22 def _load_json(path: str):23 with open("assets/%s" % (path,)) as f:24 return load(f)25 return _load_json26def test_feature_with_extra_attributes(load_json):27 o: Feature = load_json("feature_with_extra_attributes.json")28 assert o == Feature(29 id=None,30 bbox=None,31 type="Feature",32 geometry=None,33 properties={},34 extra_attributes={"foo": "bar"},35 )36def test_feature_with_id(load_json):37 o: Feature = load_json("feature_with_id.json")38 assert o == Feature(39 id="i_have_a_string_id",40 bbox=None,41 type="Feature",42 geometry=Point(coordinates=(102.0, 0.5)),43 properties={},44 )45def test_feature_with_null_geom(load_json):46 o: Feature = load_json("feature_with_null_geom.json")47 assert o == Feature(48 id=None, bbox=None, type="Feature", geometry=None, properties={},49 )50def test_feature_with_props(load_json):51 o: Feature = load_json("feature_with_props.json")52 assert o == Feature(53 id=1337,54 type="Feature",55 bbox=(100.0, 0.0, 101.0, 1.0),56 geometry=Polygon(57 coordinates=[58 [(100.0, 0.0), (101.0, 0.0), (101.0, 1.0), (100.0, 1.0), (100.0, 0.0)]59 ],60 ),61 properties={"prop0": "value0", "prop1": {"this": "that"}},62 )63def test_feature_without_geom(load_json):64 o: Feature = load_json("feature_without_geom.json")65 assert o == Feature(66 id=None, bbox=None, type="Feature", geometry=None, properties={},67 )68def test_feature(load_json):69 o: Feature = load_json("feature.json")70 assert o == Feature(71 id=None,72 type="Feature",73 bbox=None,74 geometry=LineString(75 coordinates=[(102.0, 0.0), (103.0, 1.0), (104.0, 0.0), (105.0, 1.0)]76 ),77 properties={},78 )79def test_equality(load_json):80 assert load_json("feature.json") == load_json("feature.json")81def test_inequality(load_json):82 assert load_json("feature.json") != load_json("feature_with_id.json")83def test_featurecollection_with_extra_attributes_access(load_json):84 o: FeatureCollection = load_json("featurecollection_with_extra_attributes.json")85 assert o["crs"]["properties"]["name"] == "urn:ogc:def:crs:EPSG::25832"86def test_featurecollection_not_equal(load_json):87 assert load_json("featurecollection.json") != load_json(88 "featurecollection_with_extra_attributes.json"89 )90def test_featurecollection(load_json):91 o: FeatureCollection = load_json("featurecollection.json")92 assert o == FeatureCollection(93 type="FeatureCollection",94 bbox=None,95 features=[96 Feature(97 id="i_have_a_string_id",98 geometry=Point(coordinates=(102.0, 0.5)),99 properties={"prop0": "value0"},100 ),101 Feature(102 id=1337,103 geometry=LineString(104 coordinates=[(102.0, 0.0), (103.0, 1.0), (104.0, 0.0), (105.0, 1.0)]105 ),106 properties={"prop0": "value0", "prop1": 0.0},107 ),108 ],109 )110def test_geometrycollection(load_json):111 o: GeometryCollection = load_json("geometrycollection.json")112 assert o == GeometryCollection(113 type="GeometryCollection",114 geometries=[Point((100.0, 0.0)), LineString([(101.0, 0.0), (102.0, 1.0)])],115 )116def test_linestring(load_json):117 o: LineString = load_json("linestring.json")118 assert o == LineString(119 type="LineString",120 coordinates=[(102.0, 0.0), (103.0, 1.0), (104.0, 0.0), (105.0, 1.0)],121 )122def test_multilinestring(load_json):123 o: MultiLineString = load_json("multilinestring.json")124 assert o == MultiLineString(125 type="MultiLineString",126 coordinates=[[(100.0, 0.0), (101.0, 1.0)], [(102.0, 2.0), (103.0, 3.0)]],127 )128def test_multipoint(load_json):129 o: MultiPoint = load_json("multipoint.json")130 assert o == MultiPoint(type="MultiPoint", coordinates=[(100.0, 0.0), (101.0, 1.0)])131def test_multipolygon(load_json):132 o: MultiPolygon = load_json("multipolygon.json")133 assert o == MultiPolygon(134 type="MultiPolygon",135 coordinates=[136 [[(102.0, 2.0), (103.0, 2.0), (103.0, 3.0), (102.0, 3.0), (102.0, 2.0)]],137 [138 [(100.0, 0.0), (101.0, 0.0), (101.0, 1.0), (100.0, 1.0), (100.0, 0.0)],139 [(100.2, 0.2), (100.8, 0.2), (100.8, 0.8), (100.2, 0.8), (100.2, 0.2)],140 ],141 ],142 )143def test_not_a_geojson(load_json):144 with pytest.raises(GeoJSONDecodeError):145 load_json("not_a_geojson.json")146def test_not_a_json(load_json):147 with pytest.raises(JSONDecodeError):148 load_json("not_a_json.json")149def test_point(load_json):150 o: Point = load_json("point.json")151 assert o == Point(type="Point", coordinates=(102.0, 0.5))152def test_polygon(load_json):153 o: Polygon = load_json("polygon.json")154 assert o == Polygon(155 type="Polygon",156 coordinates=[157 [(100.0, 0.0), (101.0, 0.0), (101.0, 1.0), (100.0, 1.0), (100.0, 0.0)]158 ],159 )160def test_loads():161 assert (162 loads(163 """164 {165 "type": "Point",166 "coordinates": [1.0, 2.0]167 }"""168 )169 == Point((1.0, 2.0))170 )171def test_dumps():172 assert (173 dumps(Point((1.0, 2.0))) == """{"type": "Point", "coordinates": [1.0, 2.0]}"""174 )175@pytest.mark.parametrize(176 "file",177 [178 "feature_with_extra_attributes.json",179 "feature_with_id.json",180 "feature_with_null_geom.json",181 "feature_with_props.json",182 "feature_without_geom.json",183 "feature.json",184 "featurecollection_with_extra_attributes.json",185 "geometrycollection.json",186 "linestring.json",187 "multilinestring.json",188 "multipoint.json",189 "multipolygon.json",190 "point.json",191 "polygon.json",192 ],193)194def test_read_write_read(load_json, file):195 # If we read, dump, and read again, we should have the same results as read196 assert loads(dumps(load_json(file))) == load_json(file)197def test_dump(tmp_path):198 p = Point((1.0, 2.0))199 with open(tmp_path / "point.json", "w") as f:200 dump(p, f)201 with open(tmp_path / "point.json") as f:202 assert load(f) == p203def test_loads_feature_collection():204 # should not crash205 with open("assets/featurecollection.json") as f:206 load_feature_collection(f)207def test_loads_feature_collection_invalid():208 # should not crash209 with open("assets/point.json") as f:210 with pytest.raises(TypeError):...

Full Screen

Full Screen

api.py

Source:api.py Github

copy

Full Screen

1import json2import requests3class api_func():4 # Gets data from api.5 # Returns global data in array6 def overview_corona():7 api_call = requests.get("{}/all".format("https://coronavirus-19-api.herokuapp.com"))8 load_json = json.loads(api_call.text)9 cases = load_json.get('cases')10 deaths = load_json.get('deaths')11 recovered = load_json.get('recovered')12 return [cases, deaths, recovered]13 # Gets data from API about one specific country. Output is array.14 def country_corona(country):15 try:16 api_call = requests.get("{}/countries/{}".format("https://coronavirus-19-api.herokuapp.com", country))17 load_json = json.loads(api_call.text)18 country_name = load_json.get('country')19 cases = load_json.get('cases')20 today_cases = load_json.get('todayCases')21 deaths = load_json.get('deaths')22 today_deaths = load_json.get('todayDeaths')23 recovered_people = load_json.get('recovered')24 active_cases = load_json.get('active')25 critical_cases = load_json.get('critical')26 cases_per_one_million_citizens = load_json.get('casesPerOneMillion')27 deaths_per_one_million_citizens = load_json.get('deathsPerOneMillion')28 total_tests = load_json.get('totalTests')29 tests_per_one_million = load_json.get('testsPerOneMillion')30 data = [country_name, cases, today_cases, deaths, today_deaths, recovered_people, active_cases, critical_cases, cases_per_one_million_citizens, deaths_per_one_million_citizens, total_tests, tests_per_one_million]31 return data32 except:...

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