add missing types

Author: willcl-ark

resources/scripts/apidocs.py

@@ -6,6 +6,7 @@
 
 from click import Context
 from tabulate import tabulate
+
 from warnet.cli.main import cli
 
 file_path = Path(os.path.dirname(os.path.abspath(__file__))) / ".." / ".." / "docs" / "warcli.md"

resources/scripts/graphdocs.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 
 from tabulate import tabulate
+
 from warnet.cli.util import load_schema
 
 graph_schema = load_schema()

src/warnet/cli/bitcoin.py

@@ -5,6 +5,7 @@
 from io import BytesIO
 
 import click
+
 from test_framework.messages import ser_uint256
 from test_framework.p2p import MESSAGEMAP
 
@@ -20,14 +21,14 @@ def bitcoin():
 @click.argument("node", type=int)
 @click.argument("method", type=str)
 @click.argument("params", type=str, nargs=-1)  # this will capture all remaining arguments
-def rpc(node, method, params):
+def rpc(node: int, method: str, params: str):
     """
     Call bitcoin-cli <method> [params] on <node>
     """
     print(_rpc(node, method, params))
 
 
-def _rpc(node, method, params):
+def _rpc(node: int, method: str, params: str):
     if params:
         cmd = f"kubectl exec warnet-tank-{node} -- bitcoin-cli -regtest -rpcuser='user' -rpcpassword='password' {method} {' '.join(map(str, params))}"
     else:
@@ -37,7 +38,7 @@ def _rpc(node, method, params):
 
 @bitcoin.command()
 @click.argument("node", type=int, required=True)
-def debug_log(node):
+def debug_log(node: int):
     """
     Fetch the Bitcoin Core debug log from <node>
     """
@@ -49,7 +50,7 @@ def debug_log(node):
 @click.argument("pattern", type=str, required=True)
 @click.option("--show-k8s-timestamps", is_flag=True, default=False, show_default=True)
 @click.option("--no-sort", is_flag=True, default=False, show_default=True)
-def grep_logs(pattern, show_k8s_timestamps, no_sort):
+def grep_logs(pattern: str, show_k8s_timestamps: bool, no_sort: bool):
     """
     Grep combined bitcoind logs using regex <pattern>
     """
@@ -120,7 +121,7 @@ def grep_logs(pattern, show_k8s_timestamps, no_sort):
 @click.argument("node_a", type=int, required=True)
 @click.argument("node_b", type=int, required=True)
 @click.option("--network", default="regtest", show_default=True)
-def messages(node_a, node_b, network):
+def messages(node_a: int, node_b: int, network: str):
     """
     Fetch messages sent between <node_a> and <node_b> in [network]
     """
@@ -154,7 +155,7 @@ def messages(node_a, node_b, network):
         print(f"Error fetching messages between nodes {node_a} and {node_b}: {e}")
 
 
-def get_messages(node_a, node_b, network):
+def get_messages(node_a: int, node_b: int, network: str):
     """
     Fetch messages from the message capture files
     """
@@ -198,7 +199,7 @@ def get_messages(node_a, node_b, network):
 
 # This function is a hacked-up copy of process_file() from
 # Bitcoin Core contrib/message-capture/message-capture-parser.py
-def parse_raw_messages(blob, outbound):
+def parse_raw_messages(blob: bytes, outbound: bool):
     TIME_SIZE = 8
     LENGTH_SIZE = 4
     MSGTYPE_SIZE = 12
@@ -267,7 +268,7 @@ def parse_raw_messages(blob, outbound):
     return messages
 
 
