11343896716

Committed 15 Oct 2024 10:09AM CUT coverage: 88.64%. Remained the same

Build # 11343896716

Build Type

push

github

Committed by

Uxio0

Commit Message

Bump flake8 from 7.1.0 to 7.1.1

Bumps [flake8](https://github.com/pycqa/flake8) from 7.1.0 to 7.1.1.
- [Commits](https://github.com/pycqa/flake8/compare/7.1.0...7.1.1)

---
updated-dependencies:
- dependency-name: flake8
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

Run Details

221 of 262 branches covered (84.35%)

Branch coverage included in aggregate %.

2869 of 3224 relevant lines covered (88.99%)

3.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.23

/src/safe_cli/prompt_parser.py

import argparse
import functools

from prompt_toolkit import HTML, print_formatted_text
from prompt_toolkit.formatted_text import html

from gnosis.safe.api import SafeAPIException

from .argparse_validators import (
    check_ethereum_address,
    check_hex_str,
    check_keccak256_hash,
)
from .operators import SafeServiceNotAvailable
from .operators.exceptions import (
    AccountNotLoadedException,
    ExistingOwnerException,
    FallbackHandlerNotSupportedException,
    HardwareWalletException,
    HashAlreadyApproved,
    InvalidMasterCopyException,
    InvalidMigrationContractException,
    InvalidNonceException,
    NonExistingOwnerException,
    NotEnoughEtherToSend,
    NotEnoughSignatures,
    NotEnoughTokenToSend,
    SafeAlreadyUpdatedException,
    SafeCliTerminationException,
    SafeOperatorException,
    SafeVersionNotSupportedException,
    SameFallbackHandlerException,
    SameMasterCopyException,
    SenderRequiredException,
    ThresholdLimitException,
)
from .operators.safe_operator import SafeOperator
from .safe_completer_constants import meta, safe_commands_arguments


def safe_exception(function):
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        try:
            return function(*args, **kwargs)
        except SafeAPIException as e:
            if e.args:
                print_formatted_text(HTML(f"<b><ansired>{e.args[0]}</ansired></b>"))
        except AccountNotLoadedException as e:
            print_formatted_text(
                HTML(f"<ansired>Account {e.args[0]} is not loaded</ansired>")
            )
        except NotEnoughSignatures as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Cannot find enough owners to sign. {e.args[0]} missing</ansired>"
                )
            )
        except SenderRequiredException:
            print_formatted_text(
                HTML("<ansired>Please load a default sender</ansired>")
            )
        except ExistingOwnerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Owner {e.args[0]} is already an owner of the Safe"
                    f"</ansired>"
                )
            )
        except NonExistingOwnerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Owner {e.args[0]} is not an owner of the Safe"
                    f"</ansired>"
                )
            )
        except HashAlreadyApproved as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Transaction with safe-tx-hash {e.args[0].hex()} has already been approved by "
                    f"owner {e.args[1]}</ansired>"
                )
            )
        except ThresholdLimitException:
            print_formatted_text(
                HTML(
                    "<ansired>Having less owners than threshold is not allowed"
                    "</ansired>"
                )
            )
        except SameFallbackHandlerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Fallback handler {e.args[0]} is the current one</ansired>"
                )
            )
        except FallbackHandlerNotSupportedException:
            print_formatted_text(
                HTML(
                    "<ansired>Fallback handler is not supported for your Safe, "
                    "you need to <b>update</b> first</ansired>"
                )
            )
        except SameMasterCopyException as e:
            print_formatted_text(
                HTML(f"<ansired>Master Copy {e.args[0]} is the current one</ansired>")
            )
        except InvalidMasterCopyException as e:
            print_formatted_text(
                HTML(f"<ansired>Master Copy {e.args[0]} is not valid</ansired>")
            )
        except InvalidMigrationContractException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except InvalidNonceException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except SafeAlreadyUpdatedException:
            print_formatted_text(HTML("<ansired>Safe is already updated</ansired>"))
        except SafeVersionNotSupportedException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except (NotEnoughEtherToSend, NotEnoughTokenToSend) as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Cannot find enough to send. Current balance is {e.args[0]}"
                    f"</ansired>"
                )
            )
        except SafeServiceNotAvailable as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Service not available for network {e.args[0]}</ansired>"
                )
            )
        except HardwareWalletException as e:
            print_formatted_text(
                HTML(f"<ansired>HwDevice exception: {e.args[0]}</ansired>")
            )
        except SafeOperatorException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))

    return wrapper


class PromptParser:
    def __init__(self, safe_operator: SafeOperator):
        self.mode_parser = argparse.ArgumentParser(prog="")
        self.safe_operator = safe_operator
        self.prompt_parser = build_prompt_parser(safe_operator)

    def process_command(self, command: str):
        args = self.prompt_parser.parse_args(command.split())
        return args.func(args)


