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

How to use _bytes_to_string method in tempest

Best Python code snippet using tempest_python

klaytn_token_service.py

Source:klaytn_token_service.py

...41            contract_alternative_1.functions.symbol(),42            contract_alternative_1.functions.SYMBOL(),43        )44        if isinstance(symbol, bytes):45            symbol = self._bytes_to_string(symbol)46        name = self._get_first_result(47            contract.functions.name(),48            contract.functions.NAME(),49            contract_alternative_1.functions.name(),50            contract_alternative_1.functions.NAME(),51        )52        if isinstance(name, bytes):53            name = self._bytes_to_string(name)54        try:55            decimals = self._get_first_result(56                contract.functions.decimals(), contract.functions.DECIMALS()57            )58        except ContractLogicError:59            decimals = None60        total_supply = self._get_first_result(contract.functions.totalSupply())61        return {62            'symbol': symbol,63            'name': name,64            'decimals': decimals,65            'total_supply': total_supply66        }67    def get_token(self, token_address) -> KlaytnToken:68        checksum_address = self._web3.toChecksumAddress(token_address)69        contract = self._web3.eth.contract(address=checksum_address, abi=ERC20_ABI)70        contract_alternative_1 = self._web3.eth.contract(71            address=checksum_address, abi=ERC20_ABI_ALTERNATIVE_172        )73        symbol = self._get_first_result(74            contract.functions.symbol(),75            contract.functions.SYMBOL(),76            contract_alternative_1.functions.symbol(),77            contract_alternative_1.functions.SYMBOL(),78        )79        if isinstance(symbol, bytes):80            symbol = self._bytes_to_string(symbol)81        name = self._get_first_result(82            contract.functions.name(),83            contract.functions.NAME(),84            contract_alternative_1.functions.name(),85            contract_alternative_1.functions.NAME(),86        )87        if isinstance(name, bytes):88            name = self._bytes_to_string(name)89        decimals = self._get_first_result(90            contract.functions.decimals(), contract.functions.DECIMALS()91        )92        total_supply = self._get_first_result(contract.functions.totalSupply())93        token = KlaytnToken()94        token.address = token_address95        token.symbol = symbol96        token.name = name97        token.decimals = decimals98        token.total_supply = total_supply99        return token100    def _get_first_result(self, *funcs):101        for func in funcs:102            result = self._call_contract_function(func)103            if result is not None:104                return result105        return None106    def _call_contract_function(self, func):107        # BadFunctionCallOutput exception happens if the token doesn't implement a particular function108        # or was self-destructed109        # OverflowError exception happens if the return type of the function doesn't match the expected type110        result = call_contract_function(111            func=func,112            ignore_errors=(BadFunctionCallOutput, OverflowError, ValueError, ContractLogicError),113            default_value=None,114        )115        if self._function_call_result_transformer is not None:116            return self._function_call_result_transformer(result)117        else:118            return result119    def _bytes_to_string(self, b, ignore_errors=True):120        if b is None:121            return b122        try:123            b = b.decode("utf-8")124        except UnicodeDecodeError as e:125            if ignore_errors:126                logger.debug(127                    "A UnicodeDecodeError exception occurred while trying to decode bytes to string",128                    exc_info=True,129                )130                b = None131            else:132                raise e133        if self._function_call_result_transformer is not None:...

cliente.py

Source:cliente.py

...35	def send(self, data):36		""" Sends the request """37		assert self.writer and self.reader, "no connection exists, make sure to call the connect() function before send()"38		self.writer.write(data.encode('utf-8'))39	def _bytes_to_string(self, data, headers):40		# list of data to try to decode41		to_decode = ["application", "text", "message"]	42		tmp_type = headers.get("content-type")43		if tmp_type:44			if "charset=" in tmp_type:45				for td in to_decode:46					if td in tmp_type:47						tmp = tmp_type.split("=")[1]48						try:49							return data.decode(tmp)50						except UnicodeDecodeError:51							# fallback encoding52							return data.decode("latin-1")53			else:54				to_decode = ["application", "text", "message"]	55				for td in to_decode:56					if td in tmp_type:57						#fallback encoding58						return data.decode("latin-1")59		return data60	async def receive(self):61		""" Reads data from the socket and handle headers and body retrieval """62		headers = {}63		is_first_line = True64		# valid response flag65		valid = True66		# Handle headers retrieval67		while True:68			line = await self.reader.readline()69			if line == b"\r\n":70				break71			else:72				if(is_first_line):73					# Manages first line of response74					# HTTP/{version} {status_code} {status}75					76					# Only 2 splits because some statuses will contain spaces (like "Bad Request")77					first_line = line.decode('latin1').split(" ", 2)78					if len(first_line) == 3:79						(self.http_version, self.status_code, self.status) = first_line80					# if first line is invalid break81					else:82						(self.http_version, self.status_code, self.status) = None, None, None83						valid = False84						break85					is_first_line = False86				else:87					# Extracts headers from the response88					# header: value89					# Only 1 split because some headers (like date) will contain the ":" char90					line = line.decode('latin1').split(":", 1)91					if len(line) == 2:92						(key, val) = line93						key = key.strip(" \r\n").lower()94						val = val.strip(" \r\n")95						headers[key] = val96		# Handle body retrieval97		if valid:98			length = headers.get("content-length")99		else:100			length = 0101		# Content length specified102		if length is not None:103			body = await self.reader.readexactly(int(length))104			d = decoder(headers)105			body = self._bytes_to_string(d.decode(body), headers)106		# Chunked transfer encoding107		else:108			d = decoder(headers)109			first = True110			while True:111				hex_len = self._bytes_to_string(d.decode(await self.reader.readline()), headers)112				length = int(hex_len, 16)113				114				if(length != 0):115					line = self._bytes_to_string(d.decode(await self.reader.readexactly(length)), headers)116					if first:117						body = line118						first = False119					else:120						body += line121					# ignore empty line122					_ = await self.reader.readline()123				else:124					#TODO manage headers sent after chunks125					break  126		self.headers = headers127		self.body = body...

encrypter_helper.py

Source:encrypter_helper.py

...61    def _print_do_key_defined(self) -> None:62        print("ERROR : No key defined")63    # key for encrypting/decrypting value64    def _generate_key(self) -> str:65        return self._bytes_to_string(Fernet.generate_key())66    def _string_to_bytes(self, st: str) -> bytes:67        return bytes(st, "ascii")68    def _bytes_to_string(self, b: bytes) -> str:69        return b.decode("ascii")70    # Encrypts string71    def _encrypt(self, to_be_encrypted: str, key: str) -> str:72        cipher_suite = Fernet(self._string_to_bytes(key))73        return self._bytes_to_string(74            cipher_suite.encrypt(self._string_to_bytes(to_be_encrypted))75        )76    # Decrypts string77    def _decrypt(self, encrypted: str, key: str) -> str:78        cipher_suite = Fernet(self._string_to_bytes(key))79        return self._bytes_to_string(80            cipher_suite.decrypt(self._string_to_bytes(encrypted))81        )82    def _serialize(self, key_to_be_saved: str) -> None:83        with open("encrypter.pickle", "wb") as handle:84            pickle.dump(key_to_be_saved, handle,85                        protocol=pickle.HIGHEST_PROTOCOL)86    def _deserialize(self) -> str:87        deserialized_key = None88        try:89            with open("encrypter.pickle", "rb") as handle:90                deserialized_key = pickle.load(handle)91        except:92            pass93        return deserialized_key

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.