KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _get_error_message method in localstack

Best Python code snippet using localstack_python

Dremio.py

Source:Dremio.py

...314				return response.json()315			elif response.status_code == 400:  # Bad Request316				if report_error:317					logging.info(source + ": received HTTP Response Code " + str(response.status_code) +318									" for : <" + str(url) + ">" + self._get_error_message(response))319			elif response.status_code == 404:  # Not found320				if report_error:321					logging.info(source + ": received HTTP Response Code " + str(response.status_code) +322									" for : <" + str(url) + ">" + self._get_error_message(response))323			else:324				if report_error:325					logging.error(source + ": received HTTP Response Code " + str(response.status_code) +326								" for : <" + str(url) + ">" + self._get_error_message(response))327					self.errors_encountered = self.errors_encountered + 1328			return None329		except requests.exceptions.Timeout:330			if source_name is None or source_name not in self._timed_out_sources:331				# This situation might happen when an underlying object (file system eg) is not responding332				if report_error:333					logging.error(source + ": HTTP Request Timed-out: " + " <" + str(url) + ">")334					self.errors_encountered = self.errors_encountered + 1335				else:336					logging.info(source + ": HTTP Request Timed-out: " + " <" + str(url) + ">")337			if source_name is not None and source_name not in self._timed_out_sources:338				self._timed_out_sources.append(source_name)339			return None340	# Returns JSON if success or None341	def _api_post_json(self, url, json_data, source="", as_json=True):342		try:343			if json_data is None:344				response = requests.request("POST", self._endpoint + url, headers=self._headers, timeout=self._api_timeout, verify=self._verify_ssl)345			elif as_json:346				response = requests.request("POST", self._endpoint + url, json=json_data, headers=self._headers, timeout=self._api_timeout, verify=self._verify_ssl)347			else:348				response = requests.request("POST", self._endpoint + url, data=json_data, headers=self._headers, timeout=self._api_timeout, verify=self._verify_ssl)349			if response.status_code == 200:350				return response.json()351			# Success, but no response352			elif response.status_code == 204:353				return None354			elif response.status_code == 400:355				logging.error(source + ": received HTTP Response Code " + str(response.status_code) +356							  " for : <" + str(url) + ">" + self._get_error_message(response))357			elif response.status_code == 403:  # User does not have permission358				logging.critical(source + ": received HTTP Response Code " + str(response.status_code) +359								 " for : <" + str(url) + ">" + self._get_error_message(response))360				raise RuntimeError(361					"Specified user does not have sufficient priviliges to create objects in the target Dremio Environment.")362			elif response.status_code == 409:  # Already exists.363				logging.error(source + ": received HTTP Response Code " + str(response.status_code) +364							  " for : <" + str(url) + ">" + self._get_error_message(response))365			elif response.status_code == 404:  # Not found366				logging.info(source + ": received HTTP Response Code " + str(response.status_code) +367							 " for : <" + str(url) + ">" + self._get_error_message(response))368			else:369				logging.error(source + ": received HTTP Response Code " + str(response.status_code) +370							  " for : <" + str(url) + ">" + self._get_error_message(response))371			self.errors_encountered = self.errors_encountered + 1372			return None373		except requests.exceptions.Timeout:374			# This situation might happen when an underlying object (file system eg) is not responding375			logging.error(source + ": HTTP Request Timed-out: " + " <" + str(url) + ">")376			self.errors_encountered = self.errors_encountered + 1377			return None378	# Returns JSON if success or None379	def _api_put_json(self, url, json_data, source="", report_error = True):380		try:381			response = requests.request("PUT", self._endpoint + url, json=json_data, headers=self._headers, timeout=self._api_timeout, verify=self._verify_ssl)382			if response.status_code == 200:383				return response.json()384			elif response.status_code == 400:  # The supplied CatalogEntity object is invalid.385				if report_error:386					logging.error(source + ": received HTTP Response Code 400 for : <" + str(url) + ">" +387							  self._get_error_message(response))388				else:389					logging.debug(source + ": received HTTP Response Code 400 for : <" + str(url) + ">" +390								  self._get_error_message(response))391			elif response.status_code == 403:  # User does not have permission to create the catalog entity.392				logging.critical(source + ": received HTTP Response Code 403 for : <" + str(url) + ">" +393								 self._get_error_message(response))394				raise RuntimeError(395					"Specified user does not have sufficient priviliges to create objects in the target Dremio Environment.")396			elif response.status_code == 409:  # A catalog entity with the specified path already exists.397				if report_error:398					logging.error(source + ": received HTTP Response Code 409 for : <" + str(url) + ">" +399							  self._get_error_message(response))400				else:401					logging.debug(source + ": received HTTP Response Code 409 for : <" + str(url) + ">" +402								  self._get_error_message(response))403			elif response.status_code == 404:  # Not found404				logging.info(source + ": received HTTP Response Code 404 for : <" + str(url) + ">" +405							 self._get_error_message(response))406			else:407				if report_error:408					logging.error(source + ": received HTTP Response Code " + str(response.status_code) +409							  " for : <" + str(url) + ">" + self._get_error_message(response))410				else:411					logging.debug(source + ": received HTTP Response Code " + str(response.status_code) +412								  " for : <" + str(url) + ">" + self._get_error_message(response))413			self.errors_encountered = self.errors_encountered + 1414			return None415		except requests.exceptions.Timeout:416			# This situation might happen when an underlying object (file system eg) is not responding417			logging.error(source + ": HTTP Request Timed-out: " + " <" + str(url) + ">")418			self.errors_encountered = self.errors_encountered + 1419			return None420	# Returns JSON if success or None421	def _api_delete(self, url, source="", report_error = True):422		try:423			response = requests.request("DELETE", self._endpoint + url, headers=self._headers, timeout=self._api_timeout, verify=self._verify_ssl)424			if response.status_code == 200:425				return response.json()426			elif response.status_code == 204:427				return None428			elif response.status_code == 400:  # The supplied CatalogEntity object is invalid.429				if report_error:430					logging.error(source + ": received HTTP Response Code 400 for : <" + str(url) + ">" +431								  self._get_error_message(response))432				else:433					logging.debug(source + ": received HTTP Response Code 400 for : <" + str(url) + ">" +434								  self._get_error_message(response))435			elif response.status_code == 403:  # User does not have permission to create the catalog entity.436				logging.critical(source + ": received HTTP Response Code 403 for : <" + str(url) + ">" +437								 self._get_error_message(response))438				raise RuntimeError(439					"Specified user does not have sufficient priviliges to create objects in the target Dremio Environment.")440			elif response.status_code == 409:  # A catalog entity with the specified path already exists.441				if report_error:442					logging.error(source + ": received HTTP Response Code 409 for : <" + str(url) + ">" +443								  self._get_error_message(response))444				else:445					logging.debug(source + ": received HTTP Response Code 409 for : <" + str(url) + ">" +446								  self._get_error_message(response))447			elif response.status_code == 404:  # Not found448				logging.info(source + ": received HTTP Response Code 404 for : <" + str(url) + ">" +449							 self._get_error_message(response))450			else:451				if report_error:452					logging.error(source + ": received HTTP Response Code " + str(response.status_code) +453								  " for : <" + str(url) + ">" + self._get_error_message(response))454				else:455					logging.debug(source + ": received HTTP Response Code " + str(response.status_code) +456								  " for : <" + str(url) + ">" + self._get_error_message(response))457			self.errors_encountered = self.errors_encountered + 1458			return None459		except requests.exceptions.Timeout:460			# This situation might happen when an underlying object (file system eg) is not responding461			logging.error(source + ": HTTP Request Timed-out: " + " <" + str(url) + ">")462			self.errors_encountered = self.errors_encountered + 1463			return None464	def _get_error_message(self, response):465		message = ""466		try:467			if 'errorMessage' in response.json():468				message = message + " errorMessage: " + str(response.json()['errorMessage'])469			if 'moreInfo' in response.json():470				message = message + " moreInfo: " + str(response.json()['moreInfo'])471		except:472			message = message + " content: " + str(response.content)473		return message474	def _encode_http_param(self, path):475		if sys.version_info.major > 2:476			return urllib.parse.quote_plus(path)477		else:478			return urllib.quote_plus(path)

