9679828121

Committed 26 Jun 2024 12:41PM UTC coverage: 87.47% (+0.2%) from 87.279%

Build # 9679828121

Build Type

push

github

Committed by

web-flow

Commit Message

Adding support for executing transactions in scripting mode (#419)

* Add send_ether command

* Add send_custom, send_erc20 and send_erc721 commands

* Add tx_builder command

* Add support for env keys

* Add test to tx_builder_decoder

* Update Readme

* Fix encoding of tuple type

* Merge safe-cli and safe-runner

* Apply PR suggestions

* Revert tx_builder command

* Update feedback messages to user.

* Improve default command validation

* Add interactive/no-interactive option

* Update common parameters

Run Details

750 of 869 branches covered (86.31%)

Branch coverage included in aggregate %.

291 of 319 new or added lines in 7 files covered. (91.22%)

6 existing lines in 2 files now uncovered.

2552 of 2906 relevant lines covered (87.82%)

3.51 hits per line

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

87.84

/src/safe_cli/safe_cli.py

import argparse
import os
import sys
from typing import Optional

from eth_typing import ChecksumAddress
from prompt_toolkit import HTML, PromptSession, print_formatted_text
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.history import FileHistory
from prompt_toolkit.lexers import PygmentsLexer

from .operators import (
    SafeCliTerminationException,
    SafeOperator,
    SafeServiceNotAvailable,
    SafeTxServiceOperator,
)
from .prompt_parser import PromptParser
from .safe_completer import SafeCompleter
from .safe_lexer import SafeLexer


class SafeCli:
    def __init__(self, safe_address: ChecksumAddress, node_url: str, history: bool):
        """
        :param safe_address: Safe address
        :param node_url: Ethereum RPC url
        :param history: If `True` keep command history, otherwise history is not kept after closing the CLI
        """
        self.safe_address = safe_address
        self.node_url = node_url
        if history:
            self.session = PromptSession(
                history=FileHistory(os.path.join(sys.path[0], ".history"))
            )
        else:
            self.session = PromptSession()
        self.safe_operator = SafeOperator(safe_address, node_url)
        self.prompt_parser = PromptParser(self.safe_operator)

    def print_startup_info(self):
        print_formatted_text(
            HTML("<b><ansigreen>Loading Safe information...</ansigreen></b>")
        )
        self.safe_operator.print_info()

        print_formatted_text(
            HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
        )
        print_formatted_text(
            HTML(
                "The <b>help</b> command displays all available options and the <b>exit</b> command terminates the safe-cli."
            )
        )

    def get_prompt_text(self):
        mode: Optional[str] = "blockchain"
        if isinstance(self.prompt_parser.safe_operator, SafeTxServiceOperator):
            mode = "tx-service"

        return HTML(
            f"<bold><ansiblue>{mode} > {self.safe_address}</ansiblue><ansired> > </ansired></bold>"
        )

    def get_bottom_toolbar(self):
        return HTML(
            f'<b><style fg="ansiyellow">network={self.safe_operator.network.name} '
            f"{self.safe_operator.safe_cli_info}</style></b>"
        )

    def parse_operator_mode(self, command: str) -> Optional[SafeOperator]:
        """
        Parse operator mode to switch between blockchain (default) and tx-service
        :param command:
        :return: SafeOperator if detected
        """
        split_command = command.split()
        try:
            if (split_command[0]) == "tx-service":
                print_formatted_text(
                    HTML("<b><ansigreen>Sending txs to tx service</ansigreen></b>")
                )
                return SafeTxServiceOperator(self.safe_address, self.node_url)
            elif split_command[0] == "blockchain":
                print_formatted_text(
                    HTML("<b><ansigreen>Sending txs to blockchain</ansigreen></b>")
                )
                return self.safe_operator
        except SafeServiceNotAvailable:
            print_formatted_text(
                HTML("<b><ansired>Mode not supported on this network</ansired></b>")
            )

    def get_command(self) -> str:
        return self.session.prompt(
            self.get_prompt_text,
            auto_suggest=AutoSuggestFromHistory(),
            bottom_toolbar=self.get_bottom_toolbar,
            lexer=PygmentsLexer(SafeLexer),
            completer=SafeCompleter(),
        )

    def loop(self):
        while True:
            try:
                command = self.get_command()
                if not command.strip():
                    continue

                new_operator = self.parse_operator_mode(command)
                if new_operator:
                    self.prompt_parser = PromptParser(new_operator)
                    new_operator.refresh_safe_cli_info()  # ClI info needs to be initialized
                else:
                    self.prompt_parser.process_command(command)
            except SafeCliTerminationException:
                break
            except EOFError:
                break
            except KeyboardInterrupt:
                continue
            except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):
                pass

