9464772673

Committed 11 Jun 2024 11:32AM UTC coverage: 87.16% (+6.5%) from 80.617%

Build # 9464772673

Build Type

push

github

Committed by

web-flow

Commit Message

Update project to use hatch (#404)

* Update project to use hatch

- Update pre-commit, as black previous version was having issues
- Use recommended structure: https://docs.pytest.org/en/stable/explanation/goodpractices.html
- Update CI
- Add run_tests.sh script

* Fix coverage

* Fix version

* Fix module export

* Fix linting

---------

Co-authored-by: Uxio Fuentefria <6909403+Uxio0@users.noreply.github.com>

Run Details

722 of 835 branches covered (86.47%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 2 files covered. (100.0%)

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

79.35

/src/safe_cli/main.py

#!/bin/env python3
import argparse
import os
import sys
from typing import Optional

from art import text2art
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 safe_cli.argparse_validators import check_ethereum_address
from safe_cli.operators import (
    SafeOperator,
    SafeServiceNotAvailable,
    SafeTxServiceOperator,
)
from safe_cli.prompt_parser import PromptParser
from safe_cli.safe_completer import SafeCompleter
from safe_cli.safe_lexer import SafeLexer
from safe_cli.utils import get_safe_from_owner

from . import VERSION


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(text2art("Safe CLI"))  # Print fancy text
        print_formatted_text(HTML(f"<b><ansigreen>Version {VERSION}</ansigreen></b>"))
        print_formatted_text(
            HTML("<b><ansigreen>Loading Safe information...</ansigreen></b>")
        )
        self.safe_operator.print_info()

    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 EOFError:
                break
            except KeyboardInterrupt:
                continue
            except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):
                pass


def build_safe_cli() -> Optional[SafeCli]:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "address",
        help="The address of the Safe, or an owner address if --get-safes-from-owner is specified.",
        type=check_ethereum_address,
    )
    parser.add_argument("node_url", help="Ethereum node url")
    parser.add_argument(
        "--history",
        action="store_true",
        help="Enable history. By default it's disabled due to security reasons",
        default=False,
    )
    parser.add_argument(
        "--get-safes-from-owner",
        action="store_true",
        help="Indicates that address is an owner (Safe Transaction Service is required for this feature)",
        default=False,
    )

    args = parser.parse_args()
    if args.get_safes_from_owner:
        if (
            safe_address := get_safe_from_owner(args.address, args.node_url)
        ) is not None:
            return SafeCli(safe_address, args.node_url, args.history)
    else:
        return SafeCli(args.address, args.node_url, args.history)


def main(*args, **kwargs):
    safe_cli = build_safe_cli()
    safe_cli.print_startup_info()
    safe_cli.loop()


if __name__ == "__main__":
    main()