-def to_jsonable(obj):
+def to_jsonable(obj: str):
     HASH_INTS = [
         "blockhash",
         "block_hash",

src/warnet/cli/k8s.py

@@ -1,34 +1,34 @@
 import json
-import yaml
 import tempfile
 from importlib.resources import files
 from pathlib import Path
-from typing import Any, Dict
 
+import yaml
 from kubernetes import client, config
+from kubernetes.client.models import CoreV1Event, V1PodList
 from kubernetes.dynamic import DynamicClient
 
 from .process import stream_command
 
 WAR_MANIFESTS = files("manifests")
 
 
-def get_static_client():
+def get_static_client() -> CoreV1Event:
     config.load_kube_config()
     return client.CoreV1Api()
 
 
-def get_dynamic_client():
+def get_dynamic_client() -> DynamicClient:
     config.load_kube_config()
     return DynamicClient(client.ApiClient())
 
 
-def get_pods():
+def get_pods() -> V1PodList:
     sclient = get_static_client()
     return sclient.list_namespaced_pod("warnet")
 
 
-def get_mission(mission):
+def get_mission(mission: str) -> list[V1PodList]:
     pods = get_pods()
     crew = []
     for pod in pods.items:
@@ -37,15 +37,15 @@ def get_mission(mission):
     return crew
 
 
-def get_edges():
+def get_edges() -> any:
     sclient = get_static_client()
     configmap = sclient.read_namespaced_config_map(name="edges", namespace="warnet")
     return json.loads(configmap.data["data"])
 
 
 def create_kubernetes_object(
-    kind: str, metadata: Dict[str, Any], spec: Dict[str, Any] = None
-) -> Dict[str, Any]:
+    kind: str, metadata: dict[str, any], spec: dict[str, any] = None
+) -> dict[str, any]:
     obj = {
         "apiVersion": "v1",
         "kind": kind,
@@ -60,7 +60,7 @@ def create_namespace() -> dict:
     return {"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "warnet"}}
 
 
-def set_kubectl_context(namespace: str):
+def set_kubectl_context(namespace: str) -> bool:
     """
     Set the default kubectl context to the specified namespace.
     """
@@ -73,7 +73,7 @@ def set_kubectl_context(namespace: str):
     return result
 
 
-def deploy_base_configurations():
+def deploy_base_configurations() -> bool:
     base_configs = [
         "namespace.yaml",
         "rbac-config.yaml",
@@ -87,12 +87,12 @@ def deploy_base_configurations():
     return True
 
 
-def apply_kubernetes_yaml(yaml_file: str):
+def apply_kubernetes_yaml(yaml_file: str) -> bool:
     command = f"kubectl apply -f {yaml_file}"
     return stream_command(command)
 
 
-def apply_kubernetes_yaml_obj(yaml_obj: str):
+def apply_kubernetes_yaml_obj(yaml_obj: str) -> None:
     with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as temp_file:
         yaml.dump(yaml_obj, temp_file)
         temp_file_path = temp_file.name
@@ -103,11 +103,11 @@ def apply_kubernetes_yaml_obj(yaml_obj: str):
         Path(temp_file_path).unlink()
 
 
-def delete_namespace(namespace: str):
+def delete_namespace(namespace: str) -> bool:
     command = f"kubectl delete namespace {namespace} --ignore-not-found"
     return stream_command(command)
 
 
-def delete_pod(pod_name: str):
+def delete_pod(pod_name: str) -> bool:
     command = f"kubectl delete pod {pod_name}"
     return stream_command(command)

src/warnet/cli/namespaces.py

@@ -37,7 +37,7 @@ def deploy(namespaces: str):
     for n in names:
         if not n.startswith("warnet-"):
             print(
-                f"Failled to create namespace: {n}. Namespaces must start with a 'warnet-' prefix."
+                f"Failed to create namespace: {n}. Namespaces must start with a 'warnet-' prefix."
             )
 
     # deploy namespaces

src/warnet/cli/network.py

@@ -34,7 +34,7 @@ def network():
 
 
 class Edge:
-    def __init__(self, src: str, dst: str, data: Dict[str, Any]):
+    def __init__(self, src: str, dst: str, data: dict[str, any]):
         self.src = src
         self.dst = dst
         self.data = data
@@ -43,7 +43,7 @@ def to_dict(self):
         return {"src": self.src, "dst": self.dst, "data": self.data}
 
 
-def edges_from_network_file(network_file: Dict[str, Any]) -> List[Edge]:
+def edges_from_network_file(network_file: dict[str, any]) -> list[Edge]:
     edges = []
     for node in network_file["nodes"]:
         if "connect" in node:
@@ -52,7 +52,7 @@ def edges_from_network_file(network_file: Dict[str, Any]) -> List[Edge]:
     return edges
 
 
-def create_edges_map(network_file: Dict[str, Any]):
+def create_edges_map(network_file: dict[str, any]):
     edges = []
     for edge in edges_from_network_file(network_file):
         edges.append(edge.to_dict())
@@ -149,7 +149,7 @@ def logs(follow: bool):
 
 @network.command()
 def connected():
-    """Determine if all p2p conenctions defined in graph are established"""
+    """Determine if all p2p connections defined in graph are established"""
     print(_connected())
 
 

src/warnet/cli/process.py

@@ -1,7 +1,7 @@
 import subprocess
 
 
-def run_command(command) -> str:
+def run_command(command: str) -> str:
     result = subprocess.run(
         command, shell=True, capture_output=True, text=True, executable="/bin/bash"
     )
@@ -10,7 +10,7 @@ def run_command(command) -> str:
     return result.stdout
 
 
-def stream_command(command, env=None) -> bool:
+def stream_command(command: str, env=None) -> bool:
     process = subprocess.Popen(
         ["/bin/bash", "-c", command],
         stdout=subprocess.PIPE,

src/warnet/cli/scenarios.py

@@ -11,6 +11,7 @@
 from rich import print
 from rich.console import Console
 from rich.table import Table
+
 from warnet import scenarios as SCENARIOS
 
 from .k8s import apply_kubernetes_yaml, create_namespace, get_mission
@@ -51,7 +52,7 @@ def _available():
 @scenarios.command(context_settings={"ignore_unknown_options": True})
 @click.argument("scenario", type=str)
 @click.argument("additional_args", nargs=-1, type=click.UNPROCESSED)
-def run(scenario, additional_args):
+def run(scenario: str, additional_args: tuple[str]):
     """
     Run <scenario> from the Warnet Test Framework with optional arguments
     """
@@ -69,7 +70,7 @@ def run(scenario, additional_args):
 @scenarios.command(context_settings={"ignore_unknown_options": True})
 @click.argument("scenario_path", type=str)
 @click.argument("additional_args", nargs=-1, type=click.UNPROCESSED)
-def run_file(scenario_path, additional_args):
+def run_file(scenario_path: str, additional_args: tuple[str]):
     """
     Run <scenario_path> from the Warnet Test Framework with optional arguments
     """
@@ -79,7 +80,7 @@ def run_file(scenario_path, additional_args):
     return run_scenario(scenario_path, additional_args)
 
 
-def run_scenario(scenario_path, additional_args):
+def run_scenario(scenario_path: str, additional_args: tuple[str]):
     if not os.path.exists(scenario_path):
         raise Exception(f"Scenario file not found at {scenario_path}.")
 
@@ -119,7 +120,7 @@ def run_scenario(scenario_path, additional_args):
                 "apiVersion": "v1",
                 "kind": "ConfigMap",
                 "metadata": {
-                    "name": "scnaeriopy",
+                    "name": "scenariopy",
                     "namespace": "warnet",
                 },
                 "data": {"scenario.py": scenario_text},
@@ -146,7 +147,7 @@ def run_scenario(scenario_path, additional_args):
                                     "subPath": "warnet.json",
                                 },
                                 {
-                                    "name": "scnaeriopy",
+                                    "name": "scenariopy",
                                     "mountPath": "scenario.py",
                                     "subPath": "scenario.py",
                                 },
@@ -155,7 +156,7 @@ def run_scenario(scenario_path, additional_args):
                     ],
                     "volumes": [
                         {"name": "warnetjson", "configMap": {"name": "warnetjson"}},
-                        {"name": "scnaeriopy", "configMap": {"name": "scnaeriopy"}},
+                        {"name": "scenariopy", "configMap": {"name": "scenariopy"}},
                     ],
                 },
             },
@@ -188,6 +189,6 @@ def active():
     console.print(table)
 
 
-def _active():
+def _active() -> list[str]:
     commanders = get_mission("commander")
     return [{"commander": c.metadata.name, "status": c.status.phase.lower()} for c in commanders]

test/data/scenario_p2p_interface.py

@@ -3,6 +3,7 @@
 
 # The base class exists inside the commander container
 from commander import Commander
+
 from test_framework.messages import CInv, msg_getdata
 from test_framework.p2p import P2PInterface
 

test/ln_test.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 
 from test_base import TestBase
+
 from warnet.services import ServiceType
 
 

test/scenarios_test.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 
 from test_base import TestBase
+
 from warnet.cli.k8s import delete_pod
 from warnet.cli.scenarios import _active as scenarios_active
 from warnet.cli.scenarios import _available as scenarios_available