_base_function.py

Source:_base_function.py

...43_ERROR_MSG = (44    "There is not implementation registered for {type}."45    "Please ensure you have loaded the corresponding module first."46)47def _get_error_message(*args):48    types = str(list(map(str, map(type, args))))49    return _ERROR_MSG.format(type=types)50@singledispatch51def apply_isometry_to_density_matrices(isometry, density_matrices):52    raise NotImplementedError(_get_error_message(isometry, density_matrices))53@singledispatch54def apply_kraus_ops_to_density_matrices(kraus_ops, density_matrices):55    raise NotImplementedError(_get_error_message(kraus_ops, density_matrices))56@singledispatch57def apply_unitary_transformation_to_density_matrices(unitary, density_matrices):58    raise NotImplementedError(_get_error_message(unitary, density_matrices))59@singledispatch60def commutator(a, b):61    raise NotImplementedError(_get_error_message(a, b))62@singledispatch63def dagger(mat):64    raise NotImplementedError(_get_error_message(mat))65@singledispatch66def distance(mat1, mat2):67    raise NotImplementedError(_get_error_message(mat1, mat2))68@singledispatch69def equiv(mat1, mat2, **allclose_args):70    raise NotImplementedError(_get_error_message(mat1, mat2))71@singledispatch72def format_wavefunction(wf, precision=8, skip_zeros=False, reverse_qubitstr=False):73    """Prints the components of the wavefunction visually.74    The argument reverse_qubitstr will reverse the sting representation of qubit's basis.75    For example, wf[1] will go from the |01> basis to the |10> basis.76    """77    raise NotImplementedError(_get_error_message(wf))78def get_random_ru(np, n=2):79    rr = np.random.rand(n, n)80    ri = np.random.rand(n, n)81    rc = rr + 1j * ri82    q, r = np.linalg.qr(rc)83    diag = np.diag84    r = diag(diag(r) / np.abs(diag(r)))85    return q.dot(r)86def get_random_wf(np, nqb, nwf=1):87    """Get a random wavefunction.88    Random according to Haar measure89    """90    if nwf == 1:91        wf = _get_random_wf_1(np, nqb)92    else:93        wf = np.empty(shape=(2 ** nqb, nwf), dtype=np.complex128)94        for wfidx in range(nwf):95            wf[:, wfidx] = _get_random_wf_1(np, nqb)96    return wf97def _get_random_wf_1(np, nqb):98    u = get_randu(np, nqb)99    p = u.dot(np.ones(shape=(2 ** nqb, 1)))100    p /= np.linalg.norm(p, ord=2)101    p = np.squeeze(p)102    return p103def get_randu(np, nqb):104    """105    Generates a n qubit random unitary matrix, distributed uniformly106    according to the Haar measure.107    References108    -----------109    <http://www.dr-qubit.org/matlab/randU.m>110    """111    randn = np.random.randn112    diag = np.diag113    dim = 2 ** nqb114    x = randn(dim, dim) + 1j * randn(dim, dim)115    x /= np.sqrt(2)116    q, r = np.linalg.qr(x, mode="complete")117    diag_r = diag(r)118    r = diag(diag_r / np.abs(diag_r))119    return q.dot(r)120def rand_rho(np, nqb=None, dim=None):121    p = 10 * rand_herm(np, nqb, dim)122    ppp = p.dot(np.conj(np.transpose(p)))123    return ppp / np.trace(ppp)124def rand_herm(np, nqb=None, dim=None):125    """126    Generates a random Hermitian matrix127    References128    ------------129    http://www.dr-qubit.org/matlab/randH.m130    """131    if nqb is None:132        if not isinstance(dim, int):133            raise ValueError(f"Wrong parameters: nqb={nqb}, dim={dim}")134    else:135        dim = 2 ** nqb136    randn = np.random.randn137    h = 2 * (randn(dim, dim) + 1j * randn(dim, dim)) - (1 + 1j)138    return h + np.conj(np.transpose(h))139def get_rho_from_random_wf(np, nqb):140    """Get a random density matrix rho, by generating a random wavefunction141    and return its density matrix."""142    wf = _get_random_wf_1(np, nqb)143    wf = wf.reshape(2 ** nqb, 1)144    return np.kron(wf, wf.transpose().conj())145@singledispatch146def kron_each(wf1s, wf2s):147    raise NotImplementedError(_get_error_message(wf1s, wf2s))148@singledispatch149def load1qb_into_ithqb(iwf_1qb, target_qbidx: int, numphyqb: int):150    raise NotImplementedError(_get_error_message(iwf_1qb, target_qbidx, numphyqb))151@singledispatch152def load_state_into_mqb_start_from_lqb(states, m, l: int = 0):153    """154    Loads states (columns of wavefunctions) from the smaller, and initialise the new155    qubits to 0.156    Note: l is 0-based. m is 1-based. For example, you can load state into 1 qb starts157    from 0 qubit.158    """159    raise NotImplementedError(_get_error_message(states, m, l))160def load_states_into(states, total_qb: int, pos: List[int]):161    """162    Loads the input `states` into the positions (`pos`), within a total163    qubits by `total_qb`.164    """165    assert total_qb > 0, "Illegal total_qb."166    assert all(0 <= _i < total_qb for _i in pos), "pos outside total_qb"167    assert 2 ** len(pos) == states.shape[0], "Input wf dim >= len(pos)"168    return _load_states_into(states, total_qb, pos)169@singledispatch170def _load_states_into(states, total_qb: int, pos: List[int]):171    raise NotImplementedError(_get_error_message(states, total_qb, pos))172@singledispatch173def make_density_matrix(wf):174    raise NotImplementedError(_get_error_message(wf))175@singledispatch176def partial_trace(rho, retain_qubits: Iterable[int]):177    """178    Compute the partial trace of rho.179    Parameters180    ----------181    rho182        input rho183    retain_qubits184        the qubits which we want to keep after partial trace.185    """186    raise NotImplementedError(_get_error_message(rho, retain_qubits))187@singledispatch188def partial_trace_1d(rho, retain_qubit: int):189    raise NotImplementedError(_get_error_message(rho, retain_qubit))190@singledispatch191def partial_trace_wf(iwf, retain_qubits: Iterable[int]):192    """Partial trace on wavefunctions193    Parameters194    ----------195    iwf196    retain_qubits197        the qubits which we want to keep after partial trace.198    """199    raise NotImplementedError(_get_error_message(iwf, retain_qubits))200@singledispatch201def partial_trace_wf_keep_first(iwf, n: int):202    """Partial trace which keeps only the first n qubits203    Parameters204    ----------205    iwf206    n : int207        A non-negative integer. When n = 0, this function simply returns a208        shape (1,1) array of value 1.209    Notes210    --------211    We keep the least significant bit on the left. That is, inside the bit212    string of :math:`n_0n_1\\cdots n_m`, :math:`n_0` is the first qubit.213    In this way, for a iwf of indices214    :math:`i_1\\cdots i_n j_{n+1}j_{n+2}\\cdots `, the :math:`j` indices215    needs to be summed, which should be close together on the memory since we216    assume that iwf is c-ordered (although the code would work even if iwf is217    f-ordered)218    """219    raise NotImplementedError(_get_error_message(iwf, n))220@singledispatch221def pure_state_overlap(wf1, wf2):222    raise NotImplementedError(_get_error_message(wf1, wf2))223@singledispatch224def trace_distance(target_rho, pred_rho):225    raise NotImplementedError(_get_error_message(target_rho, pred_rho))226@singledispatch227def trace_distance_1qb(target_rho, pred_rho):228    raise NotImplementedError(_get_error_message(target_rho, pred_rho))229@singledispatch230def trace_distance_using_svd(target_rho, pred_rho):...