1	#!/bin/env python3
2	import argparse	4✔
3	import os	4✔
4	import sys	4✔
5	from typing import Optional	4✔
6
7	from art import text2art	4✔
8	from eth_typing import ChecksumAddress	4✔
9	from prompt_toolkit import HTML, PromptSession, print_formatted_text	4✔
10	from prompt_toolkit.auto_suggest import AutoSuggestFromHistory	4✔
11	from prompt_toolkit.history import FileHistory	4✔
12	from prompt_toolkit.lexers import PygmentsLexer	4✔
13
14	from safe_cli.argparse_validators import check_ethereum_address	4✔
15	from safe_cli.operators import (	4✔
16	SafeOperator,
17	SafeServiceNotAvailable,
18	SafeTxServiceOperator,
19	)
20	from safe_cli.prompt_parser import PromptParser	4✔
21	from safe_cli.safe_completer import SafeCompleter	4✔
22	from safe_cli.safe_lexer import SafeLexer	4✔
23	from safe_cli.utils import get_safe_from_owner	4✔
24
25	from . import VERSION	4✔
26
27
28	class SafeCli:	4✔
29	def __init__(self, safe_address: ChecksumAddress, node_url: str, history: bool):	4✔
30	"""
31	:param safe_address: Safe address
32	:param node_url: Ethereum RPC url
33	:param history: If `True` keep command history, otherwise history is not kept after closing the CLI
34	"""
35	self.safe_address = safe_address	4✔
36	self.node_url = node_url	4✔
37	if history:	4✔
38	self.session = PromptSession(	4✔
39	history=FileHistory(os.path.join(sys.path[0], ".history"))
40	)
41	else:
42	self.session = PromptSession()	×
43	self.safe_operator = SafeOperator(safe_address, node_url)	4✔
44	self.prompt_parser = PromptParser(self.safe_operator)	4✔
45
46	def print_startup_info(self):	4✔
47	print_formatted_text(text2art("Safe CLI")) # Print fancy text	4✔
48	print_formatted_text(HTML(f"<b><ansigreen>Version {VERSION}</ansigreen></b>"))	4✔
49	print_formatted_text(	4✔
50	HTML("<b><ansigreen>Loading Safe information...</ansigreen></b>")
51	)
52	self.safe_operator.print_info()	4✔
53
54	def get_prompt_text(self):	4✔
55	mode: Optional[str] = "blockchain"	4✔
56	if isinstance(self.prompt_parser.safe_operator, SafeTxServiceOperator):	4✔
57	mode = "tx-service"	×
58
59	return HTML(	4✔
60	f"<bold><ansiblue>{mode} > {self.safe_address}</ansiblue><ansired> > </ansired></bold>"
61	)
62
63	def get_bottom_toolbar(self):	4✔
64	return HTML(	4✔
65	f'<b><style fg="ansiyellow">network={self.safe_operator.network.name} '
66	f"{self.safe_operator.safe_cli_info}</style></b>"
67	)
68
69	def parse_operator_mode(self, command: str) -> Optional[SafeOperator]:	4✔
70	"""
71	Parse operator mode to switch between blockchain (default) and tx-service
72	:param command:
73	:return: SafeOperator if detected
74	"""
75	split_command = command.split()	4✔
76	try:	4✔
77	if (split_command[0]) == "tx-service":	4✔
78	print_formatted_text(	4✔
79	HTML("<b><ansigreen>Sending txs to tx service</ansigreen></b>")
80	)
81	return SafeTxServiceOperator(self.safe_address, self.node_url)	4✔
82	elif split_command[0] == "blockchain":	4✔
83	print_formatted_text(	4✔
84	HTML("<b><ansigreen>Sending txs to blockchain</ansigreen></b>")
85	)
86	return self.safe_operator	4✔
87	except SafeServiceNotAvailable:	4✔
88	print_formatted_text(	4✔
89	HTML("<b><ansired>Mode not supported on this network</ansired></b>")
90	)
91
92	def get_command(self) -> str:	4✔
93	return self.session.prompt(	×
94	self.get_prompt_text,
95	auto_suggest=AutoSuggestFromHistory(),
96	bottom_toolbar=self.get_bottom_toolbar,
97	lexer=PygmentsLexer(SafeLexer),
98	completer=SafeCompleter(),
99	)
100
101	def loop(self):	4✔
102	while True:	3✔
103	try:	4✔
104	command = self.get_command()	4✔
105	if not command.strip():	×
106	continue	×
107
108	new_operator = self.parse_operator_mode(command)	×
109	if new_operator:	×
110	self.prompt_parser = PromptParser(new_operator)	×
111	new_operator.refresh_safe_cli_info() # ClI info needs to be initialized	×
112	else:
113	self.prompt_parser.process_command(command)	×
114	except EOFError:	4✔
115	break	4✔
116	except KeyboardInterrupt:	×
117	continue	×
118	except (argparse.ArgumentError, argparse.ArgumentTypeError, SystemExit):	×
119	pass
120
121
122	def build_safe_cli() -> Optional[SafeCli]:	4✔
123	parser = argparse.ArgumentParser()	4✔
124	parser.add_argument(	4✔
125	"address",
126	help="The address of the Safe, or an owner address if --get-safes-from-owner is specified.",
127	type=check_ethereum_address,
128	)
129	parser.add_argument("node_url", help="Ethereum node url")	4✔
130	parser.add_argument(	4✔
131	"--history",
132	action="store_true",
133	help="Enable history. By default it's disabled due to security reasons",
134	default=False,
135	)
136	parser.add_argument(	4✔
137	"--get-safes-from-owner",
138	action="store_true",
139	help="Indicates that address is an owner (Safe Transaction Service is required for this feature)",
140	default=False,
141	)
142
143	args = parser.parse_args()	4✔
144	if args.get_safes_from_owner:	4✔
145	if (	4✔
146	safe_address := get_safe_from_owner(args.address, args.node_url)
147	) is not None:
148	return SafeCli(safe_address, args.node_url, args.history)	4✔
149	else:
150	return SafeCli(args.address, args.node_url, args.history)	4✔
151
152
153	def main(args, *kwargs):	4✔
154	safe_cli = build_safe_cli()	×
155	safe_cli.print_startup_info()	×
156	safe_cli.loop()	×
157
158
159	if __name__ == "__main__":
160	main()

safe-global / safe-cli / 9464772673

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