def build_prompt_parser(safe_operator: SafeOperator) -> argparse.ArgumentParser:
    """
    Returns an ArgParse capable of decoding and executing the Safe commands
    :param safe_operator:
    :return:
    """
    prompt_parser = argparse.ArgumentParser(prog="")
    subparsers = prompt_parser.add_subparsers()

    @safe_exception
    def show_cli_owners(args):
        safe_operator.show_cli_owners()

    @safe_exception
    def load_cli_owners_from_words(args):
        safe_operator.load_cli_owners_from_words(args.words)

    @safe_exception
    def load_cli_owners(args):
        safe_operator.load_cli_owners(args.keys)

    @safe_exception
    def load_ledger_cli_owners(args):
        safe_operator.load_ledger_cli_owners(
            derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
        )

    @safe_exception
    def load_trezor_cli_owners(args):
        safe_operator.load_trezor_cli_owners(
            derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
        )

    @safe_exception
    def unload_cli_owners(args):
        safe_operator.unload_cli_owners(args.addresses)

    @safe_exception
    def approve_hash(args):
        safe_operator.approve_hash(args.hash_to_approve, args.sender)

    @safe_exception
    def sign_message(args):
        safe_operator.sign_message(args.eip712_path)

    @safe_exception
    def confirm_message(args):
        safe_operator.confirm_message(args.safe_message_hash, args.sender)

    @safe_exception
    def add_owner(args):
        safe_operator.add_owner(args.address, threshold=args.threshold)

    @safe_exception
    def remove_owner(args):
        safe_operator.remove_owner(args.address, threshold=args.threshold)

    @safe_exception
    def change_fallback_handler(args):
        safe_operator.change_fallback_handler(args.address)

    @safe_exception
    def change_guard(args):
        safe_operator.change_guard(args.address)

    @safe_exception
    def change_master_copy(args):
        safe_operator.change_master_copy(args.address)

    @safe_exception
    def change_threshold(args):
        safe_operator.change_threshold(args.threshold)

    @safe_exception
    def send_custom(args):
        safe_operator.send_custom(
            args.to,
            args.value,
            args.data,
            safe_nonce=args.safe_nonce,
            delegate_call=args.delegate,
        )

    @safe_exception
    def send_ether(args):
        safe_operator.send_ether(args.to, args.value, safe_nonce=args.safe_nonce)

    @safe_exception
    def send_erc20(args):
        safe_operator.send_erc20(
            args.to, args.token_address, args.amount, safe_nonce=args.safe_nonce
        )

    @safe_exception
    def send_erc721(args):
        safe_operator.send_erc721(
            args.to, args.token_address, args.token_id, safe_nonce=args.safe_nonce
        )

    @safe_exception
    def drain(args):
        safe_operator.drain(args.to)

    @safe_exception
    def get_threshold(args):
        safe_operator.get_threshold()

    @safe_exception
    def get_nonce(args):
        safe_operator.get_nonce()

    @safe_exception
    def get_owners(args):
        safe_operator.get_owners()

    @safe_exception
    def enable_module(args):
        safe_operator.enable_module(args.address)

    @safe_exception
    def disable_module(args):
        safe_operator.disable_module(args.address)

    @safe_exception
    def update_version(args):
        safe_operator.update_version()

    @safe_exception
    def update_version_to_l2(args):
        safe_operator.update_version_to_l2(args.migration_contract)

    @safe_exception
    def get_info(args):
        safe_operator.print_info()

    @safe_exception
    def get_refresh(args):
        safe_operator.refresh_safe_cli_info()

    @safe_exception
    def get_balances(args):
        safe_operator.get_balances()

    @safe_exception
    def get_history(args):
        safe_operator.get_transaction_history()

    @safe_exception
    def sign_tx(args):
        safe_operator.submit_signatures(args.safe_tx_hash)

    @safe_exception
    def batch_txs(args):
        safe_operator.batch_txs(args.safe_nonce, args.safe_tx_hashes)

    @safe_exception
    def execute_tx(args):
        safe_operator.execute_tx(args.safe_tx_hash)

    @safe_exception
    def get_delegates(args):
        safe_operator.get_delegates()

    @safe_exception
    def add_delegate(args):
        safe_operator.add_delegate(args.address, args.label, args.signer)

    @safe_exception
    def remove_delegate(args):
        safe_operator.remove_delegate(args.address, args.signer)

    @safe_exception
    def remove_proposed_transaction(args):
        safe_operator.remove_proposed_transaction(args.safe_tx_hash)

    def list_commands(args):
        print_formatted_text(
            HTML("<b>The following commands can be used:</b>"), end="\n\n"
        )
        for key in sorted(safe_commands_arguments.keys()):
            print_formatted_text(
                HTML(
                    f"<b>{key} {html.html_escape(safe_commands_arguments.get(key))}</b>"
                )
            )
            print_formatted_text(HTML("&#9;"), meta.get(key, ""))
        print_formatted_text(
            HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
        )

    def terminate_cli(args):
        raise SafeCliTerminationException()

    # Cli owners
    parser_show_cli_owners = subparsers.add_parser("show_cli_owners")
    parser_show_cli_owners.set_defaults(func=show_cli_owners)

    parser_load_cli_owners_from_words = subparsers.add_parser(
        "load_cli_owners_from_words"
    )
    parser_load_cli_owners_from_words.add_argument("words", type=str, nargs="+")
    parser_load_cli_owners_from_words.set_defaults(func=load_cli_owners_from_words)

    parser_load_cli_owners = subparsers.add_parser("load_cli_owners")
    parser_load_cli_owners.add_argument("keys", type=str, nargs="+")
    parser_load_cli_owners.set_defaults(func=load_cli_owners)

    parser_load_ledger_cli_owners = subparsers.add_parser("load_ledger_cli_owners")
    parser_load_ledger_cli_owners.add_argument(
        "--derivation-path",
        type=str,
        help="Load address for the provided derivation path",
    )
    parser_load_ledger_cli_owners.add_argument(
        "--legacy-accounts",
        action="store_true",
        help="Search for legacy accounts",
    )
    parser_load_ledger_cli_owners.set_defaults(func=load_ledger_cli_owners)

    parser_load_trezor_cli_owners = subparsers.add_parser("load_trezor_cli_owners")
    parser_load_trezor_cli_owners.add_argument(
        "--derivation-path",
        type=str,
        help="Load address for the provided derivation path",
    )
    parser_load_trezor_cli_owners.add_argument(
        "--legacy-accounts",
        action="store_true",
        help="Search for legacy accounts",
    )
    parser_load_trezor_cli_owners.set_defaults(func=load_trezor_cli_owners)

    parser_unload_cli_owners = subparsers.add_parser("unload_cli_owners")
    parser_unload_cli_owners.add_argument(
        "addresses", type=check_ethereum_address, nargs="+"
    )
    parser_unload_cli_owners.set_defaults(func=unload_cli_owners)

    # Change threshold
    parser_change_threshold = subparsers.add_parser("change_threshold")
    parser_change_threshold.add_argument("threshold", type=int)
    parser_change_threshold.set_defaults(func=change_threshold)

    # Approve hash
    parser_approve_hash = subparsers.add_parser("approve_hash")
    parser_approve_hash.add_argument("hash_to_approve", type=check_keccak256_hash)
    parser_approve_hash.add_argument("sender", type=check_ethereum_address)
    parser_approve_hash.set_defaults(func=approve_hash)

    # Sign message
    parser_sign_message = subparsers.add_parser("sign_message")
    group = parser_sign_message.add_mutually_exclusive_group(required=True)
    group.add_argument("--eip191_message", action="store_true")
    group.add_argument("--eip712_path", type=str)
    parser_sign_message.set_defaults(func=sign_message)

    # Confirm message
    parser_confirm_message = subparsers.add_parser("confirm_message")
    parser_confirm_message.add_argument("safe_message_hash", type=check_keccak256_hash)
    parser_confirm_message.add_argument("sender", type=check_ethereum_address)
    parser_confirm_message.set_defaults(func=confirm_message)

    # Add owner
    parser_add_owner = subparsers.add_parser("add_owner")
    parser_add_owner.add_argument("address", type=check_ethereum_address)
    parser_add_owner.add_argument("--threshold", type=int, default=None)
    parser_add_owner.set_defaults(func=add_owner)

    # Remove owner
    parser_remove_owner = subparsers.add_parser("remove_owner")
    parser_remove_owner.add_argument("address", type=check_ethereum_address)
    parser_remove_owner.add_argument("--threshold", type=int, default=None)
    parser_remove_owner.set_defaults(func=remove_owner)

    # Change FallbackHandler
    parser_change_fallback_handler = subparsers.add_parser("change_fallback_handler")
    parser_change_fallback_handler.add_argument("address", type=check_ethereum_address)
    parser_change_fallback_handler.set_defaults(func=change_fallback_handler)

    # Change Guard
    parser_change_guard = subparsers.add_parser("change_guard")
    parser_change_guard.add_argument("address", type=check_ethereum_address)
    parser_change_guard.set_defaults(func=change_guard)

    # Change MasterCopy
    parser_change_master_copy = subparsers.add_parser("change_master_copy")
    parser_change_master_copy.add_argument("address", type=check_ethereum_address)
    parser_change_master_copy.set_defaults(func=change_master_copy)

    # Update Safe to last version
    parser_update_version = subparsers.add_parser("update")
    parser_update_version.set_defaults(func=update_version)

    # Update non L2 Safe to L2 Safe
    parser_update_version_to_l2 = subparsers.add_parser("update_version_to_l2")
    parser_update_version_to_l2.add_argument(
        "migration_contract", type=check_ethereum_address
    )
    parser_update_version_to_l2.set_defaults(func=update_version_to_l2)

    # Send custom/ether/erc20/erc721
    parser_send_custom = subparsers.add_parser("send_custom")
    parser_send_ether = subparsers.add_parser("send_ether")
    parser_send_erc20 = subparsers.add_parser("send_erc20")
    parser_send_erc721 = subparsers.add_parser("send_erc721")
    parser_send_custom.set_defaults(func=send_custom)
    parser_send_ether.set_defaults(func=send_ether)
    parser_send_erc20.set_defaults(func=send_erc20)
    parser_send_erc721.set_defaults(func=send_erc721)
    # They have some common arguments
    for parser in (
        parser_send_custom,
        parser_send_ether,
        parser_send_erc20,
        parser_send_erc721,
    ):
        parser.add_argument(
            "--safe-nonce",
            type=int,
            help="Use custom safe nonce instead of "
            "the one for last executed SafeTx + 1",
        )

    # To/value is common for send custom and send ether
    for parser in (parser_send_custom, parser_send_ether):
        parser.add_argument("to", type=check_ethereum_address)
        parser.add_argument("value", type=int)

    parser_send_custom.add_argument("data", type=check_hex_str)
    parser_send_custom.add_argument(
        "--delegate", action="store_true", help="Use DELEGATE_CALL. By default use CALL"
    )

    # Send erc20/721 have common arguments
    for parser in (parser_send_erc20, parser_send_erc721):
        parser.add_argument("to", type=check_ethereum_address)
        parser.add_argument("token_address", type=check_ethereum_address)

    parser_send_erc20.add_argument("amount", type=int)
    parser_send_erc721.add_argument("token_id", type=int)

    # Drain only needs receiver account
    parser_drain = subparsers.add_parser("drain")
    parser_drain.set_defaults(func=drain)
    parser_drain.add_argument("to", type=check_ethereum_address)

    # Retrieve threshold, nonce or owners
    parser_get_threshold = subparsers.add_parser("get_threshold")
    parser_get_threshold.set_defaults(func=get_threshold)

    parser_get_nonce = subparsers.add_parser("get_nonce")
    parser_get_nonce.set_defaults(func=get_nonce)

    parser_get_owners = subparsers.add_parser("get_owners")
    parser_get_owners.set_defaults(func=get_owners)

    # Enable and disable modules
    parser_enable_module = subparsers.add_parser("enable_module")
    parser_enable_module.add_argument("address", type=check_ethereum_address)
    parser_enable_module.set_defaults(func=enable_module)

    parser_disable_module = subparsers.add_parser("disable_module")
    parser_disable_module.add_argument("address", type=check_ethereum_address)
    parser_disable_module.set_defaults(func=disable_module)

    # Info and refresh
    parser_info = subparsers.add_parser("info")
    parser_info.set_defaults(func=get_info)

    parser_refresh = subparsers.add_parser("refresh")
    parser_refresh.set_defaults(func=get_refresh)

    # Tx-Service
    # TODO Use subcommands
    parser_tx_service = subparsers.add_parser("balances")
    parser_tx_service.set_defaults(func=get_balances)

    parser_tx_service = subparsers.add_parser("history")
    parser_tx_service.set_defaults(func=get_history)

    parser_tx_service = subparsers.add_parser("sign-tx")
    parser_tx_service.set_defaults(func=sign_tx)
    parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)

    parser_tx_service = subparsers.add_parser("batch-txs")
    parser_tx_service.set_defaults(func=batch_txs)
    parser_tx_service.add_argument("safe_nonce", type=int)
    parser_tx_service.add_argument(
        "safe_tx_hashes", type=check_keccak256_hash, nargs="+"
    )

    parser_tx_service = subparsers.add_parser("execute-tx")
    parser_tx_service.set_defaults(func=execute_tx)
    parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)

    # List delegates
    parser_delegates = subparsers.add_parser("get_delegates")
    parser_delegates.set_defaults(func=get_delegates)

    # Add delegate
    parser_add_delegate = subparsers.add_parser("add_delegate")
    parser_add_delegate.set_defaults(func=add_delegate)
    parser_add_delegate.add_argument("address", type=check_ethereum_address)
    parser_add_delegate.add_argument("label", type=str)
    parser_add_delegate.add_argument("signer", type=check_ethereum_address)

    # Remove delegate
    parser_remove_delegate = subparsers.add_parser("remove_delegate")
    parser_remove_delegate.set_defaults(func=remove_delegate)
    parser_remove_delegate.add_argument("address", type=check_ethereum_address)
    parser_remove_delegate.add_argument("signer", type=check_ethereum_address)

    # Remove not executed proposed transaction
    parser_remove_proposed_transaction = subparsers.add_parser(
        "remove_proposed_transaction"
    )
    parser_remove_proposed_transaction.set_defaults(func=remove_proposed_transaction)
    parser_remove_proposed_transaction.add_argument(
        "safe_tx_hash", type=check_keccak256_hash
    )

    # List all command options
    parser_help = subparsers.add_parser("help")
    parser_help.set_defaults(func=list_commands)

    # Terminate safe cli
    parser_exit = subparsers.add_parser("exit")
    parser_exit.set_defaults(func=terminate_cli)

    return prompt_parser

