From 586abdfd660e2103d45348c481eae4d8e9bd8b43 Mon Sep 17 00:00:00 2001 From: Daniel Dereefaka Date: Tue, 9 Dec 2025 01:31:08 +0100 Subject: [PATCH] feat: Add proxy commands for managing delegate accounts (#671) Add support for Substrate proxy pallet in btcli with the following commands: - btcli proxy add: Add a proxy delegate with specific permissions - btcli proxy remove: Remove a specific proxy delegate - btcli proxy remove-all: Remove all proxy delegates at once - btcli proxy list: List all proxies for an account Supported proxy types: Any, NonTransfer, Governance, Staking, Registration, SenateVoting, Transfer, SmallTransfer, RootWeights, ChildKeys, SudoUncheckedSetCode Closes #671 --- bittensor_cli/cli.py | 207 +++++++++++++ bittensor_cli/src/commands/proxy/__init__.py | 1 + bittensor_cli/src/commands/proxy/add.py | 172 +++++++++++ bittensor_cli/src/commands/proxy/list.py | 156 ++++++++++ bittensor_cli/src/commands/proxy/remove.py | 269 +++++++++++++++++ tests/unit_tests/test_proxy.py | 299 +++++++++++++++++++ 6 files changed, 1104 insertions(+) create mode 100644 bittensor_cli/src/commands/proxy/__init__.py create mode 100644 bittensor_cli/src/commands/proxy/add.py create mode 100644 bittensor_cli/src/commands/proxy/list.py create mode 100644 bittensor_cli/src/commands/proxy/remove.py create mode 100644 tests/unit_tests/test_proxy.py diff --git a/bittensor_cli/cli.py b/bittensor_cli/cli.py index 73e77736d..a8db475d7 100755 --- a/bittensor_cli/cli.py +++ b/bittensor_cli/cli.py @@ -94,6 +94,11 @@ subnets, mechanisms as subnet_mechanisms, ) +from bittensor_cli.src.commands.proxy import ( + add as add_proxy, + remove as remove_proxy, + list as list_proxy, +) from bittensor_cli.src.commands.wallets import SortByBalance from bittensor_cli.version import __version__, __version_as_int__ @@ -769,6 +774,7 @@ def __init__(self): self.liquidity_app = typer.Typer(epilog=_epilog) self.crowd_app = typer.Typer(epilog=_epilog) self.utils_app = typer.Typer(epilog=_epilog) + self.proxy_app = typer.Typer(epilog=_epilog) # config alias self.app.add_typer( @@ -867,6 +873,14 @@ def __init__(self): no_args_is_help=True, ) + # proxy app + self.app.add_typer( + self.proxy_app, + name="proxy", + short_help="Proxy commands for managing delegate accounts", + no_args_is_help=True, + ) + # config commands self.config_app.command("set")(self.set_config) self.config_app.command("get")(self.get_config) @@ -1228,6 +1242,12 @@ def __init__(self): self.utils_app.command("convert")(self.convert) self.utils_app.command("latency")(self.best_connection) + # proxy commands + self.proxy_app.command("add")(self.proxy_add) + self.proxy_app.command("remove")(self.proxy_remove) + self.proxy_app.command("remove-all")(self.proxy_remove_all) + self.proxy_app.command("list")(self.proxy_list) + def generate_command_tree(self) -> Tree: """ Generates a rich.Tree of the commands, subcommands, and groups of this app @@ -8150,6 +8170,193 @@ def best_connection( ) return True + # ======================== + # Proxy Commands + # ======================== + + def proxy_add( + self, + delegate: str = typer.Option( + ..., + "--delegate", + "-d", + help="SS58 address of the delegate account to add as proxy.", + ), + proxy_type: str = typer.Option( + "Any", + "--type", + "-t", + help="Type of proxy permissions. Options: Any, NonTransfer, Governance, Staking, " + "Registration, SenateVoting, Transfer, SmallTransfer, RootWeights, ChildKeys.", + ), + delay: int = typer.Option( + 0, + "--delay", + help="Block delay before proxy can execute (0 for immediate).", + ), + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + network: Optional[list[str]] = Options.network, + prompt: bool = Options.prompt, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + ): + """ + Add a proxy delegate to your account. + + A proxy allows another account to execute specific actions on behalf of your coldkey. + Different proxy types grant different levels of access. + + [bold]Examples:[/bold] + + 1. Add a full proxy (any action): + [green]$[/green] btcli proxy add --delegate 5xyz... --type Any + + 2. Add a staking-only proxy: + [green]$[/green] btcli proxy add --delegate 5xyz... --type Staking + + 3. Add a delayed proxy (24 hour delay = ~14400 blocks): + [green]$[/green] btcli proxy add --delegate 5xyz... --type Transfer --delay 14400 + + [bold]Proxy Types:[/bold] + • [blue]Any[/blue]: Full access to all operations + • [blue]NonTransfer[/blue]: All operations except balance transfers + • [blue]Staking[/blue]: Staking operations only + • [blue]Transfer[/blue]: Transfer operations only + • [blue]SmallTransfer[/blue]: Transfers under 0.5 TAO only + • [blue]Governance[/blue]: Governance/voting operations + """ + self.verbosity_handler(quiet, verbose) + wallet = self.wallet_ask(wallet_name, wallet_path, validate=WV.WALLET_AND_COLDKEY) + + return self._run_command( + add_proxy.proxy_add( + wallet=wallet, + subtensor=self.initialize_chain(network), + delegate=delegate, + proxy_type=proxy_type, + delay=delay, + prompt=prompt, + ) + ) + + def proxy_remove( + self, + delegate: str = typer.Option( + ..., + "--delegate", + "-d", + help="SS58 address of the delegate account to remove.", + ), + proxy_type: str = typer.Option( + "Any", + "--type", + "-t", + help="Type of proxy permissions to remove.", + ), + delay: int = typer.Option( + 0, + "--delay", + help="Block delay that was set for the proxy.", + ), + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + network: Optional[list[str]] = Options.network, + prompt: bool = Options.prompt, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + ): + """ + Remove a proxy delegate from your account. + + [bold]Example:[/bold] + [green]$[/green] btcli proxy remove --delegate 5xyz... --type Any + """ + self.verbosity_handler(quiet, verbose) + wallet = self.wallet_ask(wallet_name, wallet_path, validate=WV.WALLET_AND_COLDKEY) + + return self._run_command( + remove_proxy.proxy_remove( + wallet=wallet, + subtensor=self.initialize_chain(network), + delegate=delegate, + proxy_type=proxy_type, + delay=delay, + prompt=prompt, + ) + ) + + def proxy_remove_all( + self, + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + network: Optional[list[str]] = Options.network, + prompt: bool = Options.prompt, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + ): + """ + Remove ALL proxies from your account. + + [bold red]WARNING:[/bold red] This removes all delegate access to your account. + + [bold]Example:[/bold] + [green]$[/green] btcli proxy remove-all + """ + self.verbosity_handler(quiet, verbose) + wallet = self.wallet_ask(wallet_name, wallet_path, validate=WV.WALLET_AND_COLDKEY) + + return self._run_command( + remove_proxy.proxy_remove_all( + wallet=wallet, + subtensor=self.initialize_chain(network), + prompt=prompt, + ) + ) + + def proxy_list( + self, + address: Optional[str] = typer.Option( + None, + "--address", + "-a", + help="SS58 address to query. If not provided, uses wallet's coldkey.", + ), + wallet_name: str = Options.wallet_name, + wallet_path: str = Options.wallet_path, + network: Optional[list[str]] = Options.network, + quiet: bool = Options.quiet, + verbose: bool = Options.verbose, + ): + """ + List all proxies for an account. + + [bold]Examples:[/bold] + + 1. List proxies for your wallet: + [green]$[/green] btcli proxy list + + 2. List proxies for any address: + [green]$[/green] btcli proxy list --address 5xyz... + """ + self.verbosity_handler(quiet, verbose) + + if address: + return self._run_command( + list_proxy.proxy_list( + subtensor=self.initialize_chain(network), + address=address, + ) + ) + else: + wallet = self.wallet_ask(wallet_name, wallet_path, validate=WV.WALLET) + return self._run_command( + list_proxy.proxy_list_for_wallet( + wallet=wallet, + subtensor=self.initialize_chain(network), + ) + ) + def run(self): self.app() diff --git a/bittensor_cli/src/commands/proxy/__init__.py b/bittensor_cli/src/commands/proxy/__init__.py new file mode 100644 index 000000000..3dc2f67fb --- /dev/null +++ b/bittensor_cli/src/commands/proxy/__init__.py @@ -0,0 +1 @@ +# Proxy commands for btcli diff --git a/bittensor_cli/src/commands/proxy/add.py b/bittensor_cli/src/commands/proxy/add.py new file mode 100644 index 000000000..98a3fa393 --- /dev/null +++ b/bittensor_cli/src/commands/proxy/add.py @@ -0,0 +1,172 @@ +""" +Proxy Add Command - Add a proxy delegate to an account. + +This command allows adding a proxy account that can execute permitted +calls on behalf of the real account. +""" + +import asyncio +from typing import TYPE_CHECKING, Optional + +from rich.prompt import Confirm +from async_substrate_interface.errors import SubstrateRequestException + +from bittensor_cli.src.bittensor.balances import Balance +from bittensor_cli.src.bittensor.utils import ( + console, + err_console, + format_error_message, + is_valid_ss58_address, + print_error, + print_verbose, + unlock_key, +) +from bittensor_wallet import Wallet + +if TYPE_CHECKING: + from bittensor_cli.src.bittensor.subtensor_interface import SubtensorInterface + + +# Proxy types supported by the Bittensor chain +PROXY_TYPES = [ + "Any", + "NonTransfer", + "Governance", + "Staking", + "Registration", + "SenateVoting", + "Transfer", + "SmallTransfer", + "RootWeights", + "ChildKeys", + "SudoUncheckedSetCode", +] + + +async def proxy_add( + wallet: Wallet, + subtensor: "SubtensorInterface", + delegate: str, + proxy_type: str, + delay: int = 0, + prompt: bool = True, + era: int = 64, +) -> bool: + """ + Add a proxy delegate to the wallet's account. + + Args: + wallet: The wallet object (will be the 'real' account) + subtensor: SubtensorInterface object + delegate: SS58 address of the delegate account + proxy_type: Type of proxy permissions to grant + delay: Block delay before proxy can execute (0 for immediate) + prompt: Whether to prompt for confirmation + era: Blocks for which the transaction should be valid + + Returns: + bool: True if proxy was added successfully, False otherwise + """ + + async def get_add_proxy_fee() -> Balance: + """Calculate the transaction fee for adding a proxy.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="add_proxy", + call_params={ + "delegate": delegate, + "proxy_type": proxy_type, + "delay": delay, + }, + ) + try: + payment_info = await subtensor.substrate.get_payment_info( + call=call, keypair=wallet.coldkeypub + ) + except SubstrateRequestException as e: + payment_info = {"partial_fee": int(2e7)} # assume 0.02 Tao + err_console.print( + f":cross_mark: [red]Failed to get payment info[/red]:[bold white]\n" + f" {format_error_message(e)}[/bold white]\n" + f" Defaulting to default fee: {Balance.from_rao(payment_info['partial_fee'])}" + ) + return Balance.from_rao(payment_info["partial_fee"]) + + async def do_add_proxy() -> tuple[bool, str, str]: + """Execute the add_proxy extrinsic.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="add_proxy", + call_params={ + "delegate": delegate, + "proxy_type": proxy_type, + "delay": delay, + }, + ) + extrinsic = await subtensor.substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey, era={"period": era} + ) + response = await subtensor.substrate.submit_extrinsic( + extrinsic, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + if await response.is_success: + return True, response.block_hash, "" + else: + return False, "", format_error_message(await response.error_message) + + # Validate delegate address + if not is_valid_ss58_address(delegate): + err_console.print( + f":cross_mark: [red]Invalid delegate SS58 address[/red]:[bold white]\n {delegate}[/bold white]" + ) + return False + + # Validate proxy type + if proxy_type not in PROXY_TYPES: + err_console.print( + f":cross_mark: [red]Invalid proxy type[/red]: {proxy_type}\n" + f" Valid types: {', '.join(PROXY_TYPES)}" + ) + return False + + console.print(f"[dark_orange]Adding proxy on network: {subtensor.network}") + + # Unlock wallet coldkey + if not unlock_key(wallet).success: + return False + + # Get and display fee + with console.status("[bold green]Calculating transaction fee..."): + fee = await get_add_proxy_fee() + + console.print( + f"\n[bold]Add Proxy Details:[/bold]\n" + f" Real Account: [cyan]{wallet.coldkeypub.ss58_address}[/cyan]\n" + f" Delegate: [cyan]{delegate}[/cyan]\n" + f" Proxy Type: [yellow]{proxy_type}[/yellow]\n" + f" Delay: [yellow]{delay}[/yellow] blocks\n" + f" Estimated Fee: [green]{fee}[/green]\n" + ) + + if prompt: + if not Confirm.ask("Do you want to proceed?"): + console.print("[yellow]Cancelled.[/yellow]") + return False + + with console.status("[bold green]Adding proxy..."): + success, block_hash, error_msg = await do_add_proxy() + + if success: + console.print( + f":white_check_mark: [green]Proxy added successfully![/green]\n" + f" Block Hash: [cyan]{block_hash}[/cyan]" + ) + return True + else: + err_console.print( + f":cross_mark: [red]Failed to add proxy[/red]:\n {error_msg}" + ) + return False diff --git a/bittensor_cli/src/commands/proxy/list.py b/bittensor_cli/src/commands/proxy/list.py new file mode 100644 index 000000000..be040b2a5 --- /dev/null +++ b/bittensor_cli/src/commands/proxy/list.py @@ -0,0 +1,156 @@ +""" +Proxy List Command - List all proxies for an account. + +This command queries and displays all proxy relationships for a given account. +""" + +import asyncio +from typing import TYPE_CHECKING, Optional + +from rich.table import Table +from async_substrate_interface.errors import SubstrateRequestException + +from bittensor_cli.src.bittensor.utils import ( + console, + err_console, + format_error_message, + is_valid_ss58_address, +) +from bittensor_wallet import Wallet + +if TYPE_CHECKING: + from bittensor_cli.src.bittensor.subtensor_interface import SubtensorInterface + + +async def proxy_list( + subtensor: "SubtensorInterface", + address: str, +) -> bool: + """ + List all proxies for a given account. + + Args: + subtensor: SubtensorInterface object + address: SS58 address to query proxies for + + Returns: + bool: True if query was successful, False otherwise + """ + + # Validate address + if not is_valid_ss58_address(address): + err_console.print( + f":cross_mark: [red]Invalid SS58 address[/red]:[bold white]\n {address}[/bold white]" + ) + return False + + console.print(f"[dark_orange]Querying proxies on network: {subtensor.network}") + + try: + with console.status("[bold green]Fetching proxies..."): + # Query the Proxy.Proxies storage + result = await subtensor.substrate.query( + module="Proxy", + storage_function="Proxies", + params=[address], + ) + + if result is None: + console.print(f"\n[yellow]No proxies found for account:[/yellow] {address}") + return True + + # Parse the result - it returns a tuple of (Vec, Balance) + proxies_data = result.value if hasattr(result, 'value') else result + + # Handle different response formats + proxies = [] + deposit = 0 + + if isinstance(proxies_data, (list, tuple)): + if len(proxies_data) >= 1: + # First element should be the list of proxies + if isinstance(proxies_data[0], list): + proxies = proxies_data[0] + elif isinstance(proxies_data[0], dict): + # Single proxy as dict + proxies = [proxies_data[0]] + # Second element is deposit if present + if len(proxies_data) > 1 and isinstance(proxies_data[1], int): + deposit = proxies_data[1] + elif isinstance(proxies_data, dict): + # Single proxy returned as dict + proxies = [proxies_data] + + if not proxies: + console.print(f"\n[yellow]No proxies found for account:[/yellow] {address}") + return True + + # Create table for display + table = Table( + title=f"Proxies for {address[:8]}...{address[-8:]}", + show_header=True, + header_style="bold magenta", + ) + table.add_column("Delegate", style="cyan") + table.add_column("Proxy Type", style="yellow") + table.add_column("Delay (blocks)", style="green", justify="right") + + for proxy in proxies: + if isinstance(proxy, dict): + delegate = proxy.get("delegate", "Unknown") + proxy_type = proxy.get("proxy_type", proxy.get("proxyType", "Unknown")) + delay = proxy.get("delay", 0) + else: + # Handle tuple format + delegate = proxy[0] if len(proxy) > 0 else "Unknown" + proxy_type = proxy[1] if len(proxy) > 1 else "Unknown" + delay = proxy[2] if len(proxy) > 2 else 0 + + # Convert proxy_type if it's a dict/enum + if isinstance(proxy_type, dict): + proxy_type = list(proxy_type.keys())[0] if proxy_type else "Unknown" + + table.add_row( + str(delegate), + str(proxy_type), + str(delay), + ) + + console.print() + console.print(table) + + # Show deposit if available + if deposit and deposit > 0: + from bittensor_cli.src.bittensor.balances import Balance + deposit_balance = Balance.from_rao(deposit) + console.print(f"\n[dim]Reserved deposit: {deposit_balance}[/dim]") + + return True + + except SubstrateRequestException as e: + err_console.print( + f":cross_mark: [red]Failed to query proxies[/red]:\n {format_error_message(e)}" + ) + return False + except Exception as e: + err_console.print( + f":cross_mark: [red]Error querying proxies[/red]:\n {str(e)}" + ) + return False + + +async def proxy_list_for_wallet( + wallet: Wallet, + subtensor: "SubtensorInterface", +) -> bool: + """ + List all proxies for the wallet's coldkey account. + + Args: + wallet: The wallet object + subtensor: SubtensorInterface object + + Returns: + bool: True if query was successful, False otherwise + """ + return await proxy_list(subtensor, wallet.coldkeypub.ss58_address) diff --git a/bittensor_cli/src/commands/proxy/remove.py b/bittensor_cli/src/commands/proxy/remove.py new file mode 100644 index 000000000..806f08549 --- /dev/null +++ b/bittensor_cli/src/commands/proxy/remove.py @@ -0,0 +1,269 @@ +""" +Proxy Remove Command - Remove a proxy delegate from an account. + +This command allows removing a previously added proxy account. +""" + +import asyncio +from typing import TYPE_CHECKING, Optional + +from rich.prompt import Confirm +from async_substrate_interface.errors import SubstrateRequestException + +from bittensor_cli.src.bittensor.balances import Balance +from bittensor_cli.src.bittensor.utils import ( + console, + err_console, + format_error_message, + is_valid_ss58_address, + print_error, + print_verbose, + unlock_key, +) +from bittensor_wallet import Wallet + +if TYPE_CHECKING: + from bittensor_cli.src.bittensor.subtensor_interface import SubtensorInterface + + +# Proxy types supported by the Bittensor chain +PROXY_TYPES = [ + "Any", + "NonTransfer", + "Governance", + "Staking", + "Registration", + "SenateVoting", + "Transfer", + "SmallTransfer", + "RootWeights", + "ChildKeys", + "SudoUncheckedSetCode", +] + + +async def proxy_remove( + wallet: Wallet, + subtensor: "SubtensorInterface", + delegate: str, + proxy_type: str, + delay: int = 0, + prompt: bool = True, + era: int = 64, +) -> bool: + """ + Remove a proxy delegate from the wallet's account. + + Args: + wallet: The wallet object (the 'real' account) + subtensor: SubtensorInterface object + delegate: SS58 address of the delegate account to remove + proxy_type: Type of proxy permissions to remove + delay: Block delay that was set for the proxy + prompt: Whether to prompt for confirmation + era: Blocks for which the transaction should be valid + + Returns: + bool: True if proxy was removed successfully, False otherwise + """ + + async def get_remove_proxy_fee() -> Balance: + """Calculate the transaction fee for removing a proxy.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="remove_proxy", + call_params={ + "delegate": delegate, + "proxy_type": proxy_type, + "delay": delay, + }, + ) + try: + payment_info = await subtensor.substrate.get_payment_info( + call=call, keypair=wallet.coldkeypub + ) + except SubstrateRequestException as e: + payment_info = {"partial_fee": int(2e7)} # assume 0.02 Tao + err_console.print( + f":cross_mark: [red]Failed to get payment info[/red]:[bold white]\n" + f" {format_error_message(e)}[/bold white]\n" + f" Defaulting to default fee: {Balance.from_rao(payment_info['partial_fee'])}" + ) + return Balance.from_rao(payment_info["partial_fee"]) + + async def do_remove_proxy() -> tuple[bool, str, str]: + """Execute the remove_proxy extrinsic.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="remove_proxy", + call_params={ + "delegate": delegate, + "proxy_type": proxy_type, + "delay": delay, + }, + ) + extrinsic = await subtensor.substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey, era={"period": era} + ) + response = await subtensor.substrate.submit_extrinsic( + extrinsic, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + if await response.is_success: + return True, response.block_hash, "" + else: + return False, "", format_error_message(await response.error_message) + + # Validate delegate address + if not is_valid_ss58_address(delegate): + err_console.print( + f":cross_mark: [red]Invalid delegate SS58 address[/red]:[bold white]\n {delegate}[/bold white]" + ) + return False + + # Validate proxy type + if proxy_type not in PROXY_TYPES: + err_console.print( + f":cross_mark: [red]Invalid proxy type[/red]: {proxy_type}\n" + f" Valid types: {', '.join(PROXY_TYPES)}" + ) + return False + + console.print(f"[dark_orange]Removing proxy on network: {subtensor.network}") + + # Unlock wallet coldkey + if not unlock_key(wallet).success: + return False + + # Get and display fee + with console.status("[bold green]Calculating transaction fee..."): + fee = await get_remove_proxy_fee() + + console.print( + f"\n[bold]Remove Proxy Details:[/bold]\n" + f" Real Account: [cyan]{wallet.coldkeypub.ss58_address}[/cyan]\n" + f" Delegate: [cyan]{delegate}[/cyan]\n" + f" Proxy Type: [yellow]{proxy_type}[/yellow]\n" + f" Delay: [yellow]{delay}[/yellow] blocks\n" + f" Estimated Fee: [green]{fee}[/green]\n" + ) + + if prompt: + if not Confirm.ask("Do you want to proceed?"): + console.print("[yellow]Cancelled.[/yellow]") + return False + + with console.status("[bold green]Removing proxy..."): + success, block_hash, error_msg = await do_remove_proxy() + + if success: + console.print( + f":white_check_mark: [green]Proxy removed successfully![/green]\n" + f" Block Hash: [cyan]{block_hash}[/cyan]" + ) + return True + else: + err_console.print( + f":cross_mark: [red]Failed to remove proxy[/red]:\n {error_msg}" + ) + return False + + +async def proxy_remove_all( + wallet: Wallet, + subtensor: "SubtensorInterface", + prompt: bool = True, + era: int = 64, +) -> bool: + """ + Remove all proxies from the wallet's account. + + Args: + wallet: The wallet object (the 'real' account) + subtensor: SubtensorInterface object + prompt: Whether to prompt for confirmation + era: Blocks for which the transaction should be valid + + Returns: + bool: True if all proxies were removed successfully, False otherwise + """ + + async def get_remove_all_fee() -> Balance: + """Calculate the transaction fee for removing all proxies.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="remove_proxies", + call_params={}, + ) + try: + payment_info = await subtensor.substrate.get_payment_info( + call=call, keypair=wallet.coldkeypub + ) + except SubstrateRequestException as e: + payment_info = {"partial_fee": int(2e7)} + err_console.print( + f":cross_mark: [red]Failed to get payment info[/red]:[bold white]\n" + f" {format_error_message(e)}[/bold white]\n" + f" Defaulting to default fee: {Balance.from_rao(payment_info['partial_fee'])}" + ) + return Balance.from_rao(payment_info["partial_fee"]) + + async def do_remove_all() -> tuple[bool, str, str]: + """Execute the remove_proxies extrinsic.""" + call = await subtensor.substrate.compose_call( + call_module="Proxy", + call_function="remove_proxies", + call_params={}, + ) + extrinsic = await subtensor.substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey, era={"period": era} + ) + response = await subtensor.substrate.submit_extrinsic( + extrinsic, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + if await response.is_success: + return True, response.block_hash, "" + else: + return False, "", format_error_message(await response.error_message) + + console.print(f"[dark_orange]Removing all proxies on network: {subtensor.network}") + + # Unlock wallet coldkey + if not unlock_key(wallet).success: + return False + + # Get and display fee + with console.status("[bold green]Calculating transaction fee..."): + fee = await get_remove_all_fee() + + console.print( + f"\n[bold]Remove All Proxies:[/bold]\n" + f" Account: [cyan]{wallet.coldkeypub.ss58_address}[/cyan]\n" + f" Estimated Fee: [green]{fee}[/green]\n" + f" [red]WARNING: This will remove ALL proxy delegates from this account![/red]\n" + ) + + if prompt: + if not Confirm.ask("Are you sure you want to remove ALL proxies?"): + console.print("[yellow]Cancelled.[/yellow]") + return False + + with console.status("[bold green]Removing all proxies..."): + success, block_hash, error_msg = await do_remove_all() + + if success: + console.print( + f":white_check_mark: [green]All proxies removed successfully![/green]\n" + f" Block Hash: [cyan]{block_hash}[/cyan]" + ) + return True + else: + err_console.print( + f":cross_mark: [red]Failed to remove all proxies[/red]:\n {error_msg}" + ) + return False diff --git a/tests/unit_tests/test_proxy.py b/tests/unit_tests/test_proxy.py new file mode 100644 index 000000000..d5a9bc129 --- /dev/null +++ b/tests/unit_tests/test_proxy.py @@ -0,0 +1,299 @@ +""" +Unit tests for proxy commands. +""" + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from bittensor_cli.src.commands.proxy.add import proxy_add, PROXY_TYPES +from bittensor_cli.src.commands.proxy.remove import proxy_remove, proxy_remove_all +from bittensor_cli.src.commands.proxy.list import proxy_list, proxy_list_for_wallet + + +class TestProxyTypes: + """Test that all expected proxy types are defined.""" + + def test_proxy_types_defined(self): + """Verify all expected proxy types are available.""" + expected_types = [ + "Any", + "NonTransfer", + "Governance", + "Staking", + "Registration", + "SenateVoting", + "Transfer", + "SmallTransfer", + "RootWeights", + "ChildKeys", + "SudoUncheckedSetCode", + ] + assert PROXY_TYPES == expected_types + + +class TestProxyAdd: + """Tests for proxy_add command.""" + + @pytest.mark.asyncio + async def test_proxy_add_invalid_delegate_address(self): + """Test that invalid SS58 address is rejected.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + + with patch( + "bittensor_cli.src.commands.proxy.add.is_valid_ss58_address", + return_value=False, + ): + result = await proxy_add( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="invalid_address", + proxy_type="Any", + delay=0, + prompt=False, + ) + + assert result is False + + @pytest.mark.asyncio + async def test_proxy_add_invalid_proxy_type(self): + """Test that invalid proxy type is rejected.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + + with patch( + "bittensor_cli.src.commands.proxy.add.is_valid_ss58_address", + return_value=True, + ): + result = await proxy_add( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + proxy_type="InvalidType", + delay=0, + prompt=False, + ) + + assert result is False + + @pytest.mark.asyncio + async def test_proxy_add_wallet_unlock_failure(self): + """Test that failed wallet unlock returns False.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + mock_unlock_result = MagicMock() + mock_unlock_result.success = False + + with ( + patch( + "bittensor_cli.src.commands.proxy.add.is_valid_ss58_address", + return_value=True, + ), + patch( + "bittensor_cli.src.commands.proxy.add.unlock_key", + return_value=mock_unlock_result, + ), + ): + result = await proxy_add( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + proxy_type="Any", + delay=0, + prompt=False, + ) + + assert result is False + + +class TestProxyRemove: + """Tests for proxy_remove command.""" + + @pytest.mark.asyncio + async def test_proxy_remove_invalid_delegate_address(self): + """Test that invalid SS58 address is rejected.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + + with patch( + "bittensor_cli.src.commands.proxy.remove.is_valid_ss58_address", + return_value=False, + ): + result = await proxy_remove( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="invalid_address", + proxy_type="Any", + delay=0, + prompt=False, + ) + + assert result is False + + @pytest.mark.asyncio + async def test_proxy_remove_invalid_proxy_type(self): + """Test that invalid proxy type is rejected.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + + with patch( + "bittensor_cli.src.commands.proxy.remove.is_valid_ss58_address", + return_value=True, + ): + result = await proxy_remove( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + proxy_type="InvalidType", + delay=0, + prompt=False, + ) + + assert result is False + + @pytest.mark.asyncio + async def test_proxy_remove_wallet_unlock_failure(self): + """Test that failed wallet unlock returns False.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + mock_unlock_result = MagicMock() + mock_unlock_result.success = False + + with ( + patch( + "bittensor_cli.src.commands.proxy.remove.is_valid_ss58_address", + return_value=True, + ), + patch( + "bittensor_cli.src.commands.proxy.remove.unlock_key", + return_value=mock_unlock_result, + ), + ): + result = await proxy_remove( + wallet=mock_wallet, + subtensor=mock_subtensor, + delegate="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + proxy_type="Any", + delay=0, + prompt=False, + ) + + assert result is False + + +class TestProxyRemoveAll: + """Tests for proxy_remove_all command.""" + + @pytest.mark.asyncio + async def test_proxy_remove_all_wallet_unlock_failure(self): + """Test that failed wallet unlock returns False.""" + mock_wallet = MagicMock() + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + mock_unlock_result = MagicMock() + mock_unlock_result.success = False + + with patch( + "bittensor_cli.src.commands.proxy.remove.unlock_key", + return_value=mock_unlock_result, + ): + result = await proxy_remove_all( + wallet=mock_wallet, + subtensor=mock_subtensor, + prompt=False, + ) + + assert result is False + + +class TestProxyList: + """Tests for proxy_list command.""" + + @pytest.mark.asyncio + async def test_proxy_list_invalid_address(self): + """Test that invalid SS58 address is rejected.""" + mock_subtensor = MagicMock() + + with patch( + "bittensor_cli.src.commands.proxy.list.is_valid_ss58_address", + return_value=False, + ): + result = await proxy_list( + subtensor=mock_subtensor, + address="invalid_address", + ) + + assert result is False + + @pytest.mark.asyncio + async def test_proxy_list_no_proxies(self): + """Test listing when no proxies exist.""" + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + mock_subtensor.substrate.query = AsyncMock(return_value=None) + + with patch( + "bittensor_cli.src.commands.proxy.list.is_valid_ss58_address", + return_value=True, + ): + result = await proxy_list( + subtensor=mock_subtensor, + address="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + ) + + assert result is True + + @pytest.mark.asyncio + async def test_proxy_list_with_proxies(self): + """Test listing when proxies exist.""" + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + + # Mock result with proxy data + mock_result = MagicMock() + mock_result.value = ( + [ + { + "delegate": "5FHneW46xGXgs5mUiveU4sbTyGBzmstUspZC92UhjJM694ty", + "proxy_type": {"Staking": None}, + "delay": 0, + } + ], + 1000000000, # deposit in rao + ) + mock_subtensor.substrate.query = AsyncMock(return_value=mock_result) + + with patch( + "bittensor_cli.src.commands.proxy.list.is_valid_ss58_address", + return_value=True, + ): + result = await proxy_list( + subtensor=mock_subtensor, + address="5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + ) + + assert result is True + + @pytest.mark.asyncio + async def test_proxy_list_for_wallet(self): + """Test listing proxies for wallet's coldkey.""" + mock_wallet = MagicMock() + mock_wallet.coldkeypub.ss58_address = ( + "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY" + ) + mock_subtensor = MagicMock() + mock_subtensor.network = "finney" + mock_subtensor.substrate.query = AsyncMock(return_value=None) + + with patch( + "bittensor_cli.src.commands.proxy.list.is_valid_ss58_address", + return_value=True, + ): + result = await proxy_list_for_wallet( + wallet=mock_wallet, + subtensor=mock_subtensor, + ) + + assert result is True