Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/test_node.py
| Show First 20 Lines • Show All 733 Lines • ▼ Show 20 Lines | class TestNodeCLIAttr: | ||||
| def get_request(self, *args, **kwargs): | def get_request(self, *args, **kwargs): | ||||
| return lambda: self(*args, **kwargs) | return lambda: self(*args, **kwargs) | ||||
| def arg_to_cli(arg): | def arg_to_cli(arg): | ||||
| if isinstance(arg, bool): | if isinstance(arg, bool): | ||||
| return str(arg).lower() | return str(arg).lower() | ||||
| elif arg is None: | |||||
| return 'null' | |||||
| elif isinstance(arg, dict) or isinstance(arg, list): | elif isinstance(arg, dict) or isinstance(arg, list): | ||||
| return json.dumps(arg, default=EncodeDecimal) | return json.dumps(arg, default=EncodeDecimal) | ||||
| else: | else: | ||||
| return str(arg) | return str(arg) | ||||
| class TestNodeCLI(): | class TestNodeCLI(): | ||||
| """Interface to bitcoin-cli for an individual node""" | """Interface to bitcoin-cli for an individual node""" | ||||
| ▲ Show 20 Lines • Show All 70 Lines • ▼ Show 20 Lines | def __init__(self, rpc, cli=False, descriptors=False): | ||||
| # For now, descriptors is always False | # For now, descriptors is always False | ||||
| self.descriptors = descriptors | self.descriptors = descriptors | ||||
| def __getattr__(self, name): | def __getattr__(self, name): | ||||
| return getattr(self.rpc, name) | return getattr(self.rpc, name) | ||||
| def createwallet(self, wallet_name, disable_private_keys=None, blank=None, | def createwallet(self, wallet_name, disable_private_keys=None, blank=None, | ||||
| passphrase='', avoid_reuse=None, descriptors=None, load_on_startup=None): | passphrase='', avoid_reuse=None, descriptors=None, load_on_startup=None): | ||||
| if self.is_cli: | |||||
| if disable_private_keys is None: | |||||
| disable_private_keys = 'null' | |||||
| if blank is None: | |||||
| blank = 'null' | |||||
| if passphrase is None: | |||||
| passphrase = '' | |||||
| if avoid_reuse is None: | |||||
| avoid_reuse = 'null' | |||||
| if load_on_startup is None: | |||||
| load_on_startup = 'null' | |||||
| if descriptors is None: | if descriptors is None: | ||||
| descriptors = self.descriptors | descriptors = self.descriptors | ||||
| return self.__getattr__('createwallet')( | return self.__getattr__('createwallet')( | ||||
| wallet_name, disable_private_keys, blank, passphrase, avoid_reuse, descriptors, load_on_startup) | wallet_name, disable_private_keys, blank, passphrase, avoid_reuse, descriptors, load_on_startup) | ||||
| def importprivkey(self, privkey, label=None, rescan=None): | def importprivkey(self, privkey, label=None, rescan=None): | ||||
| wallet_info = self.getwalletinfo() | wallet_info = self.getwalletinfo() | ||||
| if self.is_cli: | |||||
| if label is None: | |||||
| label = 'null' | |||||
| if rescan is None: | |||||
| rescan = 'null' | |||||
| if 'descriptors' not in wallet_info or ( | if 'descriptors' not in wallet_info or ( | ||||
| 'descriptors' in wallet_info and not wallet_info['descriptors']): | 'descriptors' in wallet_info and not wallet_info['descriptors']): | ||||
| return self.__getattr__('importprivkey')(privkey, label, rescan) | return self.__getattr__('importprivkey')(privkey, label, rescan) | ||||
| desc = descsum_create('combo(' + privkey + ')') | desc = descsum_create('combo(' + privkey + ')') | ||||
| req = [{ | req = [{ | ||||
| 'desc': desc, | 'desc': desc, | ||||
| 'timestamp': 0 if rescan else 'now', | 'timestamp': 0 if rescan else 'now', | ||||
| 'label': label if label else '' | 'label': label if label else '' | ||||
| }] | }] | ||||
| import_res = self.importdescriptors(req) | import_res = self.importdescriptors(req) | ||||
| if not import_res[0]['success']: | if not import_res[0]['success']: | ||||
| raise JSONRPCException(import_res[0]['error']) | raise JSONRPCException(import_res[0]['error']) | ||||
| def addmultisigaddress(self, nrequired, keys, | def addmultisigaddress(self, nrequired, keys, | ||||
| label=None): | label=None): | ||||
| wallet_info = self.getwalletinfo() | wallet_info = self.getwalletinfo() | ||||
| if self.is_cli: | |||||
| if label is None: | |||||
| label = 'null' | |||||
| if 'descriptors' not in wallet_info or ( | if 'descriptors' not in wallet_info or ( | ||||
| 'descriptors' in wallet_info and not wallet_info['descriptors']): | 'descriptors' in wallet_info and not wallet_info['descriptors']): | ||||
| return self.__getattr__('addmultisigaddress')( | return self.__getattr__('addmultisigaddress')( | ||||
| nrequired, keys, label) | nrequired, keys, label) | ||||
| cms = self.createmultisig(nrequired, keys) | cms = self.createmultisig(nrequired, keys) | ||||
| req = [{ | req = [{ | ||||
| 'desc': cms['descriptor'], | 'desc': cms['descriptor'], | ||||
| 'timestamp': 0, | 'timestamp': 0, | ||||
| 'label': label if label else '' | 'label': label if label else '' | ||||
| }] | }] | ||||
| import_res = self.importdescriptors(req) | import_res = self.importdescriptors(req) | ||||
| if not import_res[0]['success']: | if not import_res[0]['success']: | ||||
| raise JSONRPCException(import_res[0]['error']) | raise JSONRPCException(import_res[0]['error']) | ||||
| return cms | return cms | ||||
| def importpubkey(self, pubkey, label=None, rescan=None): | def importpubkey(self, pubkey, label=None, rescan=None): | ||||
| wallet_info = self.getwalletinfo() | wallet_info = self.getwalletinfo() | ||||
| if self.is_cli: | |||||
| if label is None: | |||||
| label = 'null' | |||||
| if rescan is None: | |||||
| rescan = 'null' | |||||
| if 'descriptors' not in wallet_info or ( | if 'descriptors' not in wallet_info or ( | ||||
| 'descriptors' in wallet_info and not wallet_info['descriptors']): | 'descriptors' in wallet_info and not wallet_info['descriptors']): | ||||
| return self.__getattr__('importpubkey')(pubkey, label, rescan) | return self.__getattr__('importpubkey')(pubkey, label, rescan) | ||||
| desc = descsum_create('combo(' + pubkey + ')') | desc = descsum_create('combo(' + pubkey + ')') | ||||
| req = [{ | req = [{ | ||||
| 'desc': desc, | 'desc': desc, | ||||
| 'timestamp': 0 if rescan else 'now', | 'timestamp': 0 if rescan else 'now', | ||||
| 'label': label if label else '' | 'label': label if label else '' | ||||
| }] | }] | ||||
| import_res = self.importdescriptors(req) | import_res = self.importdescriptors(req) | ||||
| if not import_res[0]['success']: | if not import_res[0]['success']: | ||||
| raise JSONRPCException(import_res[0]['error']) | raise JSONRPCException(import_res[0]['error']) | ||||
| def importaddress(self, address, label=None, rescan=None, p2sh=None): | def importaddress(self, address, label=None, rescan=None, p2sh=None): | ||||
| wallet_info = self.getwalletinfo() | wallet_info = self.getwalletinfo() | ||||
| if self.is_cli: | |||||
| if label is None: | |||||
| label = 'null' | |||||
| if rescan is None: | |||||
| rescan = 'null' | |||||
| if p2sh is None: | |||||
| p2sh = 'null' | |||||
| if 'descriptors' not in wallet_info or ( | if 'descriptors' not in wallet_info or ( | ||||
| 'descriptors' in wallet_info and not wallet_info['descriptors']): | 'descriptors' in wallet_info and not wallet_info['descriptors']): | ||||
| return self.__getattr__('importaddress')( | return self.__getattr__('importaddress')( | ||||
| address, label, rescan, p2sh) | address, label, rescan, p2sh) | ||||
| is_hex = False | is_hex = False | ||||
| try: | try: | ||||
| int(address, 16) | int(address, 16) | ||||
| is_hex = True | is_hex = True | ||||
| Show All 18 Lines | |||||