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

How to use execute_rpc method in autotest

Best Python code snippet using autotest_python

node.py

Source:node.py

...58        if self._rpc_connection is not None:59            self._rpc_connection.__dict__['_BaseProxy__conn'].close()60            logging.debug('Closed rpc')61    def stop(self):62        self.execute_rpc('stop')63        logging.info('stopping node {}'.format(self.name))64    def get_log_file(self):65        return self._path + config.bitcoin_log_file_name66    def wait_until_rpc_ready(self):67        while True:68            try:69                bash.check_output(70                    "nc -z -w1 {} {}"71                    .format(self._ip, config.rpc_port)72                )73                break74            except Exception:75                logging.debug("Port not open")76        while True:77            try:78                self.execute_rpc('getnetworkinfo')79                break80            except JSONRPCError:81                logging.debug('RPC not ready yet, sleeping for 2')82                utils.sleep(2)83    def connect_to_rpc(self):84        self._rpc_connection = Proxy(85            btc_conf_file=config.btc_conf_file.format(self.name),86            timeout=config.rpc_timeout87        )88    def rm_peers_file(self):89        return bash.check_output(bitcoincmd.rm_peers(self._name))90    def execute_rpc(self, *args):91        # No clue whats the optimal92        retry = 1093        while retry > 0:94            try:95                return self._rpc_connection.call(args[0], *args[1:])96            except (IOError, CannotSendRequest) as error:97                logging.exception('Error: {}.'98                                  ' Reconnecting and retrying, {} more tries left'99                                  .format(error, retry))100                retry -= 1101                self.connect_to_rpc()102    def transfer_coinbases_to_normal_tx(self):103        # Readable format convert104        for tx_chain in self._tx_chains:105            tx_chain.amount /= 2106            tx_chain.amount -= int(config.transaction_fee / 2)107            raw_transaction = self.execute_rpc(108                'createrawtransaction',109                [{110                    'txid': tx_chain.current_unspent_tx,111                    'vout': 0,112                }],113                OrderedDict([114                    (tx_chain.address, str(tx_chain.amount / 100000000)),115                    (self._spent_to.address, str(tx_chain.amount / 100000000))116                ])117            )118            signed_raw_transaction = self.execute_rpc(119                'signrawtransaction', raw_transaction120            )['hex']121            tx_chain.current_unspent_tx = self.execute_rpc(122                'sendrawtransaction',123                signed_raw_transaction124            )125    def generate_blocks(self, amount=1):126        logging.debug('{} trying to generate mine'.format(self._name))127        block_hash = self.execute_rpc('generate', amount)128        logging.info('{} mined a block {}'.format(self._name, block_hash))129    def generate_tx(self):130        # Normal transaction signing and raw processing131        tx_chain = self.get_next_tx_chain()132        txid = lx(tx_chain.current_unspent_tx)133        txins = [134            CMutableTxIn(COutPoint(txid, 0)),135            CMutableTxIn(COutPoint(txid, 1))136        ]137        txin_seckeys = [tx_chain.seckey, self._spent_to.seckey]138        amount_in = tx_chain.amount139        tx_chain.amount -= int(config.transaction_fee / 2)140        txout1 = CMutableTxOut(141            tx_chain.amount,142            CBitcoinAddress(tx_chain.address).to_scriptPubKey()143        )144        txout2 = CMutableTxOut(145            tx_chain.amount,146            CBitcoinAddress(self._spent_to.address).to_scriptPubKey()147        )148        tx = CMutableTransaction(txins, [txout1, txout2], nVersion=2)149        for i, txin in enumerate(txins):150            txin_scriptPubKey = CScript([151                OP_DUP,152                OP_HASH160,153                Hash160(txin_seckeys[i].pub),154                OP_EQUALVERIFY,155                OP_CHECKSIG156            ])157            sighash = SignatureHash(txin_scriptPubKey, tx, i, SIGHASH_ALL)158            sig = txin_seckeys[i].sign(sighash) + bytes([SIGHASH_ALL])159            txin.scriptSig = CScript([sig, txin_seckeys[i].pub])160        tx_serialized = tx.serialize()161        logging.debug(162            '{} is sending raw transaction'163            ' (input {} output {} fee {} bytes {})'164            ' using tx_chain id {}'165            .format(self._name,166                    amount_in,167                    txout1.nValue,168                    (amount_in * 2) - (txout1.nValue * 2),169                    len(tx_serialized),170                    self._current_tx_chain_index)171        )172        tx_hash = self.execute_rpc('sendrawtransaction', b2x(tx_serialized))173        tx_chain.current_unspent_tx = tx_hash174        #print(tx_chain.current_unspent_tx)175        logging.info(176            '{} send was ok; hash={}'177            .format(self._name, tx_hash)178        )179    def generate_spent_to_address(self):180        # Get from wallet the list of address or generate one181        address = self.execute_rpc('getnewaddress')182        seckey = CBitcoinSecret(self.execute_rpc('dumpprivkey', address))183        self._spent_to = SpentToAddress(address, seckey)184    def create_tx_chains(self):185        for unspent_tx in self.execute_rpc('listunspent'):186            seckey = CBitcoinSecret(187                self.execute_rpc('dumpprivkey', unspent_tx['address'])188            )189            tx_chain = TxChain(190                unspent_tx['txid'],191                unspent_tx['address'],192                seckey,193                unspent_tx['amount'] * 100000000194            )195            self._tx_chains.append(tx_chain)196    def get_next_tx_chain(self):197        tx_chain = self._tx_chains[self._current_tx_chain_index]198        self._current_tx_chain_index = (199            (self._current_tx_chain_index + 1) %200            len(self._tx_chains)201        )202        return tx_chain203class PublicBitcoinNode(BitcoinNode):204    __slots__ = ['_latency', '_outgoing_ips']205    def __init__(self, name, group, ip, latency, path):206        BitcoinNode.__init__(self, name, group, ip, path)207        self._latency = latency208        self._outgoing_ips = []209    def set_outgoing_ips(self, outgoing_ips):210        #peer211        self._outgoing_ips = outgoing_ips212    def calcu_latency(self, zones):213        bash.check_output(cmd)214    def run(self, connect_to_ips=None):215        if connect_to_ips is None:216            connect_to_ips = self._outgoing_ips217        super(PublicBitcoinNode, self).run(connect_to_ips)218class TxChain:219    __slots__ = ['_current_unspent_tx', '_address', '_seckey', '_amount']220    def __init__(self, current_unspent_tx, address, seckey, amount):221        self._current_unspent_tx = current_unspent_tx222        self._address = address223        self._seckey = seckey224        self._amount = amount225    @property226    def current_unspent_tx(self):227        return self._current_unspent_tx228    @current_unspent_tx.setter229    def current_unspent_tx(self, unspent_tx):230        self._current_unspent_tx = unspent_tx231    @property232    def address(self):233        return self._address234    @property235    def seckey(self):236        # Incase we need to manually decrypt?237        #store_key = self._seckey238        #print(store_key)239        return self._seckey240    @property241    def amount(self):242        return self._amount243    @amount.setter244    def amount(self, amount):245        self._amount = amount246SpentToAddress = namedtuple('SpentToAddress', 'address seckey')247def create_conf_file(node):248    # We generate and push the bit config file249    node.create_conf_file()250def start_node(node, connect_to_ips=None):251    node.run(connect_to_ips)252def check_startup_node(node, height=0):253    node.connect_to_rpc()254    node.wait_until_rpc_ready()255    wait_until_height_reached(node, height) # for genesis256def wait_until_height_reached(node, height):257    while True:258        node_height = node.execute_rpc('getblockcount')259        if height <= int(node_height):260            break261        logging.debug('Waiting until node={} with current height={} reached height={}...'262                      .format(node.name, node_height, height))263        utils.sleep(1)264def transfer_coinbase_tx_to_normal_tx(node):265    node.generate_spent_to_address()266    node.create_tx_chains()267    node.transfer_coinbases_to_normal_tx()268    logging.info("Transferred all coinbase-tx to normal tx")269def cal_latency(node, zones):270    node.cal_latency(zones)271def wait_until_node_stopped(node):272    parts = 10...

