9465099113

Committed 11 Jun 2024 11:59AM CUT coverage: 87.16%. Remained the same

Build # 9465099113

Build Type

push

github

Committed by

Uxio0

Commit Message

Fix docker image generation

Run Details

722 of 835 branches covered (86.47%)

Branch coverage included in aggregate %.

2326 of 2662 relevant lines covered (87.38%)

3.49 hits per line

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

73.04

/src/safe_cli/prompt_parser.py

import argparse
import functools

from prompt_toolkit import HTML, print_formatted_text

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,
    SafeOperatorException,
    SafeVersionNotSupportedException,
    SameFallbackHandlerException,
    SameMasterCopyException,
    SenderRequiredException,
    ThresholdLimitException,
)
from .operators.safe_operator import SafeOperator


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.eip191_message, args.eip712_path)

    @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)

    # 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", type=str)
    group.add_argument("--eip712_path", type=str)
    parser_sign_message.set_defaults(func=sign_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
    )

    return prompt_parser

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

safe-global / safe-cli / 9465099113

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