6508017815

Committed 13 Oct 2023 12:09PM UTC coverage: 83.849% (-0.4%) from 84.269%

Build # 6508017815

Build Type

Pull #277

github

Committed by

web-flow

Commit Message

Merge f49030687 into 688bc1673

Pull Request Pull Request #277: Remove relaying

Run Details

867 of 1034 relevant lines covered (83.85%)

3.35 hits per line

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

83.08

/safe_cli/prompt_parser.py

import argparse
import functools

from hexbytes import HexBytes
from prompt_toolkit import HTML, print_formatted_text
from web3 import Web3

from .api.base_api import BaseAPIException
from .operators.safe_operator import (
    AccountNotLoadedException,
    ExistingOwnerException,
    FallbackHandlerNotSupportedException,
    HashAlreadyApproved,
    InvalidMasterCopyException,
    NonExistingOwnerException,
    NotEnoughEtherToSend,
    NotEnoughSignatures,
    NotEnoughTokenToSend,
    SafeAlreadyUpdatedException,
    SafeOperator,
    SafeServiceNotAvailable,
    SameFallbackHandlerException,
    SameMasterCopyException,
    SenderRequiredException,
    ThresholdLimitException,
)


def check_ethereum_address(address: str) -> str:
    """
    Ethereum address validator for ArgParse
    :param address:
    :return:
    """
    if not Web3.is_checksum_address(address):
        raise argparse.ArgumentTypeError(
            f"{address} is not a valid checksummed ethereum address"
        )
    return address


def check_hex_str(hex_str: str) -> HexBytes:
    """
    Hexadecimal
    :param hex_str:
    :return:
    """
    try:
        return HexBytes(hex_str)
    except ValueError:
        raise argparse.ArgumentTypeError(f"{hex_str} is not a valid hexadecimal string")


def check_keccak256_hash(hex_str: str) -> HexBytes:
    """
    Hexadecimal
    :param hex_str:
    :return:
    """
    hex_str_bytes = check_hex_str(hex_str)
    if len(hex_str_bytes) != 32:
        raise argparse.ArgumentTypeError(
            f"{hex_str} is not a valid keccak256 hash hexadecimal string"
        )
    return hex_str_bytes


def to_checksummed_ethereum_address(address: str) -> str:
    try:
        return Web3.to_checksum_address(address)
    except ValueError:
        raise argparse.ArgumentTypeError(f"{address} is not a valid ethereum address")


def safe_exception(function):
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        try:
            return function(*args, **kwargs)
        except BaseAPIException 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 SafeAlreadyUpdatedException:
            print_formatted_text(HTML("<ansired>Safe is already updated</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>"
                )
            )

    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 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 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_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_master_copy = subparsers.add_parser("change_fallback_handler")
    parser_change_master_copy.add_argument("address", type=check_ethereum_address)
    parser_change_master_copy.set_defaults(func=change_fallback_handler)

    # Change FallbackHandler
    parser_change_master_copy = subparsers.add_parser("change_guard")
    parser_change_master_copy.add_argument("address", type=check_ethereum_address)
    parser_change_master_copy.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_change_master_copy = subparsers.add_parser("update")
    parser_change_master_copy.set_defaults(func=update_version)

    # 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_drain = subparsers.add_parser("drain")
    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)
    parser_drain.set_defaults(func=drain)
    # 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.add_argument("amount", type=int)

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

safe-global / safe-cli / 6508017815

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