ipc.py

Source:ipc.py

...49        if not request:50            logger.debug("Client sent empty request")51            continue52        try:53            result = execute_rpc(request)54        except Exception as e:55            logger.exception("Unrecognized exception while executing RPC")56            await write_error(writer, "unknown failure: " + str(e))57        else:58            writer.write(result.encode())59        await writer.drain()60def strip_non_json_prefix(raw_request):61    if raw_request and raw_request[0] != '{':62        prefix, bracket, rest = raw_request.partition('{')63        return prefix.strip(), bracket + rest64    else:65        return '', raw_request66async def write_error(writer, message):67    json_error = json.dumps({'error': message}) + '\n'...

rpc_handler.py

Source:rpc_handler.py

...25    execute_rpc: Callable[[Any], Any],26    json_request: Dict['str', Any]27) -> str:28    try:29        result = await execute_rpc(json_request)30    except Exception as e:31        msg = f"Unrecognized exception while executing RPC: {e}"32        raise JsonRpcCallException(msg)33    else:34        return result35class RPCHandler(BaseHTTPHandler):36    @staticmethod37    @curry38    async def handle(execute_rpc: Callable[[Any], Any], request: web.Request) -> web.Response:39        logger = logging.getLogger('trinity.http.handlers.rpc_handler')40        if request.method == 'POST':41            logger.debug('Receiving POST request: %s', request.path)42            try:43                body_json = await load_json_request(request)...

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.