From d21c21ab63c33bd88798fc021fca72d0f665f1e1 Mon Sep 17 00:00:00 2001 From: mkdev11 Date: Mon, 23 Mar 2026 23:50:05 +0100 Subject: [PATCH 1/2] feat(proxy): add list and reject commands (#742) - _parse_proxy_storage + list_proxies: query Proxy.Proxies, display table or JSON - reject_announcement: submit Proxy.reject_announcement with DB integration - CLI: register proxy list and proxy reject commands - Adapt proxy_remove to upstream all_ parameter - Tests: 7 parse_proxy, list/reject function, CLI handler - E2E: test_proxy_list --- bittensor_cli/cli.py | 282 ++++++++++++++ bittensor_cli/src/commands/proxy.py | 214 ++++++++++- tests/e2e_tests/test_proxy.py | 108 ++++++ tests/unit_tests/test_cli.py | 554 ++++++++++++++++++++++++++++ 4 files changed, 1156 insertions(+), 2 deletions(-) diff --git a/bittensor_cli/cli.py b/bittensor_cli/cli.py index ebf4d29d8..6e2b8a3b7 100755 --- a/bittensor_cli/cli.py +++ b/bittensor_cli/cli.py @@ -1243,6 +1243,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 @@ -9892,6 +9898,7 @@ def proxy_remove( delegate = is_valid_ss58_address_param(delegate) self.verbosity_handler(quiet, verbose, json_output, prompt) + wallet = self.wallet_ask( wallet_name=wallet_name, wallet_path=wallet_path, @@ -10226,6 +10233,281 @@ 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.WALLET, + ) + 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 (proxy) who made the announcement.", + ), + ] = 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, + ) + + if not delegate: + if prompt: + delegate = Prompt.ask( + "Enter the SS58 address of the delegate (proxy) who made the announcement" + ) + if not is_valid_ss58_address(delegate): + print_error(f"Invalid SS58 address: {delegate}") + return + else: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": "--delegate is required. Provide the SS58 address of the proxy that made the announcement.", + "extrinsic_identifier": None, + } + ) + else: + print_error( + "--delegate is required. Provide the SS58 address of the proxy that made the announcement." + ) + return + + # Try to find the announcement in the local DB + # DB stores address = the real account (the wallet calling reject) + real_address = wallet.coldkeypub.ss58_address + 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 == real_address 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": "No pending announcements found in the local address book. Please provide --call-hash explicitly.", + "extrinsic_identifier": None, + } + ) + else: + print_error( + "No pending announcements found in the local address book. " + "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] + if not json_output: + 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": "Multiple pending announcements found. Please provide --call-hash explicitly.", + "extrinsic_identifier": None, + } + ) + else: + print_error( + "Multiple pending announcements found. " + 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: + print_error("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 or f"0x{call_hash_}" == call_hash) + and address == real_address + 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 67c00bd0c..06b269a27 100644 --- a/bittensor_cli/src/commands/proxy.py +++ b/bittensor_cli/src/commands/proxy.py @@ -1,14 +1,16 @@ -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Any, Optional import sys from async_substrate_interface.errors import StateDiscardedError from rich.prompt import Prompt, FloatPrompt, IntPrompt +from rich.table import Table from scalecodec import GenericCall, ScaleBytes from bittensor_cli.src import COLORS from bittensor_cli.src.bittensor.balances import Balance from bittensor_cli.src.bittensor.utils import ( confirm_action, + decode_account_id, print_extrinsic_id, json_console, console, @@ -55,7 +57,215 @@ class ProxyType(StrEnum): RootClaim = "RootClaim" -# TODO add announce with also --reject and --remove +def _parse_proxy_storage(raw: Any) -> tuple[list[dict[str, Any]], Any]: + """Parse the Proxy.Proxies storage value into a list of proxy dicts and a deposit. + + The chain returns ``(Vec, Balance)`` where each proxy + definition may be a dict (``{"delegate": ..., "proxy_type": ..., "delay": ...}``) + or a positional tuple ``(account_id, proxy_type, delay)``. Account IDs + can arrive as nested byte-tuples that need unwrapping before SS58-encoding. + """ + if raw is None: + return [], None + if not isinstance(raw, (list, tuple)) or len(raw) < 1: + return [], None + + proxies_raw = raw[0] + deposit = raw[1] if len(raw) > 1 else None + + if not isinstance(proxies_raw, (list, tuple)): + return [], deposit + + rows: list[dict[str, Any]] = [] + for item in proxies_raw: + try: + # Unwrap single-element wrapper tuples produced by substrate + while ( + isinstance(item, (list, tuple)) + and len(item) == 1 + and isinstance(item[0], (dict, list, tuple)) + ): + item = item[0] + + if isinstance(item, dict): + delegate_raw = item.get("delegate") or item.get("delegatee") + ptype = item.get("proxy_type", "") + delay = item.get("delay", 0) + elif isinstance(item, (list, tuple)) and len(item) >= 3: + delegate_raw, ptype, delay = item[0], item[1], item[2] + else: + continue + + # Unwrap nested delegate tuple, e.g. ((48, 103, ...),) -> (48, 103, ...) + while ( + isinstance(delegate_raw, (list, tuple)) + and len(delegate_raw) == 1 + and isinstance(delegate_raw[0], (list, tuple)) + ): + delegate_raw = delegate_raw[0] + if isinstance(delegate_raw, list): + delegate_raw = tuple(delegate_raw) + + # Convert to SS58 + if isinstance(delegate_raw, str) and delegate_raw.startswith("5"): + delegate_ss58 = delegate_raw + else: + delegate_ss58 = decode_account_id( + delegate_raw if isinstance(delegate_raw, tuple) else (delegate_raw,) + ) + + # Normalise proxy type + if isinstance(ptype, dict): + proxy_type_str = next(iter(ptype), "") + elif isinstance(ptype, str): + proxy_type_str = ptype + else: + proxy_type_str = getattr(ptype, "value", str(ptype)) + + rows.append( + { + "delegate": delegate_ss58, + "proxy_type": proxy_type_str, + "delay": int(delay) if delay is not None else 0, + } + ) + except (KeyError, TypeError, ValueError, IndexError): + continue + + return rows, deposit + + +async def list_proxies( + subtensor: "SubtensorInterface", + address: str, + json_output: bool, +) -> None: + """Query ``Proxy.Proxies`` storage for *address* and display the result.""" + try: + raw = await subtensor.query( + module="Proxy", + storage_function="Proxies", + params=[address], + ) + except Exception as e: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": str(e), + "address": address, + "proxies": [], + "deposit": None, + } + ) + else: + print_error(f"Failed to query proxies: {e}") + return + + rows, deposit = _parse_proxy_storage(raw) + deposit_val = deposit.value if hasattr(deposit, "value") else deposit + + if json_output: + json_console.print_json( + data={ + "success": True, + "address": address, + "proxies": rows, + "deposit": deposit_val, + } + ) + return + + if not rows: + console.print("No proxies configured for this account.") + return + + table = Table(title=f"Proxies for {address}") + table.add_column("Delegate", style="cyan") + table.add_column("Proxy Type", style="green") + table.add_column("Delay", style="yellow") + for r in rows: + table.add_row(r["delegate"], r["proxy_type"], str(r["delay"])) + if deposit_val is not None: + table.caption = f"Total deposit: {deposit_val}" + console.print(table) + + +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: + """Submit ``Proxy.reject_announcement``. Returns ``True`` on success.""" + if prompt: + if not confirm_action( + f"Reject the announced call from delegate {delegate}?", + decline=decline, + quiet=quiet, + ): + if json_output: + json_console.print_json( + data={ + "success": False, + "message": "Cancelled", + "extrinsic_identifier": None, + } + ) + return False + + if not (ulw := unlock_key(wallet, print_out=not json_output)).success: + if json_output: + json_console.print_json( + data={ + "success": False, + "message": ulw.message, + "extrinsic_identifier": None, + } + ) + else: + print_error(ulw.message) + 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: + if json_output: + json_console.print_json( + data={ + "success": True, + "message": msg, + "extrinsic_identifier": await receipt.get_extrinsic_identifier(), + } + ) + else: + await print_extrinsic_id(receipt) + print_success("Success!") + else: + if json_output: + json_console.print_json( + data={"success": False, "message": msg, "extrinsic_identifier": None} + ) + else: + print_error(f"Failed: {msg}") + return success async def submit_proxy( diff --git a/tests/e2e_tests/test_proxy.py b/tests/e2e_tests/test_proxy.py index 092daa0e8..1faa71bfd 100644 --- a/tests/e2e_tests/test_proxy.py +++ b/tests/e2e_tests/test_proxy.py @@ -1137,3 +1137,111 @@ def test_remove_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 5b73fd68c..cf9bebf86 100644 --- a/tests/unit_tests/test_cli.py +++ b/tests/unit_tests/test_cli.py @@ -12,6 +12,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 ( + _parse_proxy_storage, + list_proxies, + reject_announcement, +) def test_parse_mnemonic(): @@ -768,6 +773,555 @@ 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" message + mock_console.print.assert_called_once() + assert "No proxies configured" 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 query proxies" in str(mock_print_error.call_args) + + +# ============================================================================ +# 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.print_success") as mock_print_success, + patch("bittensor_cli.src.commands.proxy.print_extrinsic_id"), + ): + mock_unlock.return_value = MagicMock(success=True) + + result = 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, + ) + + assert result is True + # 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", + }, + ) + mock_print_success.assert_called_once() + + +@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["extrinsic_identifier"] == "12345-1" + + +@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" in str(mock_print_error.call_args) + + +# ============================================================================ +# 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, + decline=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() + + +def test_proxy_reject_requires_delegate_no_prompt(): + """Test that proxy_reject errors when --delegate is not provided and prompt is disabled""" + cli_manager = CLIManager() + + with ( + patch.object(cli_manager, "verbosity_handler"), + patch.object(cli_manager, "wallet_ask") as mock_wallet_ask, + patch("bittensor_cli.cli.print_error") as mock_print_error, + ): + mock_wallet = Mock() + mock_wallet.coldkeypub = Mock() + mock_wallet.coldkeypub.ss58_address = "5GWallet..." + mock_wallet_ask.return_value = mock_wallet + + result = cli_manager.proxy_reject( + delegate=None, + call_hash="0x1234abcd", + network=None, + wallet_name="test", + wallet_path="/tmp/test", + wallet_hotkey="test", + prompt=False, + decline=False, + wait_for_inclusion=True, + wait_for_finalization=True, + period=16, + quiet=False, + verbose=False, + json_output=False, + ) + + assert result is None + mock_print_error.assert_called_once() + assert "--delegate is required" in str(mock_print_error.call_args) + + +# ============================================================================ +# Tests for _parse_proxy_storage helper +# ============================================================================ + + +def test_parse_proxy_storage_returns_empty_for_none(): + """_parse_proxy_storage returns empty list when raw is None""" + rows, deposit = _parse_proxy_storage(None) + assert rows == [] + assert deposit is None + + +def test_parse_proxy_storage_returns_empty_for_non_sequence(): + """_parse_proxy_storage returns empty list for non-sequence input""" + rows, deposit = _parse_proxy_storage("unexpected") + assert rows == [] + assert deposit is None + + +def test_parse_proxy_storage_parses_dict_format(): + """_parse_proxy_storage handles dict-format proxy definitions""" + raw = ( + [ + { + "delegate": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + "proxy_type": "Any", + "delay": 0, + }, + ], + 1000, + ) + rows, deposit = _parse_proxy_storage(raw) + assert len(rows) == 1 + assert rows[0]["delegate"] == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY" + assert rows[0]["proxy_type"] == "Any" + assert rows[0]["delay"] == 0 + assert deposit == 1000 + + +def test_parse_proxy_storage_handles_dict_proxy_type(): + """_parse_proxy_storage extracts key from dict-style proxy_type like {'Any': ()}""" + raw = ( + [ + { + "delegate": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + "proxy_type": {"Any": ()}, + "delay": 5, + }, + ], + 500, + ) + rows, deposit = _parse_proxy_storage(raw) + assert len(rows) == 1 + assert rows[0]["proxy_type"] == "Any" + assert rows[0]["delay"] == 5 + + +def test_parse_proxy_storage_unwraps_single_element_tuple(): + """_parse_proxy_storage unwraps nested single-element wrapper tuples""" + inner = { + "delegate": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + "proxy_type": "Staking", + "delay": 10, + } + raw = ([(inner,)], 0) + rows, deposit = _parse_proxy_storage(raw) + assert len(rows) == 1 + assert rows[0]["delegate"] == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY" + assert rows[0]["proxy_type"] == "Staking" + + +def test_parse_proxy_storage_handles_empty_proxy_list(): + """_parse_proxy_storage returns empty rows for empty proxy vec""" + raw = ([], 0) + rows, deposit = _parse_proxy_storage(raw) + assert rows == [] + assert deposit == 0 + + +def test_parse_proxy_storage_skips_malformed_entries(): + """_parse_proxy_storage skips entries that can't be parsed without crashing""" + raw = ( + [ + { + "delegate": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + "proxy_type": "Any", + "delay": 0, + }, + "garbage", + 42, + ], + 100, + ) + rows, deposit = _parse_proxy_storage(raw) + assert len(rows) == 1 + assert rows[0]["delegate"] == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY" + + +def test_parse_proxy_storage_handles_delegatee_key(): + """_parse_proxy_storage falls back to 'delegatee' key if 'delegate' is absent""" + raw = ( + [ + { + "delegatee": "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY", + "proxy_type": "Transfer", + "delay": 3, + }, + ], + 200, + ) + rows, deposit = _parse_proxy_storage(raw) + assert len(rows) == 1 + assert rows[0]["delegate"] == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY" + assert rows[0]["proxy_type"] == "Transfer" + + # HYPERPARAMS / HYPERPARAMS_METADATA (issue #826) NEW_HYPERPARAMS_826 = {"sn_owner_hotkey", "subnet_owner_hotkey", "recycle_or_burn"} From cca6b9c80ec13a65be90ea325f1debff4ce4c662 Mon Sep 17 00:00:00 2001 From: mkdev11 Date: Tue, 24 Mar 2026 01:33:11 +0100 Subject: [PATCH 2/2] chore: rerun CI