status: rename scenarios_active to more accurate scenarios_deployed

Author: pinheadmz

src/warnet/control.py

@@ -27,18 +27,11 @@
 
 console = Console()
 
-
-def get_active_scenarios():
-    """Get list of active scenarios"""
-    commanders = get_mission("commander")
-    return [c.metadata.name for c in commanders]
-
-
 @click.command()
 @click.argument("scenario_name", required=False)
 def stop(scenario_name):
     """Stop a running scenario or all scenarios"""
-    active_scenarios = get_active_scenarios()
+    active_scenarios = [sc.metadata.name for sc in get_mission("commander")]
 
     if not active_scenarios:
         console.print("[bold red]No active scenarios found.[/bold red]")
@@ -108,24 +101,6 @@ def stop_all_scenarios(scenarios):
     console.print("[bold green]All scenarios have been stopped.[/bold green]")
 
 
-def list_active_scenarios():
-    """List all active scenarios"""
-    active_scenarios = get_active_scenarios()
-    if not active_scenarios:
-        print("No active scenarios found.")
-        return
-
-    console = Console()
-    table = Table(title="Active Scenarios", show_header=True, header_style="bold magenta")
-    table.add_column("Name", style="cyan")
-    table.add_column("Status", style="green")
-
-    for scenario in active_scenarios:
-        table.add_row(scenario, "deployed")
-
-    console.print(table)
-
-
 @click.command()
 def down():
     """Bring down a running warnet quickly"""

src/warnet/status.py

@@ -14,7 +14,7 @@ def status():
     console = Console()
 
     tanks = _get_tank_status()
-    scenarios = _get_active_scenarios()
+    scenarios = _get_deployed_scenarios()
 
     # Create a unified table
     table = Table(title="Warnet Status", show_header=True, header_style="bold magenta")
@@ -62,6 +62,6 @@ def _get_tank_status():
     return [{"name": tank.metadata.name, "status": tank.status.phase.lower()} for tank in tanks]
 
 
-def _get_active_scenarios():
+def _get_deployed_scenarios():
     commanders = get_mission("commander")
     return [{"name": c.metadata.name, "status": c.status.phase.lower()} for c in commanders]

test/scenarios_test.py

@@ -7,7 +7,7 @@
 
 from warnet.control import stop_scenario
 from warnet.process import run_command
-from warnet.status import _get_active_scenarios as scenarios_active
+from warnet.status import _get_deployed_scenarios as scenarios_deployed
 
 
 class ScenariosTest(TestBase):
@@ -32,22 +32,22 @@ def setup_network(self):
 
     def scenario_running(self, scenario_name: str):
         """Check that we are only running a single scenario of the correct name"""
-        active = scenarios_active()
-        assert len(active) == 1
-        return scenario_name in active[0]["name"]
+        deployed = scenarios_deployed()
+        assert len(deployed) == 1
+        return scenario_name in deployed[0]["name"]
 
     def check_scenario_stopped(self):
-        running = scenarios_active()
+        running = scenarios_deployed()
         self.log.debug(f"Checking if scenario stopped. Running scenarios: {len(running)}")
         return len(running) == 0
 
     def check_scenario_clean_exit(self):
-        active = scenarios_active()
-        return all(scenario["status"] == "succeeded" for scenario in active)
+        deployed = scenarios_deployed()
+        return all(scenario["status"] == "succeeded" for scenario in deployed)
 
     def stop_scenario(self):
         self.log.info("Stopping running scenario")
-        running = scenarios_active()
+        running = scenarios_deployed()
         assert len(running) == 1, f"Expected one running scenario, got {len(running)}"
         assert running[0]["status"] == "running", "Scenario should be running"
         stop_scenario(running[0]["name"])
@@ -58,8 +58,8 @@ def check_blocks(self, target_blocks, start: int = 0):
         self.log.debug(f"Current block count: {count}, target: {start + target_blocks}")
 
         try:
-            active = scenarios_active()
-            commander = active[0]["commander"]
+            deployed = scenarios_deployed()
+            commander = deployed[0]["commander"]
             command = f"kubectl logs {commander}"
             print("\ncommander output:")
             print(run_command(command))

test/signet_test.py

@@ -6,7 +6,7 @@
 
 from test_base import TestBase
 
-from warnet.status import _get_active_scenarios as scenarios_active
+from warnet.status import _get_deployed_scenarios as scenarios_deployed
 
 
 class SignetTest(TestBase):
@@ -55,8 +55,8 @@ def check_signet_recon(self):
         self.warnet(f"run {scenario_file}")
 
         def check_scenario_clean_exit():
-            active = scenarios_active()
-            return all(scenario["status"] == "succeeded" for scenario in active)
+            deployed = scenarios_deployed()
+            return all(scenario["status"] == "succeeded" for scenario in deployed)
 
         self.wait_for_predicate(check_scenario_clean_exit)
 

test/test_base.py

@@ -10,10 +10,10 @@
 from time import sleep
 
 from warnet import SRC_DIR
-from warnet.control import get_active_scenarios
 from warnet.k8s import get_pod_exit_status
 from warnet.network import _connected as network_connected
 from warnet.status import _get_tank_status as network_status
+from warnet.status import _get_deployed_scenarios as scenarios_deployed
 
 
 class TestBase:
@@ -126,7 +126,7 @@ def wait_for_all_edges(self, timeout=20 * 60, interval=5):
 
     def wait_for_all_scenarios(self):
         def check_scenarios():
-            scns = get_active_scenarios()
+            scns = scenarios_deployed()
             if len(scns) == 0:
                 return True
             for s in scns: