7087242091

Committed 04 Dec 2023 01:23PM UTC coverage: 93.371%. Remained the same

Build # 7087242091

Build Type

Pull #320

github

Committed by

web-flow

Commit Message

Merge 489f75539 into 443cd5f78

Pull Request Pull Request #320: Bump web3 from 6.11.3 to 6.11.4

Run Details

1803 of 1931 relevant lines covered (93.37%)

3.69 hits per line

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

82.44

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

    # 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="Enable search legacy accounts",
    )
    parser_load_ledger_cli_owners.set_defaults(func=load_ledger_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)

    # 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_hex_str)

    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_hex_str, 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_hex_str)

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

    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(	4✔
53	f"<ansired>Cannot find enough owners to sign. {e.args[0]} missing</ansired>"	4✔
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 unload_cli_owners(args):	4✔
180	safe_operator.unload_cli_owners(args.addresses)	×
181
182	@safe_exception	4✔
183	def approve_hash(args):	4✔
184	safe_operator.approve_hash(args.hash_to_approve, args.sender)	4✔
185
186	@safe_exception	4✔
187	def add_owner(args):	4✔
188	safe_operator.add_owner(args.address, threshold=args.threshold)	×
189
190	@safe_exception	4✔
191	def remove_owner(args):	4✔
192	safe_operator.remove_owner(args.address, threshold=args.threshold)	4✔
193
194	@safe_exception	4✔
195	def change_fallback_handler(args):	4✔
196	safe_operator.change_fallback_handler(args.address)	×
197
198	@safe_exception	4✔
199	def change_guard(args):	4✔
200	safe_operator.change_guard(args.address)	×
201
202	@safe_exception	4✔
203	def change_master_copy(args):	4✔
204	safe_operator.change_master_copy(args.address)	×
205
206	@safe_exception	4✔
207	def change_threshold(args):	4✔
208	safe_operator.change_threshold(args.threshold)	4✔
209
210	@safe_exception	4✔
211	def send_custom(args):	4✔
212	safe_operator.send_custom(
213	args.to,
214	args.value,
215	args.data,
216	safe_nonce=args.safe_nonce,
217	delegate_call=args.delegate,
218	)
219
220	@safe_exception	4✔
221	def send_ether(args):	4✔
222	safe_operator.send_ether(args.to, args.value, safe_nonce=args.safe_nonce)	4✔
223
224	@safe_exception	4✔
225	def send_erc20(args):	4✔
226	safe_operator.send_erc20(
227	args.to, args.token_address, args.amount, safe_nonce=args.safe_nonce
228	)
229
230	@safe_exception	4✔
231	def send_erc721(args):	4✔
232	safe_operator.send_erc721(
233	args.to, args.token_address, args.token_id, safe_nonce=args.safe_nonce
234	)
235
236	@safe_exception	4✔
237	def drain(args):	4✔
238	safe_operator.drain(args.to)	×
239
240	@safe_exception	4✔
241	def get_threshold(args):	4✔
242	safe_operator.get_threshold()	×
243
244	@safe_exception	4✔
245	def get_nonce(args):	4✔
246	safe_operator.get_nonce()	×
247
248	@safe_exception	4✔
249	def get_owners(args):	4✔
250	safe_operator.get_owners()	×
251
252	@safe_exception	4✔
253	def enable_module(args):	4✔
254	safe_operator.enable_module(args.address)	×
255
256	@safe_exception	4✔
257	def disable_module(args):	4✔
258	safe_operator.disable_module(args.address)	×
259
260	@safe_exception	4✔
261	def update_version(args):	4✔
262	safe_operator.update_version()	×
263
264	@safe_exception	4✔
265	def update_version_to_l2(args):	4✔
266	safe_operator.update_version_to_l2(args.migration_contract)	×
267
268	@safe_exception	4✔
269	def get_info(args):	4✔
270	safe_operator.print_info()	×
271
272	@safe_exception	4✔
273	def get_refresh(args):	4✔
274	safe_operator.refresh_safe_cli_info()	×
275
276	@safe_exception	4✔
277	def get_balances(args):	4✔
278	safe_operator.get_balances()	×
279
280	@safe_exception	4✔
281	def get_history(args):	4✔
282	safe_operator.get_transaction_history()	×
283
284	@safe_exception	4✔
285	def sign_tx(args):	4✔
286	safe_operator.submit_signatures(args.safe_tx_hash)	×
287
288	@safe_exception	4✔
289	def batch_txs(args):	4✔
290	safe_operator.batch_txs(args.safe_nonce, args.safe_tx_hashes)	×
291
292	@safe_exception	4✔
293	def execute_tx(args):	4✔
294	safe_operator.execute_tx(args.safe_tx_hash)	×
295
296	@safe_exception	4✔
297	def get_delegates(args):	4✔
298	safe_operator.get_delegates()	×
299
300	@safe_exception	4✔
301	def add_delegate(args):	4✔
302	safe_operator.add_delegate(args.address, args.label, args.signer)	×
303
304	@safe_exception	4✔
305	def remove_delegate(args):	4✔
306	safe_operator.remove_delegate(args.address, args.signer)	×
307
308	# Cli owners
309	parser_show_cli_owners = subparsers.add_parser("show_cli_owners")	4✔
310	parser_show_cli_owners.set_defaults(func=show_cli_owners)	4✔
311
312	parser_load_cli_owners_from_words = subparsers.add_parser(	4✔
313	"load_cli_owners_from_words"	4✔
314	)
315	parser_load_cli_owners_from_words.add_argument("words", type=str, nargs="+")	4✔
316	parser_load_cli_owners_from_words.set_defaults(func=load_cli_owners_from_words)	4✔
317
318	parser_load_cli_owners = subparsers.add_parser("load_cli_owners")	4✔
319	parser_load_cli_owners.add_argument("keys", type=str, nargs="+")	4✔
320	parser_load_cli_owners.set_defaults(func=load_cli_owners)	4✔
321
322	parser_load_ledger_cli_owners = subparsers.add_parser("load_ledger_cli_owners")	4✔
323	parser_load_ledger_cli_owners.add_argument(	4✔
324	"--derivation-path",	4✔
325	type=str,	4✔
326	help="Load address for the provided derivation path",	4✔
327	)
328	parser_load_ledger_cli_owners.add_argument(	4✔
329	"--legacy-accounts",	4✔
330	action="store_true",	4✔
331	help="Enable search legacy accounts",	4✔
332	)
333	parser_load_ledger_cli_owners.set_defaults(func=load_ledger_cli_owners)	4✔
334
335	parser_unload_cli_owners = subparsers.add_parser("unload_cli_owners")	4✔
336	parser_unload_cli_owners.add_argument(	4✔
337	"addresses", type=check_ethereum_address, nargs="+"	4✔
338	)
339	parser_unload_cli_owners.set_defaults(func=unload_cli_owners)	4✔
340
341	# Change threshold
342	parser_change_threshold = subparsers.add_parser("change_threshold")	4✔
343	parser_change_threshold.add_argument("threshold", type=int)	4✔
344	parser_change_threshold.set_defaults(func=change_threshold)	4✔
345
346	# Approve hash
347	parser_approve_hash = subparsers.add_parser("approve_hash")	4✔
348	parser_approve_hash.add_argument("hash_to_approve", type=check_keccak256_hash)	4✔
349	parser_approve_hash.add_argument("sender", type=check_ethereum_address)	4✔
350	parser_approve_hash.set_defaults(func=approve_hash)	4✔
351
352	# Add owner
353	parser_add_owner = subparsers.add_parser("add_owner")	4✔
354	parser_add_owner.add_argument("address", type=check_ethereum_address)	4✔
355	parser_add_owner.add_argument("--threshold", type=int, default=None)	4✔
356	parser_add_owner.set_defaults(func=add_owner)	4✔
357
358	# Remove owner
359	parser_remove_owner = subparsers.add_parser("remove_owner")	4✔
360	parser_remove_owner.add_argument("address", type=check_ethereum_address)	4✔
361	parser_remove_owner.add_argument("--threshold", type=int, default=None)	4✔
362	parser_remove_owner.set_defaults(func=remove_owner)	4✔
363
364	# Change FallbackHandler
365	parser_change_fallback_handler = subparsers.add_parser("change_fallback_handler")	4✔
366	parser_change_fallback_handler.add_argument("address", type=check_ethereum_address)	4✔
367	parser_change_fallback_handler.set_defaults(func=change_fallback_handler)	4✔
368
369	# Change Guard
370	parser_change_guard = subparsers.add_parser("change_guard")	4✔
371	parser_change_guard.add_argument("address", type=check_ethereum_address)	4✔
372	parser_change_guard.set_defaults(func=change_guard)	4✔
373
374	# Change MasterCopy
375	parser_change_master_copy = subparsers.add_parser("change_master_copy")	4✔
376	parser_change_master_copy.add_argument("address", type=check_ethereum_address)	4✔
377	parser_change_master_copy.set_defaults(func=change_master_copy)	4✔
378
379	# Update Safe to last version
380	parser_update_version = subparsers.add_parser("update")	4✔
381	parser_update_version.set_defaults(func=update_version)	4✔
382
383	# Update non L2 Safe to L2 Safe
384	parser_update_version_to_l2 = subparsers.add_parser("update_version_to_l2")	4✔
385	parser_update_version_to_l2.add_argument(	4✔
386	"migration_contract", type=check_ethereum_address	4✔
387	)
388	parser_update_version_to_l2.set_defaults(func=update_version_to_l2)	4✔
389
390	# Send custom/ether/erc20/erc721
391	parser_send_custom = subparsers.add_parser("send_custom")	4✔
392	parser_send_ether = subparsers.add_parser("send_ether")	4✔
393	parser_send_erc20 = subparsers.add_parser("send_erc20")	4✔
394	parser_send_erc721 = subparsers.add_parser("send_erc721")	4✔
395	parser_send_custom.set_defaults(func=send_custom)	4✔
396	parser_send_ether.set_defaults(func=send_ether)	4✔
397	parser_send_erc20.set_defaults(func=send_erc20)	4✔
398	parser_send_erc721.set_defaults(func=send_erc721)	4✔
399	# They have some common arguments
400	for parser in (	4✔
401	parser_send_custom,	4✔
402	parser_send_ether,	4✔
403	parser_send_erc20,	4✔
404	parser_send_erc721,	4✔
405	):
406	parser.add_argument(	4✔
407	"--safe-nonce",	4✔
408	type=int,	4✔
409	help="Use custom safe nonce instead of "	4✔
410	"the one for last executed SafeTx + 1",
411	)
412
413	# To/value is common for send custom and send ether
414	for parser in (parser_send_custom, parser_send_ether):	4✔
415	parser.add_argument("to", type=check_ethereum_address)	4✔
416	parser.add_argument("value", type=int)	4✔
417
418	parser_send_custom.add_argument("data", type=check_hex_str)	4✔
419	parser_send_custom.add_argument(	4✔
420	"--delegate", action="store_true", help="Use DELEGATE_CALL. By default use CALL"	4✔
421	)
422
423	# Send erc20/721 have common arguments
424	for parser in (parser_send_erc20, parser_send_erc721):	4✔
425	parser.add_argument("to", type=check_ethereum_address)	4✔
426	parser.add_argument("token_address", type=check_ethereum_address)	4✔
427
428	parser_send_erc20.add_argument("amount", type=int)	4✔
429	parser_send_erc721.add_argument("token_id", type=int)	4✔
430
431	# Drain only needs receiver account
432	parser_drain = subparsers.add_parser("drain")	4✔
433	parser_drain.set_defaults(func=drain)	4✔
434	parser_drain.add_argument("to", type=check_ethereum_address)	4✔
435
436	# Retrieve threshold, nonce or owners
437	parser_get_threshold = subparsers.add_parser("get_threshold")	4✔
438	parser_get_threshold.set_defaults(func=get_threshold)	4✔
439
440	parser_get_nonce = subparsers.add_parser("get_nonce")	4✔
441	parser_get_nonce.set_defaults(func=get_nonce)	4✔
442
443	parser_get_owners = subparsers.add_parser("get_owners")	4✔
444	parser_get_owners.set_defaults(func=get_owners)	4✔
445
446	# Enable and disable modules
447	parser_enable_module = subparsers.add_parser("enable_module")	4✔
448	parser_enable_module.add_argument("address", type=check_ethereum_address)	4✔
449	parser_enable_module.set_defaults(func=enable_module)	4✔
450
451	parser_disable_module = subparsers.add_parser("disable_module")	4✔
452	parser_disable_module.add_argument("address", type=check_ethereum_address)	4✔
453	parser_disable_module.set_defaults(func=disable_module)	4✔
454
455	# Info and refresh
456	parser_info = subparsers.add_parser("info")	4✔
457	parser_info.set_defaults(func=get_info)	4✔
458
459	parser_refresh = subparsers.add_parser("refresh")	4✔
460	parser_refresh.set_defaults(func=get_refresh)	4✔
461
462	# Tx-Service
463	# TODO Use subcommands
464	parser_tx_service = subparsers.add_parser("balances")	4✔
465	parser_tx_service.set_defaults(func=get_balances)	4✔
466
467	parser_tx_service = subparsers.add_parser("history")	4✔
468	parser_tx_service.set_defaults(func=get_history)	4✔
469
470	parser_tx_service = subparsers.add_parser("sign-tx")	4✔
471	parser_tx_service.set_defaults(func=sign_tx)	4✔
472	parser_tx_service.add_argument("safe_tx_hash", type=check_hex_str)	4✔
473
474	parser_tx_service = subparsers.add_parser("batch-txs")	4✔
475	parser_tx_service.set_defaults(func=batch_txs)	4✔
476	parser_tx_service.add_argument("safe_nonce", type=int)	4✔
477	parser_tx_service.add_argument("safe_tx_hashes", type=check_hex_str, nargs="+")	4✔
478
479	parser_tx_service = subparsers.add_parser("execute-tx")	4✔
480	parser_tx_service.set_defaults(func=execute_tx)	4✔
481	parser_tx_service.add_argument("safe_tx_hash", type=check_hex_str)	4✔
482
483	# List delegates
484	parser_delegates = subparsers.add_parser("get_delegates")	4✔
485	parser_delegates.set_defaults(func=get_delegates)	4✔
486
487	# Add delegate
488	parser_add_delegate = subparsers.add_parser("add_delegate")	4✔
489	parser_add_delegate.set_defaults(func=add_delegate)	4✔
490	parser_add_delegate.add_argument("address", type=check_ethereum_address)	4✔
491	parser_add_delegate.add_argument("label", type=str)	4✔
492	parser_add_delegate.add_argument("signer", type=check_ethereum_address)	4✔
493
494	# Remove delegate
495	parser_remove_delegate = subparsers.add_parser("remove_delegate")	4✔
496	parser_remove_delegate.set_defaults(func=remove_delegate)	4✔
497	parser_remove_delegate.add_argument("address", type=check_ethereum_address)	4✔
498	parser_remove_delegate.add_argument("signer", type=check_ethereum_address)	4✔
499
500	return prompt_parser	4✔

safe-global / safe-cli / 7087242091

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