DRY out the namespace check

Replace the namespace check with this function: `get_default_namespace_or(namespace)`

Author: mplsgrant

src/warnet/bitcoin.py

@@ -11,7 +11,7 @@
 from urllib3.exceptions import MaxRetryError
 
 from .constants import BITCOINCORE_CONTAINER
-from .k8s import get_default_namespace, get_mission, pod_log
+from .k8s import get_default_namespace, get_mission, pod_log, get_default_namespace_or
 from .process import run_command
 
 
@@ -40,8 +40,7 @@ def rpc(tank: str, method: str, params: str, namespace: Optional[str]):
 def _rpc(tank: str, method: str, params: str, namespace: Optional[str] = None):
     # bitcoin-cli should be able to read bitcoin.conf inside the container
     # so no extra args like port, chain, username or password are needed
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     if params:
         cmd = f"kubectl -n {namespace} exec {tank} --container {BITCOINCORE_CONTAINER} -- bitcoin-cli {method} {' '.join(map(str, params))}"
     else:
@@ -56,8 +55,7 @@ def debug_log(tank: str, namespace: Optional[str]):
     """
     Fetch the Bitcoin Core debug log from <tank pod name>
     """
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     cmd = f"kubectl logs {tank} --namespace {namespace}"
     try:
         print(run_command(cmd))
@@ -141,10 +139,8 @@ def messages(
     Fetch messages sent between <tank_a pod name> and <tank_b pod name> in [chain]
     """
     try:
-        if not namespace_a:
-            namespace_a = get_default_namespace()
-        if not namespace_b:
-            namespace_b = get_default_namespace()
+        namespace_a = get_default_namespace_or(namespace_a)
+        namespace_b = get_default_namespace_or(namespace_b)
 
         # Get the messages
         messages = get_messages(

src/warnet/control.py

@@ -35,7 +35,7 @@
     snapshot_bitcoin_datadir,
     wait_for_init,
     wait_for_pod,
-    write_file_to_container,
+    write_file_to_container, get_default_namespace_or,
 )
 from .process import run_command, stream_command
 
@@ -236,8 +236,7 @@ def run(
     Run a scenario from a file.
     Pass `-- --help` to get individual scenario help
     """
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
 
     scenario_path = Path(scenario_file).resolve()
     scenario_dir = scenario_path.parent if not source_dir else Path(source_dir).resolve()
@@ -361,8 +360,7 @@ def logs(pod_name: str, follow: bool, namespace: str):
 
 
 def _logs(pod_name: str, follow: bool, namespace: Optional[str] = None):
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
 
     if pod_name == "":
         try:

src/warnet/deploy.py

@@ -26,7 +26,7 @@
     get_default_namespace,
     get_namespaces_by_prefix,
     wait_for_ingress_controller,
-    wait_for_pod_ready,
+    wait_for_pod_ready, get_default_namespace_or,
 )
 from .process import stream_command
 
@@ -218,12 +218,11 @@ def deploy_network(directory: Path, debug: bool = False, namespace: Optional[str
     network_file_path = directory / NETWORK_FILE
     defaults_file_path = directory / DEFAULTS_FILE
 
+    namespace = get_default_namespace_or(namespace)
+
     with network_file_path.open() as f:
         network_file = yaml.safe_load(f)
 
-    if not namespace:
-        namespace = get_default_namespace()
-
     for node in network_file["nodes"]:
         click.echo(f"Deploying node: {node.get('name')}")
         try:

src/warnet/k8s.py

@@ -51,9 +51,8 @@ def get_pods() -> list[V1Pod]:
 
 
 def get_pod(name: str, namespace: Optional[str] = None) -> V1Pod:
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
-    if not namespace:
-        namespace = get_default_namespace()
     return sclient.read_namespaced_pod(name=name, namespace=namespace)
 
 
@@ -67,8 +66,7 @@ def get_mission(mission: str) -> list[V1Pod]:
 
 
 def get_pod_exit_status(pod_name, namespace: Optional[str] = None):
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     try:
         sclient = get_static_client()
         pod = sclient.read_namespaced_pod(name=pod_name, namespace=namespace)
@@ -82,8 +80,7 @@ def get_pod_exit_status(pod_name, namespace: Optional[str] = None):
 
 
 def get_edges(namespace: Optional[str] = None) -> any:
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
     configmap = sclient.read_namespaced_config_map(name="edges", namespace=namespace)
     return json.loads(configmap.data["data"])
@@ -138,8 +135,7 @@ def delete_namespace(namespace: str) -> bool:
 
 
 def delete_pod(pod_name: str, namespace: Optional[str] = None) -> bool:
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     command = f"kubectl -n {namespace} delete pod {pod_name}"
     return stream_command(command)
 
@@ -159,15 +155,18 @@ def get_default_namespace() -> str:
     return kubectl_namespace if kubectl_namespace else DEFAULT_NAMESPACE
 
 
+def get_default_namespace_or(namespace: Optional[str]) -> str:
+    return namespace if namespace else get_default_namespace()
+
+
 def snapshot_bitcoin_datadir(
     pod_name: str,
     chain: str,
     local_path: str = "./",
     filters: list[str] = None,
     namespace: Optional[str] = None,
 ) -> None:
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
 
     try:
@@ -292,8 +291,7 @@ def wait_for_pod_ready(name, namespace, timeout=300):
 
 
 def wait_for_init(pod_name, timeout=300, namespace: Optional[str] = None):
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
     w = watch.Watch()
     for event in w.stream(
@@ -335,9 +333,9 @@ def get_ingress_ip_or_host():
 
 
 def pod_log(pod_name, container_name=None, follow=False, namespace: Optional[str] = None):
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
-    if not namespace:
-        namespace = get_default_namespace()
+
     try:
         return sclient.read_namespaced_pod_log(
             name=pod_name,
@@ -351,8 +349,7 @@ def pod_log(pod_name, container_name=None, follow=False, namespace: Optional[str
 
 
 def wait_for_pod(pod_name, timeout_seconds=10, namespace: Optional[str] = None):
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
     while timeout_seconds > 0:
         pod = sclient.read_namespaced_pod_status(name=pod_name, namespace=namespace)
@@ -365,8 +362,7 @@ def wait_for_pod(pod_name, timeout_seconds=10, namespace: Optional[str] = None):
 def write_file_to_container(
     pod_name, container_name, dst_path, data, namespace: Optional[str] = None
 ):
-    if not namespace:
-        namespace = get_default_namespace()
+    namespace = get_default_namespace_or(namespace)
     sclient = get_static_client()
     exec_command = ["sh", "-c", f"cat > {dst_path}"]
     try:
@@ -428,3 +424,4 @@ def get_service_accounts_in_namespace(namespace):
     # skip the default service account created by k8s
     service_accounts = run_command(command).split()
     return [sa for sa in service_accounts if sa != "default"]
+