diff --git a/bittensor_cli/cli.py b/bittensor_cli/cli.py index c407e5b9b..be88a082f 100755 --- a/bittensor_cli/cli.py +++ b/bittensor_cli/cli.py @@ -1236,6 +1236,12 @@ def __init__(self): "execute", rich_help_panel=HELP_PANELS["PROXY"]["MGMT"], )(self.proxy_execute_announced) + self.proxy_app.command("list", rich_help_panel=HELP_PANELS["PROXY"]["MGMT"])( + self.proxy_list + ) + self.proxy_app.command("reject", rich_help_panel=HELP_PANELS["PROXY"]["MGMT"])( + self.proxy_reject + ) # Sub command aliases # Wallet @@ -9335,13 +9341,19 @@ def proxy_add( def proxy_remove( self, delegate: Annotated[ - str, + Optional[str], typer.Option( callback=is_valid_ss58_address_param, - prompt="Enter the SS58 address of the delegate to remove, e.g. 5dxds...", help="The SS58 address of the delegate to remove", ), - ] = "", + ] = None, + all_proxies: Annotated[ + bool, + typer.Option( + "--all", + help="Remove all proxies for the account", + ), + ] = False, network: Optional[list[str]] = Options.network, proxy_type: ProxyType = Options.proxy_type, delay: int = typer.Option(0, help="Delay, in number of blocks"), @@ -9373,10 +9385,10 @@ def proxy_remove( [green]$[/green] btcli proxy remove --all """ - # TODO should add a --all flag to call Proxy.remove_proxies ? logger.debug( "args:\n" f"delegate: {delegate}\n" + f"all_proxies: {all_proxies}\n" f"network: {network}\n" f"proxy_type: {proxy_type}\n" f"delay: {delay}\n" @@ -9385,6 +9397,31 @@ def proxy_remove( f"era: {period}\n" ) self.verbosity_handler(quiet, verbose, json_output, prompt) + + # Validate that either --all or --delegate is provided, but not both + if all_proxies and delegate: + err_console.print( + ":cross_mark:[red]Cannot use both --all and --delegate. " + "Use --all to remove all proxies or --delegate to remove a specific proxy.[/red]" + ) + raise typer.Exit(1) + + if not all_proxies and not delegate: + if prompt: + delegate = Prompt.ask( + "Enter the SS58 address of the delegate to remove, e.g. 5dxds..." + ) + if not is_valid_ss58_address(delegate): + err_console.print( + f":cross_mark:[red]Invalid SS58 address: {delegate}[/red]" + ) + raise typer.Exit(1) + else: + err_console.print( + ":cross_mark:[red]Either --delegate or --all must be specified.[/red]" + ) + raise typer.Exit(1) + wallet = self.wallet_ask( wallet_name=wallet_name, wallet_path=wallet_path, @@ -9392,22 +9429,38 @@ def proxy_remove( ask_for=[WO.NAME, WO.PATH], validate=WV.WALLET, ) - return self._run_command( - proxy_commands.remove_proxy( - subtensor=self.initialize_chain(network), - wallet=wallet, - delegate=delegate, - proxy_type=proxy_type, - delay=delay, - prompt=prompt, - decline=decline, - quiet=quiet, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - period=period, - json_output=json_output, + + if all_proxies: + return self._run_command( + proxy_commands.remove_all_proxies( + subtensor=self.initialize_chain(network), + wallet=wallet, + prompt=prompt, + decline=decline, + quiet=quiet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + period=period, + json_output=json_output, + ) + ) + else: + return self._run_command( + proxy_commands.remove_proxy( + subtensor=self.initialize_chain(network), + wallet=wallet, + proxy_type=proxy_type, + delegate=delegate, + delay=delay, + prompt=prompt, + decline=decline, + quiet=quiet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + period=period, + json_output=json_output, + ) ) - ) def proxy_kill( self, @@ -9695,6 +9748,266 @@ def proxy_execute_announced( with ProxyAnnouncements.get_db() as (conn, cursor): ProxyAnnouncements.mark_as_executed(conn, cursor, got_call_from_db) + def proxy_list( + self, + address: Annotated[ + Optional[str], + typer.Option( + callback=is_valid_ss58_address_param, + help="The SS58 address to list proxies for. If not provided, uses the wallet's coldkey.", + ), + ] = None, + network: Optional[list[str]] = Options.network, + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + wallet_hotkey: str = Options.wallet_hotkey, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + json_output: bool = Options.json_output, + ): + """ + Lists all proxies for an account. + + Queries the chain to display all proxy delegates configured for the specified address, + including their proxy types and delay settings. + + [bold]Common Examples:[/bold] + 1. List proxies for your wallet + [green]$[/green] btcli proxy list + + 2. List proxies for a specific address + [green]$[/green] btcli proxy list --address 5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY + + """ + self.verbosity_handler(quiet, verbose, json_output, prompt=False) + + # If no address provided, use wallet's coldkey + if address is None: + wallet = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.PATH], + validate=WV.NONE, + ) + address = wallet.coldkeypub.ss58_address + + logger.debug(f"args:\naddress: {address}\nnetwork: {network}\n") + + return self._run_command( + proxy_commands.list_proxies( + subtensor=self.initialize_chain(network), + address=address, + json_output=json_output, + ) + ) + + def proxy_reject( + self, + delegate: Annotated[ + Optional[str], + typer.Option( + callback=is_valid_ss58_address_param, + help="The SS58 address of the delegate who made the announcement. If omitted, the wallet's coldkey ss58 is used.", + ), + ] = None, + call_hash: Annotated[ + Optional[str], + typer.Option( + help="The hash of the announced call to reject", + ), + ] = None, + network: Optional[list[str]] = Options.network, + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + wallet_hotkey: str = Options.wallet_hotkey, + prompt: bool = Options.prompt, + decline: bool = Options.decline, + wait_for_inclusion: bool = Options.wait_for_inclusion, + wait_for_finalization: bool = Options.wait_for_finalization, + period: int = Options.period, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + json_output: bool = Options.json_output, + ): + """ + Rejects an announced proxy call. + + Removes a previously announced call from the pending announcements, preventing it + from being executed. This must be called by the real account (the account that + granted the proxy permissions). + + [bold]Common Examples:[/bold] + 1. Reject an announced call + [green]$[/green] btcli proxy reject --delegate 5GDel... --call-hash 0x1234... + + """ + self.verbosity_handler(quiet, verbose, json_output, prompt, decline) + + logger.debug( + "args:\n" + f"delegate: {delegate}\n" + f"call_hash: {call_hash}\n" + f"network: {network}\n" + f"wait_for_finalization: {wait_for_finalization}\n" + f"wait_for_inclusion: {wait_for_inclusion}\n" + f"era: {period}\n" + ) + + wallet = self.wallet_ask( + wallet_name=wallet_name, + wallet_path=wallet_path, + wallet_hotkey=wallet_hotkey, + ask_for=[WO.NAME, WO.PATH], + validate=WV.WALLET, + ) + + delegate = delegate or wallet.coldkeypub.ss58_address + + # Try to find the announcement in the local DB + got_call_from_db: Optional[int] = None + with ProxyAnnouncements.get_db() as (conn, cursor): + announcements = ProxyAnnouncements.read_rows(conn, cursor) + + if not call_hash: + potential_call_matches = [] + for row in announcements: + ( + id_, + address, + epoch_time, + block_, + call_hash_, + call_hex_, + call_serialized, + executed_int, + ) = row + executed = bool(executed_int) + if address == delegate and executed is False: + potential_call_matches.append(row) + + if len(potential_call_matches) == 0: + if not prompt: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": f"No pending announcements found for delegate {delegate}. Please provide --call-hash explicitly.", + "extrinsic_identifier": None, + } + ) + else: + err_console.print( + f":cross_mark:[red]Error: No pending announcements found for delegate {delegate}. " + f"Please provide --call-hash explicitly." + ) + return + call_hash = Prompt.ask( + "Enter the call hash of the announcement to reject" + ) + elif len(potential_call_matches) == 1: + call_hash = potential_call_matches[0][4] + got_call_from_db = potential_call_matches[0][0] + console.print(f"Found announcement with call hash: {call_hash}") + else: + if not prompt: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": f"Multiple pending announcements found for delegate {delegate}. Please provide --call-hash explicitly.", + "extrinsic_identifier": None, + } + ) + else: + err_console.print( + f":cross_mark:[red]Error: Multiple pending announcements found for delegate {delegate}. " + f"Please run without {arg__('--no-prompt')} to select one, or provide --call-hash explicitly." + ) + return + else: + console.print( + f"Found {len(potential_call_matches)} pending announcements. " + f"Please select the one to reject:" + ) + for row in potential_call_matches: + ( + id_, + address, + epoch_time, + block_, + call_hash_, + call_hex_, + call_serialized, + executed_int, + ) = row + console.print( + f"Time: {datetime.datetime.fromtimestamp(epoch_time)}\n" + f"Call Hash: {call_hash_}\nCall:\n" + ) + console.print_json(call_serialized) + if confirm_action( + "Is this the announcement to reject?", + decline=decline, + quiet=quiet, + ): + call_hash = call_hash_ + got_call_from_db = id_ + break + if call_hash is None: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": "No announcement selected.", + "extrinsic_identifier": None, + } + ) + else: + console.print("No announcement selected.") + return + else: + # call_hash provided, try to find it in DB + for row in announcements: + ( + id_, + address, + epoch_time, + block_, + call_hash_, + call_hex_, + call_serialized, + executed_int, + ) = row + executed = bool(executed_int) + if ( + call_hash_ == call_hash + and address == delegate + and executed is False + ): + got_call_from_db = id_ + break + + success = self._run_command( + proxy_commands.reject_announcement( + subtensor=self.initialize_chain(network), + wallet=wallet, + delegate=delegate, + call_hash=call_hash, + prompt=prompt, + decline=decline, + quiet=quiet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + period=period, + json_output=json_output, + ) + ) + + if success and got_call_from_db is not None: + with ProxyAnnouncements.get_db() as (conn, cursor): + ProxyAnnouncements.mark_as_executed(conn, cursor, got_call_from_db) + @staticmethod def convert( from_rao: Optional[str] = typer.Option( diff --git a/bittensor_cli/src/commands/proxy.py b/bittensor_cli/src/commands/proxy.py index bbca6d0e0..f50872824 100644 --- a/bittensor_cli/src/commands/proxy.py +++ b/bittensor_cli/src/commands/proxy.py @@ -2,7 +2,9 @@ import sys from rich.prompt import Prompt, FloatPrompt, IntPrompt +from rich.table import Table, Column from scalecodec import GenericCall, ScaleBytes +from scalecodec.utils.ss58 import ss58_encode from bittensor_cli.src import COLORS from bittensor_cli.src.bittensor.balances import Balance @@ -53,7 +55,343 @@ class ProxyType(StrEnum): RootClaim = "RootClaim" -# TODO add announce with also --reject and --remove +async def list_proxies( + subtensor: "SubtensorInterface", + address: str, + json_output: bool, +) -> None: + """ + Lists all proxies for a given account by querying the chain. + + Args: + subtensor: The SubtensorInterface instance. + address: The SS58 address to query proxies for. + json_output: Whether to output in JSON format. + """ + try: + # Use subtensor.query() which automatically extracts .value + proxies_data = await subtensor.query( + module="Proxy", + storage_function="Proxies", + params=[address], + ) + + # Handle different possible data structures from the chain + # Chain returns: [(proxy_dicts_tuple,), deposit] + proxies_list = [] + deposit = 0 + if proxies_data: + # Handle tuple/list format (proxies_tuple, deposit) + if isinstance(proxies_data, (list, tuple)) and len(proxies_data) >= 2: + raw_proxies = proxies_data[0] if proxies_data[0] else () + deposit = proxies_data[1] if len(proxies_data) > 1 else 0 + # Unwrap nested tuples: (({...},),) -> [{...}] + if isinstance(raw_proxies, (list, tuple)): + for item in raw_proxies: + # Each item might be a tuple containing a dict + if isinstance(item, (list, tuple)): + for sub_item in item: + if isinstance(sub_item, dict): + proxies_list.append(sub_item) + elif isinstance(item, dict): + proxies_list.append(item) + + # Normalize proxy data - convert chain format to user-friendly format + normalized_proxies = [] + for p in proxies_list: + if isinstance(p, dict): + # Handle delegate - might be bytes tuple or string + delegate_raw = p.get("delegate") or p.get("delegatee", "") + if isinstance(delegate_raw, (list, tuple)): + # Convert bytes tuple to SS58 address + # Unwrap nested tuple: ((bytes,),) -> bytes + while ( + isinstance(delegate_raw, (list, tuple)) + and len(delegate_raw) == 1 + ): + delegate_raw = delegate_raw[0] + if ( + isinstance(delegate_raw, (list, tuple)) + and len(delegate_raw) == 32 + ): + # Convert 32-byte tuple to SS58 address + try: + delegate = ss58_encode(bytes(delegate_raw), ss58_format=42) + except Exception as e: + # Fallback: try with substrateinterface + try: + from substrateinterface import Keypair + + delegate = Keypair( + public_key=bytes(delegate_raw) + ).ss58_address + except Exception: + delegate = f"error:{e}:{delegate_raw}" + else: + delegate = str(delegate_raw) + else: + delegate = str(delegate_raw) if delegate_raw else "" + + # Handle proxy_type - might be dict like {'Any': ()} or string + proxy_type_raw = p.get("proxy_type") or p.get("proxyType", "") + if isinstance(proxy_type_raw, dict): + # Extract the key as the proxy type + proxy_type = ( + list(proxy_type_raw.keys())[0] if proxy_type_raw else "" + ) + else: + proxy_type = str(proxy_type_raw) if proxy_type_raw else "" + + normalized_proxies.append( + { + "delegate": delegate, + "proxy_type": proxy_type, + "delay": p.get("delay", 0), + } + ) + + if json_output: + json_console.print_json( + data={ + "success": True, + "address": address, + "deposit": str(deposit), + "proxies": normalized_proxies, + } + ) + else: + if not normalized_proxies: + console.print(f"No proxies found for address {address}") + return + + table = Table( + Column("Delegate", style="cyan"), + Column("Proxy Type", style="green"), + Column("Delay", style="yellow"), + title=f"Proxies for {address}", + caption=f"Total deposit: {deposit}", + ) + for proxy in normalized_proxies: + table.add_row( + proxy["delegate"], + proxy["proxy_type"], + str(proxy["delay"]), + ) + console.print(table) + + except Exception as e: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": str(e), + "address": address, + "proxies": None, + } + ) + else: + print_error(f"Failed to list proxies: {e}") + + +async def remove_all_proxies( + subtensor: "SubtensorInterface", + wallet: "Wallet", + prompt: bool, + decline: bool, + quiet: bool, + wait_for_inclusion: bool, + wait_for_finalization: bool, + period: int, + json_output: bool, +) -> None: + """ + Removes all proxies for the wallet's coldkey by calling the removeProxies extrinsic. + + Args: + subtensor: The SubtensorInterface instance. + wallet: The wallet whose proxies will be removed. + prompt: Whether to prompt for confirmation. + decline: If True, automatically decline confirmation prompts. + quiet: If True, suppress output when auto-declining. + wait_for_inclusion: Wait for the transaction to be included in a block. + wait_for_finalization: Wait for the transaction to be finalized. + period: The era period for the extrinsic. + json_output: Whether to output in JSON format. + """ + if prompt: + if not confirm_action( + "[bold red]Warning:[/bold red] This will remove ALL proxies for your account. " + "Do you want to proceed?", + decline=decline, + quiet=quiet, + ): + if json_output: + json_console.print_json( + data={ + "success": False, + "message": "Operation cancelled by user", + "extrinsic_identifier": None, + } + ) + return None + + if not (ulw := unlock_key(wallet, print_out=not json_output)).success: + if not json_output: + print_error(ulw.message) + else: + json_console.print_json( + data={ + "success": ulw.success, + "message": ulw.message, + "extrinsic_identifier": None, + } + ) + return None + + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="remove_proxies", + call_params={}, + ) + + success, msg, receipt = await subtensor.sign_and_send_extrinsic( + call=call, + wallet=wallet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + era={"period": period}, + ) + + if success: + await print_extrinsic_id(receipt) + if json_output: + json_console.print_json( + data={ + "success": success, + "message": "All proxies removed successfully", + "extrinsic_identifier": await receipt.get_extrinsic_identifier(), + } + ) + else: + console.print( + ":white_check_mark:[green]All proxies removed successfully![/green]" + ) + else: + if json_output: + json_console.print_json( + data={ + "success": success, + "message": msg, + "extrinsic_identifier": None, + } + ) + else: + print_error(f"Failed to remove all proxies: {msg}") + + +async def reject_announcement( + subtensor: "SubtensorInterface", + wallet: "Wallet", + delegate: str, + call_hash: str, + prompt: bool, + decline: bool, + quiet: bool, + wait_for_inclusion: bool, + wait_for_finalization: bool, + period: int, + json_output: bool, +) -> bool: + """ + Rejects an announced proxy call by calling the reject_announcement extrinsic. + + This removes a previously announced call from the pending announcements, + preventing it from being executed. + + Args: + subtensor: The SubtensorInterface instance. + wallet: The wallet to sign the transaction (must be the real account). + delegate: The SS58 address of the delegate who made the announcement. + call_hash: The hash of the call to reject. + prompt: Whether to prompt for confirmation. + decline: If True, automatically decline confirmation prompts. + quiet: If True, suppress output when auto-declining. + wait_for_inclusion: Wait for the transaction to be included in a block. + wait_for_finalization: Wait for the transaction to be finalized. + period: The era period for the extrinsic. + json_output: Whether to output in JSON format. + + Returns: + True if the rejection was successful, False otherwise. + """ + if prompt: + if not confirm_action( + f"This will reject the announced call with hash {call_hash} from delegate {delegate}. " + "Do you want to proceed?", + decline=decline, + quiet=quiet, + ): + return False + + if not (ulw := unlock_key(wallet, print_out=not json_output)).success: + if not json_output: + print_error(ulw.message) + else: + json_console.print_json( + data={ + "success": ulw.success, + "message": ulw.message, + "extrinsic_identifier": None, + } + ) + return False + + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="reject_announcement", + call_params={ + "delegate": delegate, + "call_hash": call_hash, + }, + ) + + success, msg, receipt = await subtensor.sign_and_send_extrinsic( + call=call, + wallet=wallet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + era={"period": period}, + ) + + if success: + await print_extrinsic_id(receipt) + if json_output: + json_console.print_json( + data={ + "success": success, + "message": "Announcement rejected successfully", + "delegate": delegate, + "call_hash": call_hash, + "extrinsic_identifier": await receipt.get_extrinsic_identifier(), + } + ) + else: + console.print( + ":white_check_mark:[green]Announcement rejected successfully![/green]" + ) + return True + else: + if json_output: + json_console.print_json( + data={ + "success": success, + "message": msg, + "extrinsic_identifier": None, + } + ) + else: + print_error(f"Failed to reject announcement: {msg}") + return False async def submit_proxy( @@ -84,6 +422,7 @@ async def submit_proxy( announce_only=announce_only, ) if success: + await print_extrinsic_id(receipt) if json_output: json_console.print_json( data={ @@ -93,8 +432,8 @@ async def submit_proxy( } ) else: - await print_extrinsic_id(receipt) console.print(":white_check_mark:[green]Success![/green]") + else: if json_output: json_console.print_json( @@ -168,6 +507,7 @@ async def create_proxy( ), ) if success: + await print_extrinsic_id(receipt) created_pure = None created_spawner = None created_proxy_type = None @@ -177,12 +517,47 @@ async def create_proxy( created_pure = attrs["pure"] created_spawner = attrs["who"] created_proxy_type = getattr(ProxyType, attrs["proxy_type"]) + arg_start = "`" if json_output else "[blue]" + arg_end = "`" if json_output else "[/blue]" msg = ( f"Created pure '{created_pure}' " f"from spawner '{created_spawner}' " f"with proxy type '{created_proxy_type.value}' " f"with delay {delay}." ) + console.print(msg) + if not prompt: + console.print( + f" You can add this to your config with {arg_start}" + f"btcli config add-proxy " + f"--name --address {created_pure} --proxy-type {created_proxy_type.value} " + f"--delay {delay} --spawner {created_spawner}" + f"{arg_end}" + ) + else: + if confirm_action( + "Would you like to add this to your address book?", + decline=decline, + quiet=quiet, + ): + proxy_name = Prompt.ask("Name this proxy") + note = Prompt.ask("[Optional] Add a note for this proxy", default="") + with ProxyAddressBook.get_db() as (conn, cursor): + ProxyAddressBook.add_entry( + conn, + cursor, + name=proxy_name, + ss58_address=created_pure, + delay=delay, + proxy_type=created_proxy_type.value, + note=note, + spawner=created_spawner, + ) + console.print( + f"Added to Proxy Address Book.\n" + f"Show this information with [{COLORS.G.ARG}]btcli config proxies[/{COLORS.G.ARG}]" + ) + return None if json_output: json_console.print_json( @@ -198,43 +573,7 @@ async def create_proxy( "extrinsic_identifier": await receipt.get_extrinsic_identifier(), } ) - else: - await print_extrinsic_id(receipt) - console.print(msg) - if not prompt: - console.print( - f" You can add this to your config with [blue]" - f"btcli config add-proxy " - f"--name --address {created_pure} --proxy-type {created_proxy_type.value} " - f"--delay {delay} --spawner {created_spawner}" - f"[/blue]" - ) - else: - if confirm_action( - "Would you like to add this to your address book?", - decline=decline, - quiet=quiet, - ): - proxy_name = Prompt.ask("Name this proxy") - note = Prompt.ask( - "[Optional] Add a note for this proxy", default="" - ) - with ProxyAddressBook.get_db() as (conn, cursor): - ProxyAddressBook.add_entry( - conn, - cursor, - name=proxy_name, - ss58_address=created_pure, - delay=delay, - proxy_type=created_proxy_type.value, - note=note, - spawner=created_spawner, - ) - console.print( - f"Added to Proxy Address Book.\n" - f"Show this information with [{COLORS.G.ARG}]btcli config proxies[/{COLORS.G.ARG}]" - ) - return None + else: if json_output: json_console.print_json( @@ -369,6 +708,7 @@ async def add_proxy( era={"period": period}, ) if success: + await print_extrinsic_id(receipt) delegatee = None delegator = None created_proxy_type = None @@ -379,12 +719,46 @@ async def add_proxy( delegator = attrs["delegator"] created_proxy_type = getattr(ProxyType, attrs["proxy_type"]) break + arg_start = "`" if json_output else "[blue]" + arg_end = "`" if json_output else "[/blue]" msg = ( f"Added proxy delegatee '{delegatee}' " f"from delegator '{delegator}' " f"with proxy type '{created_proxy_type.value}' " f"with delay {delay}." ) + console.print(msg) + if not prompt: + console.print( + f" You can add this to your config with {arg_start}" + f"btcli config add-proxy " + f"--name --address {delegatee} --proxy-type {created_proxy_type.value} --delegator " + f"{delegator} --delay {delay}" + f"{arg_end}" + ) + else: + if confirm_action( + "Would you like to add this to your address book?", + decline=decline, + quiet=quiet, + ): + proxy_name = Prompt.ask("Name this proxy") + note = Prompt.ask("[Optional] Add a note for this proxy", default="") + with ProxyAddressBook.get_db() as (conn, cursor): + ProxyAddressBook.add_entry( + conn, + cursor, + name=proxy_name, + ss58_address=delegator, + delay=delay, + proxy_type=created_proxy_type.value, + note=note, + spawner=delegatee, + ) + console.print( + f"Added to Proxy Address Book.\n" + f"Show this information with [{COLORS.G.ARG}]btcli config proxies[/{COLORS.G.ARG}]" + ) if json_output: json_console.print_json( @@ -400,42 +774,7 @@ async def add_proxy( "extrinsic_identifier": await receipt.get_extrinsic_identifier(), } ) - else: - await print_extrinsic_id(receipt) - console.print(msg) - if not prompt: - console.print( - f" You can add this to your config with [blue]" - f"btcli config add-proxy " - f"--name --address {delegatee} --proxy-type {created_proxy_type.value} --delegator " - f"{delegator} --delay {delay}" - f"[/blue]" - ) - else: - if confirm_action( - "Would you like to add this to your address book?", - decline=decline, - quiet=quiet, - ): - proxy_name = Prompt.ask("Name this proxy") - note = Prompt.ask( - "[Optional] Add a note for this proxy", default="" - ) - with ProxyAddressBook.get_db() as (conn, cursor): - ProxyAddressBook.add_entry( - conn, - cursor, - name=proxy_name, - ss58_address=delegator, - delay=delay, - proxy_type=created_proxy_type.value, - note=note, - spawner=delegatee, - ) - console.print( - f"Added to Proxy Address Book.\n" - f"Show this information with [{COLORS.G.ARG}]btcli config proxies[/{COLORS.G.ARG}]" - ) + else: if json_output: json_console.print_json( diff --git a/tests/e2e_tests/test_proxy.py b/tests/e2e_tests/test_proxy.py index e5e76724e..7d1af6cf8 100644 --- a/tests/e2e_tests/test_proxy.py +++ b/tests/e2e_tests/test_proxy.py @@ -10,8 +10,11 @@ * btcli proxy create * btcli proxy add * btcli proxy remove +* btcli proxy remove --all * btcli proxy kill * btcli proxy execute +* btcli proxy list +* btcli proxy reject """ @@ -684,3 +687,111 @@ def test_add_proxy(local_chain, wallet_setup): os.environ["BTCLI_PROXIES_PATH"] = "" if os.path.exists(testing_db_loc): os.remove(testing_db_loc) + + +def test_proxy_list(local_chain, wallet_setup): + """ + Tests the proxy list command. + + Steps: + 1. Add a proxy to Alice's account + 2. List proxies for Alice's account + 3. Verify the proxy is in the list + 4. Remove the proxy + """ + wallet_path_alice = "//Alice" + wallet_path_bob = "//Bob" + + keypair_alice, wallet_alice, wallet_path_alice, exec_command_alice = wallet_setup( + wallet_path_alice + ) + keypair_bob, wallet_bob, wallet_path_bob, exec_command_bob = wallet_setup( + wallet_path_bob + ) + proxy_type = "Any" + delay = 0 + + # Add Bob as a proxy for Alice + add_result = exec_command_alice( + command="proxy", + sub_command="add", + extra_args=[ + "--wallet-path", + wallet_path_alice, + "--chain", + "ws://127.0.0.1:9945", + "--wallet-name", + "default", + "--delegate", + wallet_bob.coldkeypub.ss58_address, + "--proxy-type", + proxy_type, + "--delay", + str(delay), + "--period", + "128", + "--no-prompt", + "--json-output", + ], + ) + add_result_output = json.loads(add_result.stdout) + assert add_result_output["success"] is True + print("Passed proxy add for list test") + + # Wait for chain state to propagate + time.sleep(2) + + # List proxies for Alice + list_result = exec_command_alice( + command="proxy", + sub_command="list", + extra_args=[ + "--address", + wallet_alice.coldkeypub.ss58_address, + "--chain", + "ws://127.0.0.1:9945", + "--json-output", + ], + ) + list_result_output = json.loads(list_result.stdout) + assert list_result_output["success"] is True + assert list_result_output["address"] == wallet_alice.coldkeypub.ss58_address + assert len(list_result_output["proxies"]) >= 1 + + # Verify Bob is in the proxy list + found_bob = False + for proxy in list_result_output["proxies"]: + if proxy["delegate"] == wallet_bob.coldkeypub.ss58_address: + found_bob = True + assert proxy["proxy_type"] == proxy_type + assert proxy["delay"] == delay + break + assert found_bob, "Bob should be in Alice's proxy list" + print("Passed proxy list") + + # Clean up - remove the proxy + remove_result = exec_command_alice( + command="proxy", + sub_command="remove", + extra_args=[ + "--wallet-path", + wallet_path_alice, + "--chain", + "ws://127.0.0.1:9945", + "--wallet-name", + "default", + "--delegate", + wallet_bob.coldkeypub.ss58_address, + "--proxy-type", + proxy_type, + "--delay", + str(delay), + "--period", + "128", + "--no-prompt", + "--json-output", + ], + ) + remove_result_output = json.loads(remove_result.stdout) + assert remove_result_output["success"] is True + print("Passed proxy removal cleanup") diff --git a/tests/unit_tests/test_cli.py b/tests/unit_tests/test_cli.py index 60cc10708..02a01e558 100644 --- a/tests/unit_tests/test_cli.py +++ b/tests/unit_tests/test_cli.py @@ -11,6 +11,11 @@ from unittest.mock import AsyncMock, patch, MagicMock, Mock from bittensor_cli.src.bittensor.subtensor_interface import SubtensorInterface +from bittensor_cli.src.commands.proxy import ( + list_proxies, + remove_all_proxies, + reject_announcement, +) def test_parse_mnemonic(): @@ -805,3 +810,640 @@ async def test_set_root_weights_skips_current_weights_without_prompt(): ) mock_get_current.assert_not_called() + + +# ============================================================================ +# Tests for proxy list command +# ============================================================================ + + +@pytest.mark.asyncio +async def test_list_proxies_success(): + """Test that list_proxies correctly queries and displays proxies""" + mock_subtensor = AsyncMock() + + # Mock the query result - list_proxies uses subtensor.query() not substrate.query + # Returns tuple: (proxies_list, deposit) + mock_subtensor.query = AsyncMock( + return_value=( + [ + {"delegate": "5GDel1...", "proxy_type": "Staking", "delay": 0}, + {"delegate": "5GDel2...", "proxy_type": "Transfer", "delay": 100}, + ], + 1000000, # deposit + ) + ) + + with patch("bittensor_cli.src.commands.proxy.console") as mock_console: + await list_proxies( + subtensor=mock_subtensor, + address="5GTest...", + json_output=False, + ) + + # Verify query was called correctly + mock_subtensor.query.assert_awaited_once_with( + module="Proxy", + storage_function="Proxies", + params=["5GTest..."], + ) + + # Verify console output was called (table was printed) + assert mock_console.print.called + + +@pytest.mark.asyncio +async def test_list_proxies_json_output(): + """Test that list_proxies outputs JSON correctly""" + mock_subtensor = AsyncMock() + + # Mock the query result - list_proxies uses subtensor.query() + mock_subtensor.query = AsyncMock( + return_value=( + [{"delegate": "5GDel1...", "proxy_type": "Staking", "delay": 0}], + 500000, + ) + ) + + with patch("bittensor_cli.src.commands.proxy.json_console") as mock_json_console: + await list_proxies( + subtensor=mock_subtensor, + address="5GTest...", + json_output=True, + ) + + # Verify JSON output was called + mock_json_console.print_json.assert_called_once() + call_args = mock_json_console.print_json.call_args + data = call_args.kwargs["data"] + assert data["success"] is True + assert data["address"] == "5GTest..." + assert len(data["proxies"]) == 1 + + +@pytest.mark.asyncio +async def test_list_proxies_empty(): + """Test that list_proxies handles empty proxy list""" + mock_subtensor = AsyncMock() + + # Mock the query result - empty proxies list + mock_subtensor.query = AsyncMock(return_value=([], 0)) + + with patch("bittensor_cli.src.commands.proxy.console") as mock_console: + await list_proxies( + subtensor=mock_subtensor, + address="5GTest...", + json_output=False, + ) + + # Verify "no proxies found" message + mock_console.print.assert_called_once() + assert "No proxies found" in str(mock_console.print.call_args) + + +@pytest.mark.asyncio +async def test_list_proxies_error_handling(): + """Test that list_proxies handles errors gracefully""" + mock_subtensor = AsyncMock() + mock_subtensor.query = AsyncMock(side_effect=Exception("Connection error")) + + with patch("bittensor_cli.src.commands.proxy.print_error") as mock_print_error: + await list_proxies( + subtensor=mock_subtensor, + address="5GTest...", + json_output=False, + ) + + # Verify error was printed + mock_print_error.assert_called_once() + assert "Failed to list proxies" in str(mock_print_error.call_args) + + +# ============================================================================ +# Tests for proxy remove --all command +# ============================================================================ + + +@pytest.mark.asyncio +async def test_remove_all_proxies_success(): + """Test that remove_all_proxies successfully removes all proxies""" + mock_subtensor = MagicMock() + mock_substrate = AsyncMock() + mock_subtensor.substrate = mock_substrate + + mock_call = MagicMock() + mock_substrate.compose_call = AsyncMock(return_value=mock_call) + + mock_receipt = AsyncMock() + mock_receipt.get_extrinsic_identifier = AsyncMock(return_value="12345-1") + mock_subtensor.sign_and_send_extrinsic = AsyncMock( + return_value=(True, "", mock_receipt) + ) + + mock_wallet = MagicMock() + + with ( + patch("bittensor_cli.src.commands.proxy.unlock_key") as mock_unlock, + patch("bittensor_cli.src.commands.proxy.console") as mock_console, + patch("bittensor_cli.src.commands.proxy.print_extrinsic_id"), + ): + mock_unlock.return_value = MagicMock(success=True) + + await remove_all_proxies( + subtensor=mock_subtensor, + wallet=mock_wallet, + prompt=False, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + # Verify compose_call was called with remove_proxies + mock_substrate.compose_call.assert_awaited_once_with( + call_module="Proxy", + call_function="remove_proxies", + call_params={}, + ) + + # Verify success message + assert mock_console.print.called + assert "All proxies removed" in str(mock_console.print.call_args) + + +@pytest.mark.asyncio +async def test_remove_all_proxies_with_prompt_declined(): + """Test that remove_all_proxies exits when user declines prompt""" + mock_subtensor = MagicMock() + mock_wallet = MagicMock() + + with patch("bittensor_cli.src.commands.proxy.confirm_action") as mock_confirm: + mock_confirm.return_value = False + + result = await remove_all_proxies( + subtensor=mock_subtensor, + wallet=mock_wallet, + prompt=True, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + assert result is None + mock_confirm.assert_called_once() + + +@pytest.mark.asyncio +async def test_remove_all_proxies_unlock_failure(): + """Test that remove_all_proxies handles wallet unlock failure""" + mock_subtensor = MagicMock() + mock_wallet = MagicMock() + + with ( + patch("bittensor_cli.src.commands.proxy.unlock_key") as mock_unlock, + patch("bittensor_cli.src.commands.proxy.print_error") as mock_print_error, + ): + mock_unlock.return_value = MagicMock(success=False, message="Wrong password") + + result = await remove_all_proxies( + subtensor=mock_subtensor, + wallet=mock_wallet, + prompt=False, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + assert result is None + mock_print_error.assert_called_once() + + +# ============================================================================ +# Tests for proxy reject command +# ============================================================================ + + +@pytest.mark.asyncio +async def test_reject_announcement_success(): + """Test that reject_announcement successfully rejects an announcement""" + mock_subtensor = MagicMock() + mock_substrate = AsyncMock() + mock_subtensor.substrate = mock_substrate + + mock_call = MagicMock() + mock_substrate.compose_call = AsyncMock(return_value=mock_call) + + mock_receipt = AsyncMock() + mock_receipt.get_extrinsic_identifier = AsyncMock(return_value="12345-1") + mock_subtensor.sign_and_send_extrinsic = AsyncMock( + return_value=(True, "", mock_receipt) + ) + + mock_wallet = MagicMock() + + with ( + patch("bittensor_cli.src.commands.proxy.unlock_key") as mock_unlock, + patch("bittensor_cli.src.commands.proxy.console") as mock_console, + patch("bittensor_cli.src.commands.proxy.print_extrinsic_id"), + ): + mock_unlock.return_value = MagicMock(success=True) + + await reject_announcement( + subtensor=mock_subtensor, + wallet=mock_wallet, + delegate="5GDelegate...", + call_hash="0x1234abcd", + prompt=False, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + # Verify compose_call was called with reject_announcement + mock_substrate.compose_call.assert_awaited_once_with( + call_module="Proxy", + call_function="reject_announcement", + call_params={ + "delegate": "5GDelegate...", + "call_hash": "0x1234abcd", + }, + ) + + # Verify success message + assert mock_console.print.called + assert "rejected successfully" in str(mock_console.print.call_args) + + +@pytest.mark.asyncio +async def test_reject_announcement_json_output(): + """Test that reject_announcement outputs JSON correctly""" + mock_subtensor = MagicMock() + mock_substrate = AsyncMock() + mock_subtensor.substrate = mock_substrate + + mock_call = MagicMock() + mock_substrate.compose_call = AsyncMock(return_value=mock_call) + + mock_receipt = AsyncMock() + mock_receipt.get_extrinsic_identifier = AsyncMock(return_value="12345-1") + mock_subtensor.sign_and_send_extrinsic = AsyncMock( + return_value=(True, "", mock_receipt) + ) + + mock_wallet = MagicMock() + + with ( + patch("bittensor_cli.src.commands.proxy.unlock_key") as mock_unlock, + patch("bittensor_cli.src.commands.proxy.json_console") as mock_json_console, + patch("bittensor_cli.src.commands.proxy.print_extrinsic_id"), + ): + mock_unlock.return_value = MagicMock(success=True) + + await reject_announcement( + subtensor=mock_subtensor, + wallet=mock_wallet, + delegate="5GDelegate...", + call_hash="0x1234abcd", + prompt=False, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=True, + ) + + # Verify JSON output + mock_json_console.print_json.assert_called_once() + call_args = mock_json_console.print_json.call_args + data = call_args.kwargs["data"] + assert data["success"] is True + assert data["delegate"] == "5GDelegate..." + assert data["call_hash"] == "0x1234abcd" + + +@pytest.mark.asyncio +async def test_reject_announcement_with_prompt_declined(): + """Test that reject_announcement exits when user declines prompt""" + mock_subtensor = MagicMock() + mock_wallet = MagicMock() + + with patch("bittensor_cli.src.commands.proxy.confirm_action") as mock_confirm: + mock_confirm.return_value = False + + result = await reject_announcement( + subtensor=mock_subtensor, + wallet=mock_wallet, + delegate="5GDelegate...", + call_hash="0x1234abcd", + prompt=True, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + # Function returns False when user declines confirmation + assert result is False + mock_confirm.assert_called_once() + + +@pytest.mark.asyncio +async def test_reject_announcement_failure(): + """Test that reject_announcement handles extrinsic failure""" + mock_subtensor = MagicMock() + mock_substrate = AsyncMock() + mock_subtensor.substrate = mock_substrate + + mock_call = MagicMock() + mock_substrate.compose_call = AsyncMock(return_value=mock_call) + mock_subtensor.sign_and_send_extrinsic = AsyncMock( + return_value=(False, "Announcement not found", None) + ) + + mock_wallet = MagicMock() + + with ( + patch("bittensor_cli.src.commands.proxy.unlock_key") as mock_unlock, + patch("bittensor_cli.src.commands.proxy.print_error") as mock_print_error, + ): + mock_unlock.return_value = MagicMock(success=True) + + await reject_announcement( + subtensor=mock_subtensor, + wallet=mock_wallet, + delegate="5GDelegate...", + call_hash="0x1234abcd", + prompt=False, + decline=False, + quiet=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + json_output=False, + ) + + # Verify error message + mock_print_error.assert_called_once() + assert "Failed to reject" in str(mock_print_error.call_args) + + +# ============================================================================ +# Tests for CLI proxy_remove with --all flag +# ============================================================================ + + +def test_proxy_remove_all_and_delegate_mutually_exclusive(): + """Test that --all and --delegate cannot be used together""" + cli_manager = CLIManager() + + with pytest.raises(typer.Exit): + cli_manager.proxy_remove( + delegate="5GDelegate...", + all_proxies=True, # Both specified + network=None, + proxy_type=None, + delay=0, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + # Error message is printed to stderr, test passes if typer.Exit is raised + + +def test_proxy_remove_requires_delegate_or_all(): + """Test that either --delegate or --all must be specified""" + cli_manager = CLIManager() + + with pytest.raises(typer.Exit): + cli_manager.proxy_remove( + delegate=None, + all_proxies=False, # Neither specified + network=None, + proxy_type=None, + delay=0, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, # No prompt to ask for delegate + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + # Error message is printed to stderr, test passes if typer.Exit is raised + + +def test_proxy_remove_with_all_flag_calls_remove_all_proxies(): + """Test that --all flag calls remove_all_proxies""" + cli_manager = CLIManager() + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "wallet_ask") as mock_wallet_ask, + patch.object(cli_manager, "initialize_chain") as mock_init_chain, + patch.object(cli_manager, "_run_command") as mock_run_command, + patch("bittensor_cli.cli.proxy_commands.remove_all_proxies"), + ): + mock_wallet = Mock() + mock_wallet_ask.return_value = mock_wallet + mock_subtensor = Mock() + mock_init_chain.return_value = mock_subtensor + + cli_manager.proxy_remove( + delegate=None, + all_proxies=True, + network=None, + proxy_type=None, + delay=0, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + + # Verify _run_command was called (which wraps remove_all_proxies) + mock_run_command.assert_called_once() + + +def test_proxy_remove_with_delegate_calls_remove_proxy(): + """Test that --delegate flag calls remove_proxy""" + cli_manager = CLIManager() + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "wallet_ask") as mock_wallet_ask, + patch.object(cli_manager, "initialize_chain") as mock_init_chain, + patch.object(cli_manager, "_run_command") as mock_run_command, + patch("bittensor_cli.cli.proxy_commands.remove_proxy"), + ): + mock_wallet = Mock() + mock_wallet_ask.return_value = mock_wallet + mock_subtensor = Mock() + mock_init_chain.return_value = mock_subtensor + + cli_manager.proxy_remove( + delegate="5GDelegate...", + all_proxies=False, + network=None, + proxy_type=None, + delay=0, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + + # Verify _run_command was called + mock_run_command.assert_called_once() + + +# ============================================================================ +# Tests for CLI proxy_list command +# ============================================================================ + + +def test_proxy_list_with_address(): + """Test that proxy_list uses provided address""" + cli_manager = CLIManager() + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "initialize_chain") as mock_init_chain, + patch.object(cli_manager, "_run_command") as mock_run_command, + patch("bittensor_cli.cli.proxy_commands.list_proxies"), + ): + mock_subtensor = Mock() + mock_init_chain.return_value = mock_subtensor + + cli_manager.proxy_list( + address="5GAddress...", + network=None, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + quiet=False, + verbose=False, + json_output=False, + ) + + # Verify _run_command was called + mock_run_command.assert_called_once() + + +def test_proxy_list_without_address_uses_wallet(): + """Test that proxy_list uses wallet coldkey when no address provided""" + cli_manager = CLIManager() + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "wallet_ask") as mock_wallet_ask, + patch.object(cli_manager, "initialize_chain") as mock_init_chain, + patch.object(cli_manager, "_run_command") as mock_run_command, + ): + mock_wallet = Mock() + mock_wallet.coldkeypub.ss58_address = "5GWalletColdkey..." + mock_wallet_ask.return_value = mock_wallet + mock_subtensor = Mock() + mock_init_chain.return_value = mock_subtensor + + cli_manager.proxy_list( + address=None, # No address provided + network=None, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + quiet=False, + verbose=False, + json_output=False, + ) + + # Verify wallet_ask was called to get wallet + mock_wallet_ask.assert_called_once() + # Verify _run_command was called + mock_run_command.assert_called_once() + + +# ============================================================================ +# Tests for CLI proxy_reject command +# ============================================================================ + + +def test_proxy_reject_calls_reject_announcement(): + """Test that proxy_reject calls reject_announcement""" + cli_manager = CLIManager() + + # Create a mock context manager for the database + mock_db_context = MagicMock() + mock_db_context.__enter__ = MagicMock(return_value=(MagicMock(), MagicMock())) + mock_db_context.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "wallet_ask") as mock_wallet_ask, + patch.object(cli_manager, "initialize_chain") as mock_init_chain, + patch.object(cli_manager, "_run_command") as mock_run_command, + patch("bittensor_cli.cli.proxy_commands.reject_announcement"), + patch( + "bittensor_cli.cli.ProxyAnnouncements.get_db", return_value=mock_db_context + ), + ): + mock_wallet = Mock() + mock_wallet.coldkeypub = Mock() + mock_wallet.coldkeypub.ss58_address = "5GDelegate..." + mock_wallet_ask.return_value = mock_wallet + mock_subtensor = Mock() + mock_init_chain.return_value = mock_subtensor + + cli_manager.proxy_reject( + delegate="5GDelegate...", + call_hash="0x1234abcd", + network=None, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + + # Verify _run_command was called + mock_run_command.assert_called_once()