KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use update_endpoint method in localstack
Best Python code snippet using localstack_python
test_fuseki.py
Source:test_fuseki.py 
1"""Test cases for the fuseki triplestore."""2from os import environ as env3from typing import Any4from dotenv import load_dotenv5import pytest6from rdflib import Graph, Literal, URIRef7from rdflib.compare import graph_diff, isomorphic8load_dotenv()9DATASET = env.get("DATASET_1", "ds")10PASSWORD = env.get("PASSWORD")11PREFIX = """12 PREFIX dct: <http://purl.org/dc/terms/>13 PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>14 PREFIX owl: <http://www.w3.org/2002/07/owl#>15 PREFIX xml: <http://www.w3.org/XML/1998/namespace>16 PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>17 PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>18 PREFIX vcard: <http://www.w3.org/2006/vcard/ns#>19 PREFIX dcat: <http://www.w3.org/ns/dcat#>20 PREFIX dc: <http://purl.org/dc/elements/1.1/>21 """22def test_insert_triples_into_named_graph_with_SPARQLWrapper(http_service: Any) -> None:23 """Should return some status and the graph is persisted."""24 from SPARQLWrapper import SPARQLWrapper, POST, TURTLE25 update_endpoint = f"{http_service}/{DATASET}/update"26 print(update_endpoint)27 sparql = SPARQLWrapper(update_endpoint)28 sparql.setCredentials("admin", PASSWORD)29 sparql.setMethod(POST)30 querystring = (31 PREFIX32 + """33 INSERT DATA34 { GRAPH <http://example.com/publisher/2>35 {36 <http://example.com/publisher/2/catalogs/1>37 a dcat:Catalog ;38 dct:publisher <https://example.com/publishers/2> ;39 dct:title "Dataservicekatalog for Anna Eksempel AS"@nb ;40 dcat:service <http://example.com/dataservices/2> ,41 <http://example.com/dataservices/1>42 .43 <http://example.com/dataservices/1> a dcat:DataService ;44 dct:description "Exposes a model-catalog"@nb ;45 dct:title "Model-catalog of Digdir"@nb ;46 dcat:contactPoint <http://example.com/contactpoint/2> ;47 dcat:endpointDescription48 <http://example.com/description/model-catalog.yaml> ;49 .50 <http://example.com/dataservices/2> a dcat:DataService ;51 dct:description "Exposes a collection of dataservice-catalogs"@nb ;52 dct:title "Dataservice-catalog of Digdir"@nb ;53 dcat:contactPoint <http://example.com/contactpoint/2> ;54 dcat:endpointDescription55 <http://example.com/description/dataservice-catalog.yaml> ;56 .57 <http://example.com/contactpoint/2> a vcard:Organization ;58 vcard:hasOrganizationName "Digitaliseringsdirektoratet"@nb ;59 vcard:hasURL <https://digdir.no> ;60 .61 }62 }63 """64 )65 sparql.setQuery(querystring)66 results = sparql.query()67 assert 200 == results.response.status68 query_endpoint = f"{http_service}/{DATASET}/query"69 querystring = """70 CONSTRUCT { ?s ?p ?o }71 WHERE {72 GRAPH <http://example.com/publisher/2> {?s ?p ?o}73 }74 """75 sparql = SPARQLWrapper(query_endpoint)76 sparql.setQuery(querystring)77 sparql.setReturnFormat(TURTLE)78 sparql.setOnlyConneg(True)79 results = sparql.query()80 assert 200 == results.response.status81 assert "text/turtle; charset=utf-8" == results.response.headers["Content-Type"]82 data = results.convert()83 g = Graph()84 g.parse(data=data, format="turtle")85 assert len(g) == 1886def test_insert_graph_into_named_graph_with_SPARQLWrapper(http_service: Any) -> None:87 """Should return some status and the graph is persisted."""88 from SPARQLWrapper import SPARQLWrapper, POST, TURTLE89 identifier = "http://example.com/publisher/1"90 g1 = Graph().parse("tests/catalog_1.ttl", format="turtle")91 update_endpoint = f"{http_service}/{DATASET}/update"92 print(update_endpoint)93 sparql = SPARQLWrapper(update_endpoint)94 sparql.setCredentials("admin", PASSWORD)95 sparql.setMethod(POST)96 prefixes = ""97 for ns in g1.namespaces():98 prefixes += f"PREFIX {ns[0]}: <{ns[1]}>\n"99 print(prefixes)100 for s, p, o in g1:101 if isinstance(o, Literal):102 querystring = (103 prefixes104 + """105 INSERT DATA {GRAPH <%s> {<%s> <%s> "%s"@%s}}106 """107 % (identifier, s, p, o, o.language,)108 )109 else:110 querystring = (111 prefixes112 + """113 INSERT DATA {GRAPH <%s> {<%s> <%s> <%s>}}114 """115 % (identifier, s, p, o,)116 )117 print(querystring)118 sparql.setQuery(querystring)119 results = sparql.query()120 assert 200 == results.response.status121 query_endpoint = f"{http_service}/{DATASET}/query"122 querystring = """123 CONSTRUCT { ?s ?p ?o }124 WHERE {125 GRAPH %s {?s ?p ?o}126 }127 """ % (128 identifier129 )130 sparql = SPARQLWrapper(query_endpoint)131 sparql.setQuery(querystring)132 sparql.setReturnFormat(TURTLE)133 sparql.setOnlyConneg(True)134 results = sparql.query()135 assert 200 == results.response.status136 assert "text/turtle; charset=utf-8" == results.response.headers["Content-Type"]137 data = results.convert()138 g2 = Graph()139 g2.parse(data=data, format="turtle")140 assert len(g1) == len(g2)141 _isomorphic = isomorphic(g1, g2)142 if not _isomorphic:143 _dump_diff(g1, g2)144 pass145 assert _isomorphic146def test_describe_query_with_SPARQLWrapper(http_service: Any) -> None:147 """Should return some status and the graph is persisted."""148 from SPARQLWrapper import SPARQLWrapper, TURTLE149 query_endpoint = f"{http_service}/{DATASET}/query"150 print(query_endpoint)151 querystring = "DESCRIBE <http://example.com/publisher/1/catalogs/1>"152 sparql = SPARQLWrapper(query_endpoint)153 sparql.setQuery(querystring)154 sparql.setReturnFormat(TURTLE)155 sparql.setOnlyConneg(True)156 results = sparql.query()157 assert 200 == results.response.status158 assert "text/turtle; charset=utf-8" == results.response.headers["Content-Type"]159 data = results.convert()160 g1 = Graph()161 g1.parse(data=data, format="turtle")162 src = (163 PREFIX164 + """165 <http://example.com/publisher/1/catalogs/1> a dcat:Catalog ;166 dct:publisher <https://example.com/publishers/1> ;167 dct:title "Dataservicekatalog for Eksempel AS"@nb ;168 dcat:service <http://example.com/dataservices/1>,169 <http://example.com/dataservices/2>170 .171 """172 )173 g2 = Graph().parse(data=src, format="turtle")174 assert len(g1) == len(g2)175 _isomorphic = isomorphic(g1, g2)176 if not _isomorphic:177 _dump_diff(g1, g2)178 pass179 assert _isomorphic180def test_construct_query_with_SPARQLWrapper(http_service: Any) -> None:181 """Should return 200 status and the graph."""182 from SPARQLWrapper import SPARQLWrapper, TURTLE183 query_endpoint = f"{http_service}/{DATASET}/query"184 print(query_endpoint)185 querystring = """186 CONSTRUCT { ?s ?p ?o }187 WHERE {188 GRAPH <http://example.com/publisher/1> {?s ?p ?o}189 }190 """191 sparql = SPARQLWrapper(query_endpoint)192 sparql.setQuery(querystring)193 sparql.setReturnFormat(TURTLE)194 sparql.setOnlyConneg(True)195 results = sparql.query()196 assert 200 == results.response.status197 assert "text/turtle; charset=utf-8" == results.response.headers["Content-Type"]198 data = results.convert()199 g1 = Graph()200 g1.parse(data=data, format="turtle")201 g2 = Graph().parse("tests/catalog_1.ttl", format="turtle")202 assert len(g1) == len(g2)203 _isomorphic = isomorphic(g1, g2)204 if not _isomorphic:205 _dump_diff(g1, g2)206 pass207 assert _isomorphic208@pytest.mark.xfail209def test_add_graph_with_SPARQLUpdateStore(http_service: Any) -> None:210 """Should return some status and the graph is persisted."""211 from rdflib.plugins.stores import sparqlstore212 query_endpoint = f"{http_service}/{DATASET}/query"213 update_endpoint = f"{http_service}/{DATASET}/update"214 print("query_endpoint: ", query_endpoint)215 print("update_endpoint: ", update_endpoint)216 store = sparqlstore.SPARQLUpdateStore(auth=("admin", PASSWORD))217 store.open((query_endpoint, update_endpoint))218 g = Graph(identifier=URIRef("http://www.example.com/"))219 g.parse("tests/catalog_1.ttl", encoding="utf-8", format="turtle")220 result = store.add_graph(g)221 assert result222def test_construct_graph_of_all_catalogs_with_SPARQLWrapper(http_service: Any) -> None:223 """Should return all the graphs."""224 from SPARQLWrapper import SPARQLWrapper, TURTLE225 query_endpoint = f"{http_service}/{DATASET}/query"226 print(query_endpoint)227 querystring = """228 PREFIX dcat: <http://www.w3.org/ns/dcat#>229 PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>230 CONSTRUCT { ?s a dcat:Catalog .}231 WHERE { GRAPH ?g { ?s a dcat:Catalog .} }232 """233 sparql = SPARQLWrapper(query_endpoint)234 sparql.setQuery(querystring)235 sparql.setReturnFormat(TURTLE)236 sparql.setOnlyConneg(True)237 results = sparql.query()238 assert 200 == results.response.status239 assert "text/turtle; charset=utf-8" == results.response.headers["Content-Type"]240 data = results.convert()241 g1 = Graph()242 g1.parse(data=data, format="turtle")243 src = (244 PREFIX245 + """246 <http://example.com/publisher/1/catalogs/1>247 a dcat:Catalog .248 <http://example.com/publisher/2/catalogs/1>249 a dcat:Catalog .250 """251 )252 g2 = Graph().parse(data=src, format="turtle")253 assert len(g1) == len(g2)254 _isomorphic = isomorphic(g1, g2)255 if not _isomorphic:256 _dump_diff(g1, g2)257 pass258 assert _isomorphic259# ---------------------------------------------------------------------- #260# Utils for displaying debug information261def _dump_diff(g1: Graph, g2: Graph) -> None:262 in_both, in_first, in_second = graph_diff(g1, g2)263 print("\nin both:")264 _dump_turtle(in_both)265 print("\nin first:")266 _dump_turtle(in_first)267 print("\nin second:")268 _dump_turtle(in_second)269def _dump_turtle(g: Graph) -> None:270 for _l in g.serialize(format="turtle").splitlines():271 if _l:...
order_handler_strategy.py
Source:order_handler_strategy.py
1import abc2import logging3from django.conf import settings4from utils.request.request.request_interface import IRequest5class OrderHandlerStrategy(abc.ABC):6 def __init__(self, request_handler: IRequest) -> None:7 self._request = request_handler8 @abc.abstractmethod9 def place_order(self, request, *args, **kwargs):10 raise NotImplemented('Concrete implementation of method not found')11 @abc.abstractmethod12 def cancel_order(self, request, *args, **kwargs):13 raise NotImplemented('Concrete implementation of method not found')14class EagerOrderHandlerStrategy(OrderHandlerStrategy):15 def _create_endpoint(self, service, method):16 base_url = service['url']17 route = service['api'][method.upper()]18 return f'{base_url}{route}'19 def place_order(self, request, *args, **kwargs):20 data = request.data21 logging.info('#'*50)22 logging.info(f'Placing order with data {data}')23 payment_data = data.pop('payment')24 # 1. Create Order25 # print(type(data))26 order = self._request.post(url=self._create_endpoint(settings.ORDER_SERVICE, 'create'),27 json=data)28 logging.info(f'Created order {order}')29 # 2.1. Crete payment30 payment_data['order'] = order['id']31 payment_data['order_number'] = order['order_number']32 payment_data['customer'] = order['customer']33 payment = self._request.post(url=self._create_endpoint(settings.PAYMENT_SERVICE, 'create'),34 json=payment_data)35 logging.info(f'Created payment {payment}')36 # 2.2 Update order37 update_data = {38 'payment': payment['id']39 }40 update_endpoint = self._create_endpoint(settings.ORDER_SERVICE, 'update')41 update_endpoint = update_endpoint.format(id=order['id'])42 order = self._request.patch(url=update_endpoint, json=update_data)43 order['payment'] = payment44 logging.info(f'Updated order {order}')45 # 3. Update Inventory46 order_items = data.pop('order_items')47 for order_item in order_items:48 order_item = {49 'product': order_item['product'],50 'quantity': order_item['quantity'],51 'price': order_item['price']52 }53 update_endpoint = self._create_endpoint(settings.PRODUCT_SERVICE, 'update')54 update_endpoint = update_endpoint.format(id=order_item['product'])55 updated_data = {56 'count_in_stock': order_item['quantity']57 }58 self._request.patch(url=update_endpoint, data=updated_data)59 logging.info(f'Updated inventory')60 # 4. Clear Cart61 delete_endpoint = self._create_endpoint(settings.CART_SERVICE, 'delete')62 delete_endpoint = delete_endpoint.format(customer=data['customer'])63 self._request.delete(url=delete_endpoint)64 logging.info(f'Cleared cart')65 logging.info('#' * 50)66 # TODO: 5. Store in redis with callback with timout of 20 min to check if67 # order payment fails then cancel the order.68 return order69 def cancel_order(self, request, *args, **kwargs):70 data = request.data71 # 1. Update order status to Cancelled72 update_data = {73 'status': 'CL'74 }75 update_endpoint = self._create_endpoint(settings.ORDER_SERVICE, 'update')76 update_endpoint = update_endpoint.format(id=data['order'])77 order = self._request.patch(url=update_endpoint, json=update_data)78 logging.info(f'Order cancelled {order}')79 # 2. Update payment status to Refund80 update_data = {81 'status': 'RE'82 }83 update_endpoint = self._create_endpoint(settings.PAYMENT_SERVICE, 'update')84 update_endpoint = update_endpoint.format(id=order['payment'])85 payment = self._request.patch(url=update_endpoint, json=update_data)86 order['payment'] = payment87 logging.info(f'Payment updated {payment}')88 # 3. Update Inventory89 for order_item in order['order_items']:90 order_item = {91 'product': order_item['product'],92 'quantity': order_item['quantity'],93 'price': order_item['price']94 }95 update_endpoint = self._create_endpoint(settings.PRODUCT_SERVICE, 'update')96 update_endpoint = update_endpoint.format(id=order_item['product'])97 updated_data = {98 'count_in_stock': order_item['quantity']99 }100 self._request.patch(url=update_endpoint, data=updated_data)...
main.py
Source:main.py
1import requests2# NOTE : FILL THIS AREA WITH YOUR INFORMATIONS3USER_NAME = ""4TOKEN = ""5GRAPH_ID = ""6pixela_enpoint = "https://pixe.la/v1/users"7user_params = {8 "token": TOKEN,9 "username": USER_NAME,10 "agreeTermsOfService": "yes",11 "notMinor": "yes"12}13# NOTE: For creating user. MUST RUN FIRST TIME THAN COMMENT IT14resp = requests.post(url=pixela_enpoint, json=user_params)15print(resp.text)16graph_endpoint = f"{pixela_enpoint}/{USER_NAME}/graphs"17graph_config = {18 "id": GRAPH_ID,19 "name": "Game Play Time Graph",20 "unit": "hours",21 "type": "int",22 "color": "kuro"23}24headers = {25 "X-USER-TOKEN": TOKEN26}27# NOTE: FOR CREATING GRAPH RUN FIRST TIME AND THAN COMMENT IT28resp_second = requests.post(29 url=graph_endpoint, json=graph_config, headers=headers)30print(resp_second.text)31# NOTE: ADDING PIXEL TO GRAPH32pixel_endpoint = f"{graph_endpoint}/{GRAPH_ID}"33pixel_adding_config = {34 "date": "20220605",35 "quantity": "5"36}37resp_third = requests.post(38 url=pixel_endpoint, json=pixel_adding_config, headers=headers)39print(resp_third.text)40UPDATE_DATE = 2022060841update_endpoint = f"{pixel_endpoint}/{UPDATE_DATE}"42# NOTE: UPDATE PIXEL43new_piexldata = {44 "quantity": "3"45}46# RUN WHEN YOU NEED UPDATE47# resp_forth = requests.put(48# url=update_endpoint, json=new_piexldata, headers=headers)49# print(resp_forth.text)50# NOTE: DELETE PIXEL51# resp_fifth = requests.delete(52# url=update_endpoint, headers=headers)...
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?