1	import argparse	4✔
2	import functools	4✔
3
4	from prompt_toolkit import HTML, print_formatted_text	4✔
5	from prompt_toolkit.formatted_text import html	4✔
6
7	from gnosis.safe.api import SafeAPIException	4✔
8
9	from .argparse_validators import (	4✔
10	check_ethereum_address,
11	check_hex_str,
12	check_keccak256_hash,
13	)
14	from .operators import SafeServiceNotAvailable	4✔
15	from .operators.exceptions import (	4✔
16	AccountNotLoadedException,
17	ExistingOwnerException,
18	FallbackHandlerNotSupportedException,
19	HardwareWalletException,
20	HashAlreadyApproved,
21	InvalidMasterCopyException,
22	InvalidMigrationContractException,
23	InvalidNonceException,
24	NonExistingOwnerException,
25	NotEnoughEtherToSend,
26	NotEnoughSignatures,
27	NotEnoughTokenToSend,
28	SafeAlreadyUpdatedException,
29	SafeCliTerminationException,
30	SafeOperatorException,
31	SafeVersionNotSupportedException,
32	SameFallbackHandlerException,
33	SameMasterCopyException,
34	SenderRequiredException,
35	ThresholdLimitException,
36	)
37	from .operators.safe_operator import SafeOperator	4✔
38	from .safe_completer_constants import meta, safe_commands_arguments	4✔
39
40
41	def safe_exception(function):	4✔
42	@functools.wraps(function)	4✔
43	def wrapper(args, *kwargs):	4✔
44	try:	4✔
45	return function(args, *kwargs)	4✔
46	except SafeAPIException as e:	4✔
47	if e.args:	×
48	print_formatted_text(HTML(f"<b><ansired>{e.args[0]}</ansired></b>"))	×
49	except AccountNotLoadedException as e:	4✔
50	print_formatted_text(	×
51	HTML(f"<ansired>Account {e.args[0]} is not loaded</ansired>")
52	)
53	except NotEnoughSignatures as e:	4✔
54	print_formatted_text(	4✔
55	HTML(
56	f"<ansired>Cannot find enough owners to sign. {e.args[0]} missing</ansired>"
57	)
58	)
59	except SenderRequiredException:	×
60	print_formatted_text(	×
61	HTML("<ansired>Please load a default sender</ansired>")
62	)
63	except ExistingOwnerException as e:	×
64	print_formatted_text(	×
65	HTML(
66	f"<ansired>Owner {e.args[0]} is already an owner of the Safe"
67	f"</ansired>"
68	)
69	)
70	except NonExistingOwnerException as e:	×
71	print_formatted_text(	×
72	HTML(
73	f"<ansired>Owner {e.args[0]} is not an owner of the Safe"
74	f"</ansired>"
75	)
76	)
77	except HashAlreadyApproved as e:	×
78	print_formatted_text(	×
79	HTML(
80	f"<ansired>Transaction with safe-tx-hash {e.args[0].hex()} has already been approved by "
81	f"owner {e.args[1]}</ansired>"
82	)
83	)
84	except ThresholdLimitException:	×
85	print_formatted_text(	×
86	HTML(
87	"<ansired>Having less owners than threshold is not allowed"
88	"</ansired>"
89	)
90	)
91	except SameFallbackHandlerException as e:	×
92	print_formatted_text(	×
93	HTML(
94	f"<ansired>Fallback handler {e.args[0]} is the current one</ansired>"
95	)
96	)
97	except FallbackHandlerNotSupportedException:	×
98	print_formatted_text(	×
99	HTML(
100	"<ansired>Fallback handler is not supported for your Safe, "
101	"you need to <b>update</b> first</ansired>"
102	)
103	)
104	except SameMasterCopyException as e:	×
105	print_formatted_text(	×
106	HTML(f"<ansired>Master Copy {e.args[0]} is the current one</ansired>")
107	)
108	except InvalidMasterCopyException as e:	×
109	print_formatted_text(	×
110	HTML(f"<ansired>Master Copy {e.args[0]} is not valid</ansired>")
111	)
112	except InvalidMigrationContractException as e:	×
113	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
114	except InvalidNonceException as e:	×
115	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
116	except SafeAlreadyUpdatedException:	×
117	print_formatted_text(HTML("<ansired>Safe is already updated</ansired>"))	×
118	except SafeVersionNotSupportedException as e:	×
119	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
120	except (NotEnoughEtherToSend, NotEnoughTokenToSend) as e:	×
121	print_formatted_text(	×
122	HTML(
123	f"<ansired>Cannot find enough to send. Current balance is {e.args[0]}"
124	f"</ansired>"
125	)
126	)
127	except SafeServiceNotAvailable as e:	×
128	print_formatted_text(	×
129	HTML(
130	f"<ansired>Service not available for network {e.args[0]}</ansired>"
131	)
132	)
133	except HardwareWalletException as e:	×
134	print_formatted_text(	×
135	HTML(f"<ansired>HwDevice exception: {e.args[0]}</ansired>")
136	)
137	except SafeOperatorException as e:	×
138	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
139
140	return wrapper	4✔
141
142
143	class PromptParser:	4✔
144	def __init__(self, safe_operator: SafeOperator):	4✔
145	self.mode_parser = argparse.ArgumentParser(prog="")	4✔
146	self.safe_operator = safe_operator	4✔
147	self.prompt_parser = build_prompt_parser(safe_operator)	4✔
148
149	def process_command(self, command: str):	4✔
150	args = self.prompt_parser.parse_args(command.split())	4✔
151	return args.func(args)	4✔
152
153
154	def build_prompt_parser(safe_operator: SafeOperator) -> argparse.ArgumentParser:	4✔
155	"""
156	Returns an ArgParse capable of decoding and executing the Safe commands
157	:param safe_operator:
158	:return:
159	"""
160	prompt_parser = argparse.ArgumentParser(prog="")	4✔
161	subparsers = prompt_parser.add_subparsers()	4✔
162
163	@safe_exception	4✔
164	def show_cli_owners(args):	4✔
165	safe_operator.show_cli_owners()	×
166
167	@safe_exception	4✔
168	def load_cli_owners_from_words(args):	4✔
169	safe_operator.load_cli_owners_from_words(args.words)	×
170
171	@safe_exception	4✔
172	def load_cli_owners(args):	4✔
173	safe_operator.load_cli_owners(args.keys)	4✔
174
175	@safe_exception	4✔
176	def load_ledger_cli_owners(args):	4✔
177	safe_operator.load_ledger_cli_owners(	×
178	derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
179	)
180
181	@safe_exception	4✔
182	def load_trezor_cli_owners(args):	4✔
183	safe_operator.load_trezor_cli_owners(	×
184	derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
185	)
186
187	@safe_exception	4✔
188	def unload_cli_owners(args):	4✔
189	safe_operator.unload_cli_owners(args.addresses)	×
190
191	@safe_exception	4✔
192	def approve_hash(args):	4✔
193	safe_operator.approve_hash(args.hash_to_approve, args.sender)	4✔
194
195	@safe_exception	4✔
196	def sign_message(args):	4✔
197	safe_operator.sign_message(args.eip712_path)	×
198
199	@safe_exception	4✔
200	def confirm_message(args):	4✔
201	safe_operator.confirm_message(args.safe_message_hash, args.sender)	×
202
203	@safe_exception	4✔
204	def add_owner(args):	4✔
205	safe_operator.add_owner(args.address, threshold=args.threshold)	×
206
207	@safe_exception	4✔
208	def remove_owner(args):	4✔
209	safe_operator.remove_owner(args.address, threshold=args.threshold)	4✔
210
211	@safe_exception	4✔
212	def change_fallback_handler(args):	4✔
213	safe_operator.change_fallback_handler(args.address)	×
214
215	@safe_exception	4✔
216	def change_guard(args):	4✔
217	safe_operator.change_guard(args.address)	×
218
219	@safe_exception	4✔
220	def change_master_copy(args):	4✔
221	safe_operator.change_master_copy(args.address)	×
222
223	@safe_exception	4✔
224	def change_threshold(args):	4✔
225	safe_operator.change_threshold(args.threshold)	4✔
226
227	@safe_exception	4✔
228	def send_custom(args):	4✔
229	safe_operator.send_custom(	×
230	args.to,
231	args.value,
232	args.data,
233	safe_nonce=args.safe_nonce,
234	delegate_call=args.delegate,
235	)
236
237	@safe_exception	4✔
238	def send_ether(args):	4✔
239	safe_operator.send_ether(args.to, args.value, safe_nonce=args.safe_nonce)	4✔
240
241	@safe_exception	4✔
242	def send_erc20(args):	4✔
243	safe_operator.send_erc20(	×
244	args.to, args.token_address, args.amount, safe_nonce=args.safe_nonce
245	)
246
247	@safe_exception	4✔
248	def send_erc721(args):	4✔
249	safe_operator.send_erc721(	×
250	args.to, args.token_address, args.token_id, safe_nonce=args.safe_nonce
251	)
252
253	@safe_exception	4✔
254	def drain(args):	4✔
255	safe_operator.drain(args.to)	×
256
257	@safe_exception	4✔
258	def get_threshold(args):	4✔
259	safe_operator.get_threshold()	×
260
261	@safe_exception	4✔
262	def get_nonce(args):	4✔
263	safe_operator.get_nonce()	×
264
265	@safe_exception	4✔
266	def get_owners(args):	4✔
267	safe_operator.get_owners()	×
268
269	@safe_exception	4✔
270	def enable_module(args):	4✔
271	safe_operator.enable_module(args.address)	×
272
273	@safe_exception	4✔
274	def disable_module(args):	4✔
275	safe_operator.disable_module(args.address)	×
276
277	@safe_exception	4✔
278	def update_version(args):	4✔
279	safe_operator.update_version()	×
280
281	@safe_exception	4✔
282	def update_version_to_l2(args):	4✔
283	safe_operator.update_version_to_l2(args.migration_contract)	×
284
285	@safe_exception	4✔
286	def get_info(args):	4✔
287	safe_operator.print_info()	×
288
289	@safe_exception	4✔
290	def get_refresh(args):	4✔
291	safe_operator.refresh_safe_cli_info()	×
292
293	@safe_exception	4✔
294	def get_balances(args):	4✔
295	safe_operator.get_balances()	×
296
297	@safe_exception	4✔
298	def get_history(args):	4✔
299	safe_operator.get_transaction_history()	×
300
301	@safe_exception	4✔
302	def sign_tx(args):	4✔
303	safe_operator.submit_signatures(args.safe_tx_hash)	×
304
305	@safe_exception	4✔
306	def batch_txs(args):	4✔
307	safe_operator.batch_txs(args.safe_nonce, args.safe_tx_hashes)	×
308
309	@safe_exception	4✔
310	def execute_tx(args):	4✔
311	safe_operator.execute_tx(args.safe_tx_hash)	×
312
313	@safe_exception	4✔
314	def get_delegates(args):	4✔
315	safe_operator.get_delegates()	×
316
317	@safe_exception	4✔
318	def add_delegate(args):	4✔
319	safe_operator.add_delegate(args.address, args.label, args.signer)	×
320
321	@safe_exception	4✔
322	def remove_delegate(args):	4✔
323	safe_operator.remove_delegate(args.address, args.signer)	×
324
325	@safe_exception	4✔
326	def remove_proposed_transaction(args):	4✔
327	safe_operator.remove_proposed_transaction(args.safe_tx_hash)	×
328
329	def list_commands(args):	4✔
330	print_formatted_text(	×
331	HTML("<b>The following commands can be used:</b>"), end="\n\n"
332	)
333	for key in sorted(safe_commands_arguments.keys()):	×
334	print_formatted_text(	×
335	HTML(
336	f"<b>{key} {html.html_escape(safe_commands_arguments.get(key))}</b>"
337	)
338	)
339	print_formatted_text(HTML(" "), meta.get(key, ""))	×
340	print_formatted_text(	×
341	HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
342	)
343
344	def terminate_cli(args):	4✔
345	raise SafeCliTerminationException()	4✔
346
347	# Cli owners
348	parser_show_cli_owners = subparsers.add_parser("show_cli_owners")	4✔
349	parser_show_cli_owners.set_defaults(func=show_cli_owners)	4✔
350
351	parser_load_cli_owners_from_words = subparsers.add_parser(	4✔
352	"load_cli_owners_from_words"
353	)
354	parser_load_cli_owners_from_words.add_argument("words", type=str, nargs="+")	4✔
355	parser_load_cli_owners_from_words.set_defaults(func=load_cli_owners_from_words)	4✔
356
357	parser_load_cli_owners = subparsers.add_parser("load_cli_owners")	4✔
358	parser_load_cli_owners.add_argument("keys", type=str, nargs="+")	4✔
359	parser_load_cli_owners.set_defaults(func=load_cli_owners)	4✔
360
361	parser_load_ledger_cli_owners = subparsers.add_parser("load_ledger_cli_owners")	4✔
362	parser_load_ledger_cli_owners.add_argument(	4✔
363	"--derivation-path",
364	type=str,
365	help="Load address for the provided derivation path",
366	)
367	parser_load_ledger_cli_owners.add_argument(	4✔
368	"--legacy-accounts",
369	action="store_true",
370	help="Search for legacy accounts",
371	)
372	parser_load_ledger_cli_owners.set_defaults(func=load_ledger_cli_owners)	4✔
373
374	parser_load_trezor_cli_owners = subparsers.add_parser("load_trezor_cli_owners")	4✔
375	parser_load_trezor_cli_owners.add_argument(	4✔
376	"--derivation-path",
377	type=str,
378	help="Load address for the provided derivation path",
379	)
380	parser_load_trezor_cli_owners.add_argument(	4✔
381	"--legacy-accounts",
382	action="store_true",
383	help="Search for legacy accounts",
384	)
385	parser_load_trezor_cli_owners.set_defaults(func=load_trezor_cli_owners)	4✔
386
387	parser_unload_cli_owners = subparsers.add_parser("unload_cli_owners")	4✔
388	parser_unload_cli_owners.add_argument(	4✔
389	"addresses", type=check_ethereum_address, nargs="+"
390	)
391	parser_unload_cli_owners.set_defaults(func=unload_cli_owners)	4✔
392
393	# Change threshold
394	parser_change_threshold = subparsers.add_parser("change_threshold")	4✔
395	parser_change_threshold.add_argument("threshold", type=int)	4✔
396	parser_change_threshold.set_defaults(func=change_threshold)	4✔
397
398	# Approve hash
399	parser_approve_hash = subparsers.add_parser("approve_hash")	4✔
400	parser_approve_hash.add_argument("hash_to_approve", type=check_keccak256_hash)	4✔
401	parser_approve_hash.add_argument("sender", type=check_ethereum_address)	4✔
402	parser_approve_hash.set_defaults(func=approve_hash)	4✔
403
404	# Sign message
405	parser_sign_message = subparsers.add_parser("sign_message")	4✔
406	group = parser_sign_message.add_mutually_exclusive_group(required=True)	4✔
407	group.add_argument("--eip191_message", action="store_true")	4✔
408	group.add_argument("--eip712_path", type=str)	4✔
409	parser_sign_message.set_defaults(func=sign_message)	4✔
410
411	# Confirm message
412	parser_confirm_message = subparsers.add_parser("confirm_message")	4✔
413	parser_confirm_message.add_argument("safe_message_hash", type=check_keccak256_hash)	4✔
414	parser_confirm_message.add_argument("sender", type=check_ethereum_address)	4✔
415	parser_confirm_message.set_defaults(func=confirm_message)	4✔
416
417	# Add owner
418	parser_add_owner = subparsers.add_parser("add_owner")	4✔
419	parser_add_owner.add_argument("address", type=check_ethereum_address)	4✔
420	parser_add_owner.add_argument("--threshold", type=int, default=None)	4✔
421	parser_add_owner.set_defaults(func=add_owner)	4✔
422
423	# Remove owner
424	parser_remove_owner = subparsers.add_parser("remove_owner")	4✔
425	parser_remove_owner.add_argument("address", type=check_ethereum_address)	4✔
426	parser_remove_owner.add_argument("--threshold", type=int, default=None)	4✔
427	parser_remove_owner.set_defaults(func=remove_owner)	4✔
428
429	# Change FallbackHandler
430	parser_change_fallback_handler = subparsers.add_parser("change_fallback_handler")	4✔
431	parser_change_fallback_handler.add_argument("address", type=check_ethereum_address)	4✔
432	parser_change_fallback_handler.set_defaults(func=change_fallback_handler)	4✔
433
434	# Change Guard
435	parser_change_guard = subparsers.add_parser("change_guard")	4✔
436	parser_change_guard.add_argument("address", type=check_ethereum_address)	4✔
437	parser_change_guard.set_defaults(func=change_guard)	4✔
438
439	# Change MasterCopy
440	parser_change_master_copy = subparsers.add_parser("change_master_copy")	4✔
441	parser_change_master_copy.add_argument("address", type=check_ethereum_address)	4✔
442	parser_change_master_copy.set_defaults(func=change_master_copy)	4✔
443
444	# Update Safe to last version
445	parser_update_version = subparsers.add_parser("update")	4✔
446	parser_update_version.set_defaults(func=update_version)	4✔
447
448	# Update non L2 Safe to L2 Safe
449	parser_update_version_to_l2 = subparsers.add_parser("update_version_to_l2")	4✔
450	parser_update_version_to_l2.add_argument(	4✔
451	"migration_contract", type=check_ethereum_address
452	)
453	parser_update_version_to_l2.set_defaults(func=update_version_to_l2)	4✔
454
455	# Send custom/ether/erc20/erc721
456	parser_send_custom = subparsers.add_parser("send_custom")	4✔
457	parser_send_ether = subparsers.add_parser("send_ether")	4✔
458	parser_send_erc20 = subparsers.add_parser("send_erc20")	4✔
459	parser_send_erc721 = subparsers.add_parser("send_erc721")	4✔
460	parser_send_custom.set_defaults(func=send_custom)	4✔
461	parser_send_ether.set_defaults(func=send_ether)	4✔
462	parser_send_erc20.set_defaults(func=send_erc20)	4✔
463	parser_send_erc721.set_defaults(func=send_erc721)	4✔
464	# They have some common arguments
465	for parser in (	4✔
466	parser_send_custom,
467	parser_send_ether,
468	parser_send_erc20,
469	parser_send_erc721,
470	):
471	parser.add_argument(	4✔
472	"--safe-nonce",
473	type=int,
474	help="Use custom safe nonce instead of "
475	"the one for last executed SafeTx + 1",
476	)
477
478	# To/value is common for send custom and send ether
479	for parser in (parser_send_custom, parser_send_ether):	4✔
480	parser.add_argument("to", type=check_ethereum_address)	4✔
481	parser.add_argument("value", type=int)	4✔
482
483	parser_send_custom.add_argument("data", type=check_hex_str)	4✔
484	parser_send_custom.add_argument(	4✔
485	"--delegate", action="store_true", help="Use DELEGATE_CALL. By default use CALL"
486	)
487
488	# Send erc20/721 have common arguments
489	for parser in (parser_send_erc20, parser_send_erc721):	4✔
490	parser.add_argument("to", type=check_ethereum_address)	4✔
491	parser.add_argument("token_address", type=check_ethereum_address)	4✔
492
493	parser_send_erc20.add_argument("amount", type=int)	4✔
494	parser_send_erc721.add_argument("token_id", type=int)	4✔
495
496	# Drain only needs receiver account
497	parser_drain = subparsers.add_parser("drain")	4✔
498	parser_drain.set_defaults(func=drain)	4✔
499	parser_drain.add_argument("to", type=check_ethereum_address)	4✔
500
501	# Retrieve threshold, nonce or owners
502	parser_get_threshold = subparsers.add_parser("get_threshold")	4✔
503	parser_get_threshold.set_defaults(func=get_threshold)	4✔
504
505	parser_get_nonce = subparsers.add_parser("get_nonce")	4✔
506	parser_get_nonce.set_defaults(func=get_nonce)	4✔
507
508	parser_get_owners = subparsers.add_parser("get_owners")	4✔
509	parser_get_owners.set_defaults(func=get_owners)	4✔
510
511	# Enable and disable modules
512	parser_enable_module = subparsers.add_parser("enable_module")	4✔
513	parser_enable_module.add_argument("address", type=check_ethereum_address)	4✔
514	parser_enable_module.set_defaults(func=enable_module)	4✔
515
516	parser_disable_module = subparsers.add_parser("disable_module")	4✔
517	parser_disable_module.add_argument("address", type=check_ethereum_address)	4✔
518	parser_disable_module.set_defaults(func=disable_module)	4✔
519
520	# Info and refresh
521	parser_info = subparsers.add_parser("info")	4✔
522	parser_info.set_defaults(func=get_info)	4✔
523
524	parser_refresh = subparsers.add_parser("refresh")	4✔
525	parser_refresh.set_defaults(func=get_refresh)	4✔
526
527	# Tx-Service
528	# TODO Use subcommands
529	parser_tx_service = subparsers.add_parser("balances")	4✔
530	parser_tx_service.set_defaults(func=get_balances)	4✔
531
532	parser_tx_service = subparsers.add_parser("history")	4✔
533	parser_tx_service.set_defaults(func=get_history)	4✔
534
535	parser_tx_service = subparsers.add_parser("sign-tx")	4✔
536	parser_tx_service.set_defaults(func=sign_tx)	4✔
537	parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)	4✔
538
539	parser_tx_service = subparsers.add_parser("batch-txs")	4✔
540	parser_tx_service.set_defaults(func=batch_txs)	4✔
541	parser_tx_service.add_argument("safe_nonce", type=int)	4✔
542	parser_tx_service.add_argument(	4✔
543	"safe_tx_hashes", type=check_keccak256_hash, nargs="+"
544	)
545
546	parser_tx_service = subparsers.add_parser("execute-tx")	4✔
547	parser_tx_service.set_defaults(func=execute_tx)	4✔
548	parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)	4✔
549
550	# List delegates
551	parser_delegates = subparsers.add_parser("get_delegates")	4✔
552	parser_delegates.set_defaults(func=get_delegates)	4✔
553
554	# Add delegate
555	parser_add_delegate = subparsers.add_parser("add_delegate")	4✔
556	parser_add_delegate.set_defaults(func=add_delegate)	4✔
557	parser_add_delegate.add_argument("address", type=check_ethereum_address)	4✔
558	parser_add_delegate.add_argument("label", type=str)	4✔
559	parser_add_delegate.add_argument("signer", type=check_ethereum_address)	4✔
560
561	# Remove delegate
562	parser_remove_delegate = subparsers.add_parser("remove_delegate")	4✔
563	parser_remove_delegate.set_defaults(func=remove_delegate)	4✔
564	parser_remove_delegate.add_argument("address", type=check_ethereum_address)	4✔
565	parser_remove_delegate.add_argument("signer", type=check_ethereum_address)	4✔
566
567	# Remove not executed proposed transaction
568	parser_remove_proposed_transaction = subparsers.add_parser(	4✔
569	"remove_proposed_transaction"
570	)
571	parser_remove_proposed_transaction.set_defaults(func=remove_proposed_transaction)	4✔
572	parser_remove_proposed_transaction.add_argument(	4✔
573	"safe_tx_hash", type=check_keccak256_hash
574	)
575
576	# List all command options
577	parser_help = subparsers.add_parser("help")	4✔
578	parser_help.set_defaults(func=list_commands)	4✔
579
580	# Terminate safe cli
581	parser_exit = subparsers.add_parser("exit")	4✔
582	parser_exit.set_defaults(func=terminate_cli)	4✔
583
584	return prompt_parser	4✔

safe-global / safe-cli / 11343896716

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous