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

How to use raise_not_found method in tempest

Best Python code snippet using tempest_python

main.py

Source:main.py

1import json2import requests3from typing import Optional, List, Dict, Union4from .exceptions import REServerError, RERequestError, RENotFound5_QUERY_METHOD = "POST"6_QUERY_ENDPOINT = "/api/v1/query_results"7_SAVE_METHOD = "PUT"8_SAVE_ENDPOINT = "/api/v1/documents"9class REClient:10    def __init__(self, api_url: str, token: str = None):11        self.api_url = api_url12        self.token = token13        # Type check the constructor parameters14        if not self.api_url or not isinstance(self.api_url, str):15            raise TypeError("The Relation Engine API URL was not provided.")16        # Remove any trailing slash in the API URL so we can append paths17        self.api_url = self.api_url.strip("/")18    def admin_query(self, query: str, bind_vars: dict, raise_not_found=False):19        """20        Run an ad-hoc query using admin privs.21        Params:22            query - string - AQL query to execute23            bind_vars - dict - JSON serializable bind variables for the query24            raise_not_found - bool - Whether to raise an error if there are zero results. Defaults to False25        Exceptions raised:26            RERequestError - 400-499 error from the RE API27            REServerError - 500+ error from the RE API28            RENotFound - raised when raise_not_found is True and there are 0 results29        """30        # Type-check the parameters31        if not isinstance(query, str):32            raise TypeError("`query` argument must be a str")33        if not isinstance(bind_vars, dict):34            raise TypeError("`bind_vars` argument must be a dict")35        if not isinstance(raise_not_found, bool):36            raise TypeError("`raise_not_found` argument must be a bool")37        # Construct and execute the request38        req_body = dict(bind_vars)39        req_body["query"] = query40        url = str(self.api_url) + _QUERY_ENDPOINT41        resp = self._make_request(42            method=_QUERY_METHOD,43            url=url,44            data=json.dumps(req_body),45            params={},46            raise_not_found=raise_not_found,47        )48        return resp49    def stored_query(self, stored_query: str, bind_vars: dict, raise_not_found=False):50        """51        Run a stored query.52        Params:53            stored_query - string - name of the stored query to execute54            bind_vars - JSON serializable - bind variables for the query (JSON serializable)55            raise_not_found - bool - Whether to raise an error if there are zero results. Defaults to False56        Exceptions raised:57            RERequestError - 400-499 from the RE API (client error)58            REServerError - 500+ error from the RE API59            RENotFound - raised when raise_not_found is True and there are 0 results60        """61        # Type-check the parameters62        if not isinstance(stored_query, str):63            raise TypeError("`stored_query` argument must be a str")64        if not isinstance(bind_vars, dict):65            raise TypeError("`bind_vars` argument must be a dict")66        if not isinstance(raise_not_found, bool):67            raise TypeError("`raise_not_found` argument must be a bool`")68        # Construct and execute the request69        req_body = dict(bind_vars)70        url = str(self.api_url) + _QUERY_ENDPOINT71        return self._make_request(72            method=_QUERY_METHOD,73            url=url,74            data=json.dumps(req_body),75            params={"stored_query": stored_query},76            raise_not_found=raise_not_found,77        )78    def save_docs(79        self,80        coll: str,81        docs: Union[Dict, List[Dict]],82        on_duplicate: Optional[str] = None,83        display_errors=False,84    ):85        """86        Save documents to a collection in the relation engine.87        Requires an auth token with RE admin privileges.88        Params:89            coll - str - collection name to save to90            docs - a single dict or list of dicts - json-serializable documents to save91            on_duplicate - str (defaults to 'error') - what to do when a provided document92                already exists in the collection. See options here:93                https://github.com/kbase/relation_engine_api#put-apiv1documents94            display_errors - bool (defaults to False) - whether to respond with95                document save errors (the response will give you an error for every96                document that failed to save).97        Exceptions raised:98            RERequestError - 400-499 from the RE API (client error)99            REServerError - 500+ error from the RE API100        """101        if isinstance(docs, dict):102            docs = [docs]103        if not docs:104            raise TypeError("No documents provided to save")105        if not isinstance(docs, list):106            raise TypeError("`docs` argument must be a list")107        if on_duplicate and not isinstance(on_duplicate, str):108            raise TypeError("`on_duplicate` argument must bea str")109        if not isinstance(display_errors, bool):110            raise TypeError("`display_errors` argument must be a bool")111        params = {"collection": coll}112        if display_errors:113            params["display_errors"] = "1"114        params["on_duplicate"] = on_duplicate or "error"115        req_body = "\n".join(json.dumps(d) for d in docs)116        url = str(self.api_url) + _SAVE_ENDPOINT117        return self._make_request(118            method=_SAVE_METHOD,119            url=url,120            data=req_body,121            params=params,122            raise_not_found=False,123        )124    def _make_request(self, method, url, data, params, raise_not_found):125        """126        Internal utility to make a generic request to the RE API and handle the127        response.128        """129        headers = {}130        if self.token:131            headers["Authorization"] = self.token132        resp = requests.request(133            method=method, url=url, data=data, params=params, headers=headers134        )135        if resp.status_code >= 500:136            # Server error137            raise REServerError(resp)138        elif resp.status_code >= 400 and resp.status_code < 500:139            # Client error140            raise RERequestError(resp)141        elif not resp.ok:142            raise RuntimeError(143                f"Unknown RE API error:\nURL: {resp.url}\nMethod: {method}\n{resp.text}"144            )145        resp_json = resp.json()146        if raise_not_found and not len(resp_json["results"]):147            # Results were required to be non-empty148            raise RENotFound(req_body=data, req_params=params)...

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.