Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.