1	import argparse	4✔
2	import os	4✔
3	import sys	4✔
4	from typing import Optional	4✔
5
6	from eth_typing import ChecksumAddress	4✔
7	from prompt_toolkit import HTML, PromptSession, print_formatted_text	4✔
8	from prompt_toolkit.auto_suggest import AutoSuggestFromHistory	4✔
9	from prompt_toolkit.history import FileHistory	4✔
10	from prompt_toolkit.lexers import PygmentsLexer	4✔
11
12	from .operators import (	4✔
13	SafeCliTerminationException,
14	SafeOperator,
15	SafeServiceNotAvailable,
16	SafeTxServiceOperator,
17	)
18	from .prompt_parser import PromptParser	4✔
19	from .safe_completer import SafeCompleter	4✔
20	from .safe_lexer import SafeLexer	4✔
21
22
23	class SafeCli:	4✔
24	def __init__(self, safe_address: ChecksumAddress, node_url: str, history: bool):	4✔
25	"""
26	:param safe_address: Safe address
27	:param node_url: Ethereum RPC url
28	:param history: If `True` keep command history, otherwise history is not kept after closing the CLI
29	"""
30	self.safe_address = safe_address	4✔
31	self.node_url = node_url	4✔
32	if history:	4✔
33	self.session = PromptSession(	4✔
34	history=FileHistory(os.path.join(sys.path[0], ".history"))
35	)
36	else:
NEW 37	self.session = PromptSession()	×
38	self.safe_operator = SafeOperator(safe_address, node_url)	4✔
39	self.prompt_parser = PromptParser(self.safe_operator)	4✔
40
41	def print_startup_info(self):	4✔
42	print_formatted_text(	4✔
43	HTML("<b><ansigreen>Loading Safe information...</ansigreen></b>")
44	)
45	self.safe_operator.print_info()	4✔
46
47	print_formatted_text(	4✔
48	HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
49	)
50	print_formatted_text(	4✔
51	HTML(
52	"The <b>help</b> command displays all available options and the <b>exit</b> command terminates the safe-cli."
53	)
54	)
55
56	def get_prompt_text(self):	4✔
57	mode: Optional[str] = "blockchain"	4✔
58	if isinstance(self.prompt_parser.safe_operator, SafeTxServiceOperator):	4✔
NEW 59	mode = "tx-service"	×
60
61	return HTML(	4✔
62	f"<bold><ansiblue>{mode} > {self.safe_address}</ansiblue><ansired> > </ansired></bold>"
63	)
64
65	def get_bottom_toolbar(self):	4✔
NEW 66	return HTML(	×
67	f'<b><style fg="ansiyellow">network={self.safe_operator.network.name} '
68	f"{self.safe_operator.safe_cli_info}</style></b>"
69	)
70
71	def parse_operator_mode(self, command: str) -> Optional[SafeOperator]:	4✔
72	"""
73	Parse operator mode to switch between blockchain (default) and tx-service
74	:param command:
75	:return: SafeOperator if detected
76	"""
77	split_command = command.split()	4✔
78	try:	4✔
79	if (split_command[0]) == "tx-service":	4✔
80	print_formatted_text(	4✔
81	HTML("<b><ansigreen>Sending txs to tx service</ansigreen></b>")
82	)
83	return SafeTxServiceOperator(self.safe_address, self.node_url)	4✔
84	elif split_command[0] == "blockchain":	4✔
NEW 85	print_formatted_text(	×
86	HTML("<b><ansigreen>Sending txs to blockchain</ansigreen></b>")
87	)
NEW 88	return self.safe_operator	×
89	except SafeServiceNotAvailable:	4✔
90	print_formatted_text(	4✔
91	HTML("<b><ansired>Mode not supported on this network</ansired></b>")
92	)
93
94	def get_command(self) -> str:	4✔
95	return self.session.prompt(	4✔
96	self.get_prompt_text,
97	auto_suggest=AutoSuggestFromHistory(),
98	bottom_toolbar=self.get_bottom_toolbar,
99	lexer=PygmentsLexer(SafeLexer),
100	completer=SafeCompleter(),
101	)
102
103	def loop(self):	4✔
104	while True:	3✔
105	try:	4✔
106	command = self.get_command()	4✔
107	if not command.strip():	4✔
NEW 108	continue	×
109
110	new_operator = self.parse_operator_mode(command)	4✔
111	if new_operator:	4✔
NEW 112	self.prompt_parser = PromptParser(new_operator)	×
NEW 113	new_operator.refresh_safe_cli_info() # ClI info needs to be initialized	×
114	else:
115	self.prompt_parser.process_command(command)	4✔
116	except SafeCliTerminationException:	4✔
117	break	4✔
118	except EOFError:	4✔
119	break	4✔
120	except KeyboardInterrupt:	4✔
NEW 121	continue	×
122	except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):	4✔
123	pass

safe-global / safe-cli / 9679828121

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