http.py

Source:http.py

1from typing import Optional, Tuple, ClassVar, Union, Dict, Any2from aiohttp import ClientSession, ClientResponse3from .errors import *4async def _get_error_message(response: ClientResponse) -> str:5    if response.content_type == "application/json":6        return str((await response.json()).get("description", None))7    return str(await response.text())8class HTTPClient:9    BASE_URL: ClassVar[str] = "https://api.alexflipnote.dev"10    BASE_URL_COFFEE: ClassVar[str] = "https://coffee.alexflipnote.dev"11    __slots__: Tuple[str, ...] = ("__session",)12    def __init__(self, session: Optional[ClientSession] = None) -> None:13        self.__session = session14    # Aiohttp client sessions must be created in async functions15    async def create_session(self) -> None:16        self.__session = ClientSession()17    async def query(self, url: str, method: Optional[str] = "GET", **kwargs) -> ClientResponse:18        if self.__session is None:19            await self.create_session()20        return await self.__session.request(method, url, **kwargs)  # type: ignore21    async def __call__(self, endpoint: str) -> Union[Dict[str, Any], ClientResponse]:22        if endpoint.startswith("coffee"):23            url = f"{self.BASE_URL_COFFEE}/random.json"24        elif endpoint.startswith("support"):25            url = self.BASE_URL26        else:27            url = f"{self.BASE_URL}/{endpoint}"28        response = await self.query(str(url))29        if response.status == 200:30            if response.content_type == "application/json":31                return await response.json()32            return response33        elif response.status == 400:34            raise BadRequest(await _get_error_message(response))35        elif response.status == 403:36            raise Forbidden(await _get_error_message(response))37        elif response.status == 404:38            raise NotFound(await _get_error_message(response))39        elif response.status == 500:40            raise InternalServerError(await _get_error_message(response))41        else:42            raise HTTPException(response, await _get_error_message(response))43    async def close(self) -> None:44        if self.__session is not None and not self.__session.closed:45            await self.__session.close()...

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.