black-ify everything

Author: willcl-ark

pyproject.toml

@@ -13,6 +13,7 @@ classifiers = [
     "Programming Language :: Python :: 3",
 ]
 dependencies = [
+    "black==23.7.0",
     "click==8.1.7",
     "docker==6.1.3",
     "flask==2.3.3",

src/warnet/cli.py

@@ -29,19 +29,28 @@ def rpc_call(rpc_method, params: Optional[Union[Dict[str, Any], Tuple[Any, ...]]
 def cli():
     pass
 
+
 @click.group(name="debug")
 def debug():
     """Debug commands"""
+
+
 cli.add_command(debug)
 
+
 @click.group(name="scenarios")
 def scenarios():
     """Scenario commands"""
+
+
 cli.add_command(scenarios)
 
+
 @click.group(name="network")
 def network():
     """Network commands"""
+
+
 cli.add_command(network)
 
 
@@ -69,31 +78,42 @@ def help_command(ctx, command):
     help_info = cmd_obj.get_help(ctx).split("\n", 1)[-1].strip()
 
     # Extract the arguments of the command
-    arguments = [param.human_readable_name.upper() for param in cmd_obj.params if isinstance(param, click.Argument)]
+    arguments = [
+        param.human_readable_name.upper()
+        for param in cmd_obj.params
+        if isinstance(param, click.Argument)
+    ]
 
     # Determine the correct usage string based on whether the command has subcommands
     if isinstance(cmd_obj, click.Group) and cmd_obj.list_commands(ctx):
-        usage_str = f"Usage: warnet {command} [OPTIONS] COMMAND [ARGS...]\n\n{help_info}"
+        usage_str = (
+            f"Usage: warnet {command} [OPTIONS] COMMAND [ARGS...]\n\n{help_info}"
+        )
     else:
         args_str = " ".join(arguments)
         usage_str = f"Usage: warnet {command} [OPTIONS] {args_str}\n\n{help_info}"
- 
+
     print(usage_str)
 
+
 cli.add_command(help_command)
 
 
 @cli.command(context_settings={"ignore_unknown_options": True})
-@click.argument('node', type=int)
-@click.argument('method', type=str, nargs=-1)  # this will capture all remaining arguments
-@click.option('--params', type=str, multiple=True, default=())
-@click.option('--network', default='warnet', show_default=True)
+@click.argument("node", type=int)
+@click.argument(
+    "method", type=str, nargs=-1
+)  # this will capture all remaining arguments
+@click.option("--params", type=str, multiple=True, default=())
+@click.option("--network", default="warnet", show_default=True)
 def rpc(node, method, params, network):
     """
     Call bitcoin-cli <method> <params> on <node> in <--network>
     """
     if len(method) > 2:
-        raise click.BadArgumentUsage("You can provide at most two arguments for 'method'.")
+        raise click.BadArgumentUsage(
+            "You can provide at most two arguments for 'method'."
+        )
 
     # Convert tuple to space-separated string
     method_str = " ".join(method)
@@ -109,8 +129,8 @@ def rpc(node, method, params, network):
 
 
 @cli.command()
-@click.argument('node', type=int, required=True)
-@click.option('--network', default='warnet', show_default=True)
+@click.argument("node", type=int, required=True)
+@click.option("--network", default="warnet", show_default=True)
 def debug_log(node, network):
     """
     Fetch the Bitcoin Core debug log from <node> in [network]
@@ -123,14 +143,15 @@ def debug_log(node, network):
 
 
 @cli.command()
-@click.argument('node_a', type=int, required=True)
-@click.argument('node_b', type=int, required=True)
-@click.option('--network', default='warnet', show_default=True)
+@click.argument("node_a", type=int, required=True)
+@click.argument("node_b", type=int, required=True)
+@click.option("--network", default="warnet", show_default=True)
 def messages(node_a, node_b, network):
     """
     Fetch messages sent between <node_a> and <node_b> in <network>
     """
     import logging
+
     logging.warning(f"got args: {node_a}, {node_b}, {network}")
     try:
         result = rpc_call(
@@ -156,7 +177,7 @@ def list():
 
 
 @scenarios.command()
-@click.argument('scenario', type=str)
+@click.argument("scenario", type=str)
 def run(scenario):
     """
     Run <scenario> from the Warnet Test Framework
@@ -169,49 +190,59 @@ def run(scenario):
 
 
 @debug.command()
-@click.argument('graph_file', type=str)
-@click.option('--network', default='warnet', show_default=True)
+@click.argument("graph_file", type=str)
+@click.option("--network", default="warnet", show_default=True)
 def generate_compose(graph_file: str, network: str = "warnet"):
     """
     Generate the docker-compose file for a given <graph_file> and <--network> name and return it.
     """
     try:
-        result = rpc_call("generate_compose", {"graph_file": graph_file, "network": network})
+        result = rpc_call(
+            "generate_compose", {"graph_file": graph_file, "network": network}
+        )
         print(result)
     except Exception as e:
         print(f"Error generating compose: {e}")
 
+
 @debug.command()
-@click.argument('graph_file', type=str)
-@click.option('--network', default='warnet', show_default=True)
+@click.argument("graph_file", type=str)
+@click.option("--network", default="warnet", show_default=True)
 def update_dns_seed(graph_file: Path = EXAMPLE_GRAPH_FILE, network: str = "warnet"):
     """
     Update the dns seed database using a graph file
     """
     try:
-        result = rpc_call("update_dns_seeder", {"graph_file": str(graph_file), "network": network})
+        result = rpc_call(
+            "update_dns_seeder", {"graph_file": str(graph_file), "network": network}
+        )
         print(result)
     except Exception as e:
         print(f"Error updating dns seed addresses: {e}")
 
 
 @network.command()
-@click.argument('graph_file', default=EXAMPLE_GRAPH_FILE, type=click.Path())
-@click.option('--force', default=False, is_flag=True, type=bool)
-@click.option('--network', default='warnet', show_default=True)
-def start(graph_file: Path = EXAMPLE_GRAPH_FILE, force: bool = False, network: str = "warnet"):
+@click.argument("graph_file", default=EXAMPLE_GRAPH_FILE, type=click.Path())
+@click.option("--force", default=False, is_flag=True, type=bool)
+@click.option("--network", default="warnet", show_default=True)
+def start(
+    graph_file: Path = EXAMPLE_GRAPH_FILE, force: bool = False, network: str = "warnet"
+):
     """
     Start a warnet with topology loaded from a <graph_file> into <--network> (default: "warnet")
     """
     try:
-        result = rpc_call("from_file", {"graph_file": str(graph_file), "force": force, "network": network})
+        result = rpc_call(
+            "from_file",
+            {"graph_file": str(graph_file), "force": force, "network": network},
+        )
         print(result)
     except Exception as e:
         print(f"Error creating network: {e}")
 
 
 @network.command()
-@click.option('--network', default='warnet', show_default=True)
+@click.option("--network", default="warnet", show_default=True)
 def up(network: str = "warnet"):
     """
     Run 'docker-compose up' on a warnet named <--network> (default: "warnet").
@@ -224,7 +255,7 @@ def up(network: str = "warnet"):
 
 
 @network.command()
-@click.option('--network', default='warnet', show_default=True)
+@click.option("--network", default="warnet", show_default=True)
 def down(network: str = "warnet"):
     """
     Run 'docker-compose down on a warnet named <--network> (default: "warnet").

src/warnet/client.py

@@ -72,10 +72,12 @@ def stop_container(c):
     logger.info(f"stopping container: {c.name}")
     c.stop()
 
+
 def stop_network(network="warnet") -> bool:
     """
     Stop all containers in the network in parallel using a background thread
     """
+
     def thread_stop():
         d = docker.from_env()
         network_obj = d.networks.get(network)
@@ -87,6 +89,7 @@ def thread_stop():
     threading.Thread(target=thread_stop).start()
     return True
 
+
 def compose_down(network="warnet") -> bool:
     """
     Run docker-compose down on a warnet

src/warnet/tank.py

@@ -45,14 +45,16 @@ def __init__(self):
         self.config_dir = Path()
 
     def __str__(self) -> str:
-        return (f"Tank(\n"
-                f"\tIndex: {self.index}\n"
-                f"\tVersion: {self.version}\n"
-                f"\tConf: {self.conf}\n"
-                f"\tConf File: {self.conf_file}\n"
-                f"\tNetem: {self.netem}\n"
-                f"\tIPv4: {self._ipv4}\n"
-                f"\t)")
+        return (
+            f"Tank(\n"
+            f"\tIndex: {self.index}\n"
+            f"\tVersion: {self.version}\n"
+            f"\tConf: {self.conf}\n"
+            f"\tConf File: {self.conf_file}\n"
+            f"\tNetem: {self.netem}\n"
+            f"\tIPv4: {self._ipv4}\n"
+            f"\t)"
+        )
 
     @classmethod
     def from_graph_node(cls, index, warnet):
@@ -67,14 +69,17 @@ def from_graph_node(cls, index, warnet):
         if "version" in node:
             if not "/" and "#" in self.version:
                 if node["version"] not in SUPPORTED_TAGS:
-                    raise Exception(f"Unsupported version: can't be generated from Docker images: {node['version']}")
+                    raise Exception(
+                        f"Unsupported version: can't be generated from Docker images: {node['version']}"
+                    )
             self.version = node["version"]
         if "bitcoin_config" in node:
             self.conf = node["bitcoin_config"]
         if "tc_netem" in node:
             self.netem = node["tc_netem"]
         with open(self.warnet.fork_observer_config, "a") as f:
-            f.write(f'''
+            f.write(
+                f"""
     [[networks.nodes]]
     id = {self.index}
     name = "Node {self.index}"
@@ -83,7 +88,8 @@ def from_graph_node(cls, index, warnet):
     rpc_port = {self.rpc_port}
     rpc_user = "{self.rpc_user}"
     rpc_password = "{self.rpc_password}"
-''')
+"""
+            )
         self.config_dir = self.warnet.config_dir / str(self.suffix)
         self.config_dir.mkdir(parents=True, exist_ok=True)
         self.write_torrc()
@@ -133,8 +139,10 @@ def container(self) -> Container:
     def exec(self, cmd: str, user: str = "root"):
         result = self.container.exec_run(cmd=cmd, user=user)
         if result.exit_code != 0:
-            raise Exception(f"Command failed with exit code {result.exit_code}: {result.output.decode('utf-8')}")
-        return result.output.decode('utf-8')
+            raise Exception(
+                f"Command failed with exit code {result.exit_code}: {result.output.decode('utf-8')}"
+            )
+        return result.output.decode("utf-8")
 
     def apply_network_conditions(self):
         if self.netem is None:
@@ -181,7 +189,7 @@ def write_bitcoin_conf(self, base_bitcoin_conf):
         self.conf_file = path
 
     def write_torrc(self):
-        src_tor_conf_file = TEMPLATES / 'torrc'
+        src_tor_conf_file = TEMPLATES / "torrc"
 
         dest_path = self.config_dir / "torrc"
         shutil.copyfile(src_tor_conf_file, dest_path)
@@ -214,23 +222,23 @@ def add_services(self, services):
             services[self.bitcoind_name].update({"entrypoint": "/warnet_entrypoint.sh"})
 
         # Add the bitcoind service
-        services[self.bitcoind_name].update({
-            "container_name": self.bitcoind_name,
-            "build": build,
-            "volumes": [
-                f"{self.conf_file}:/home/bitcoin/.bitcoin/bitcoin.conf",
-                f"{self.torrc_file}:/etc/tor/torrc_original",
-            ],
-            "networks": {
-                self.docker_network: {
-                    "ipv4_address": f"{self.ipv4}",
-                }
-            },
-            "labels": {
-                "warnet": "tank"
-            },
-            "privileged": True,
-        })
+        services[self.bitcoind_name].update(
+            {
+                "container_name": self.bitcoind_name,
+                "build": build,
+                "volumes": [
+                    f"{self.conf_file}:/home/bitcoin/.bitcoin/bitcoin.conf",
+                    f"{self.torrc_file}:/etc/tor/torrc_original",
+                ],
+                "networks": {
+                    self.docker_network: {
+                        "ipv4_address": f"{self.ipv4}",
+                    }
+                },
+                "labels": {"warnet": "tank"},
+                "privileged": True,
+            }
+        )
 
         # Add the prometheus data exporter in a neighboring container
         # services[self.exporter_name] = {