status: only count "running"/"pending" scenarios as "active"

src/warnet/control.py

@@ -28,17 +28,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 +102,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")
@@ -31,9 +31,12 @@ def status():
         table.add_row("", "", "")
 
     # Add scenarios to the table
+    active = 0
     if scenarios:
         for scenario in scenarios:
             table.add_row("Scenario", scenario["name"], scenario["status"])
+            if scenario["status"] == "running" or scenario["status"] == "pending":
+                active += 1
     else:
         table.add_row("Scenario", "No active scenarios", "")
 
@@ -52,7 +55,7 @@ def status():
     # Print summary
     summary = Text()
     summary.append(f"\nTotal Tanks: {len(tanks)}", style="bold cyan")
-    summary.append(f" | Active Scenarios: {len(scenarios)}", style="bold green")
+    summary.append(f" | Active Scenarios: {active}", style="bold green")
     console.print(summary)
     _connected(end="\r")
 
@@ -62,6 +65,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/data/scenario_buggy_failure.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+
+# The base class exists inside the commander container
+try:
+    from commander import Commander
+except Exception:
+    from resources.scenarios.commander import Commander
+
+
+class Failure(Commander):
+    def set_test_params(self):
+        self.num_nodes = 1
+
+    def add_options(self, parser):
+        parser.description = "This test will fail and exit with code 222"
+        parser.usage = "warnet run /path/to/scenario_buggy_failure.py"
+
+    def run_test(self):
+        raise Exception("Failed execution!")
+
+
+if __name__ == "__main__":
+    Failure().main()

test/data/scenario_connect_dag.py

@@ -7,10 +7,6 @@
 from commander import Commander
 
 
-def cli_help():
-    return "Connect a complete DAG from a set of unconnected nodes"
-
-
 @unique
 class ConnectionType(Enum):
     IP = auto()
@@ -22,6 +18,10 @@ def set_test_params(self):
         # This is just a minimum
         self.num_nodes = 10
 
+    def add_options(self, parser):
+        parser.description = "Connect a complete DAG from a set of unconnected nodes"
+        parser.usage = "warnet run /path/to/scenario_connect_dag.py"
+
     def run_test(self):
         # All permutations of a directed acyclic graph with zero, one, or two inputs/outputs
         #

test/data/scenario_p2p_interface.py

@@ -12,10 +12,6 @@
 from test_framework.p2p import P2PInterface
 
 
-def cli_help():
-    return "Run P2P GETDATA test"
-
-
 class P2PStoreBlock(P2PInterface):
     def __init__(self):
         super().__init__()
@@ -30,6 +26,10 @@ class GetdataTest(Commander):
     def set_test_params(self):
         self.num_nodes = 1
 
+    def add_options(self, parser):
+        parser.description = "Run P2P GETDATA test"
+        parser.usage = "warnet run /path/to/scenario_p2p_interface.py"
+
     def run_test(self):
         self.log.info("Adding the p2p connection")
 

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):
@@ -18,7 +18,10 @@ def __init__(self):
     def run_test(self):
         try:
             self.setup_network()
-            self.test_scenarios()
+            self.run_and_check_miner_scenario_from_file()
+            self.run_and_check_scenario_from_file()
+            self.check_regtest_recon()
+            self.check_active_count()
         finally:
             self.cleanup()
 
@@ -28,49 +31,36 @@ def setup_network(self):
         self.wait_for_all_tanks_status(target="running")
         self.wait_for_all_edges()
 
-    def test_scenarios(self):
-        self.run_and_check_miner_scenario_from_file()
-        self.run_and_check_scenario_from_file()
-        self.check_regtest_recon()
-
     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"]
-
-    def run_and_check_scenario_from_file(self):
-        scenario_file = "test/data/scenario_p2p_interface.py"
-        self.log.info(f"Running scenario from: {scenario_file}")
-        self.warnet(f"run {scenario_file}")
-        self.wait_for_predicate(self.check_scenario_clean_exit)
-
-    def run_and_check_miner_scenario_from_file(self):
-        scenario_file = "resources/scenarios/miner_std.py"
-        self.log.info(f"Running scenario from file: {scenario_file}")
-        self.warnet(f"run {scenario_file} --allnodes --interval=1")
-        start = int(self.warnet("bitcoin rpc tank-0000 getblockcount"))
-        self.wait_for_predicate(lambda: self.scenario_running("commander-minerstd"))
-        self.wait_for_predicate(lambda: self.check_blocks(2, start=start))
-        self.stop_scenario()
+        deployed = scenarios_deployed()
+        assert len(deployed) == 1
+        return scenario_name in deployed[0]["name"]
 
-    def check_regtest_recon(self):
-        scenario_file = "resources/scenarios/reconnaissance.py"
-        self.log.info(f"Running scenario from file: {scenario_file}")
-        self.warnet(f"run {scenario_file}")
-        self.wait_for_predicate(self.check_scenario_clean_exit)
+    def check_scenario_stopped(self):
+        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_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"])
+        self.wait_for_predicate(self.check_scenario_stopped)
 
     def check_blocks(self, target_blocks, start: int = 0):
         count = int(self.warnet("bitcoin rpc tank-0000 getblockcount"))
         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))
@@ -80,18 +70,43 @@ def check_blocks(self, target_blocks, start: int = 0):
 
         return count >= start + target_blocks
 
-    def stop_scenario(self):
-        self.log.info("Stopping running scenario")
-        running = scenarios_active()
-        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"])
-        self.wait_for_predicate(self.check_scenario_stopped)
+    def run_and_check_miner_scenario_from_file(self):
+        scenario_file = "resources/scenarios/miner_std.py"
+        self.log.info(f"Running scenario from file: {scenario_file}")
+        self.warnet(f"run {scenario_file} --allnodes --interval=1")
+        start = int(self.warnet("bitcoin rpc tank-0000 getblockcount"))
+        self.wait_for_predicate(lambda: self.scenario_running("commander-minerstd"))
+        self.wait_for_predicate(lambda: self.check_blocks(2, start=start))
+        table = self.warnet("status")
+        assert "Active Scenarios: 1" in table
+        self.stop_scenario()
 
-    def check_scenario_stopped(self):
-        running = scenarios_active()
-        self.log.debug(f"Checking if scenario stopped. Running scenarios: {len(running)}")
-        return len(running) == 0
+    def run_and_check_scenario_from_file(self):
+        scenario_file = "test/data/scenario_p2p_interface.py"
+        self.log.info(f"Running scenario from: {scenario_file}")
+        self.warnet(f"run {scenario_file}")
+        self.wait_for_predicate(self.check_scenario_clean_exit)
+
+    def check_regtest_recon(self):
+        scenario_file = "resources/scenarios/reconnaissance.py"
+        self.log.info(f"Running scenario from file: {scenario_file}")
+        self.warnet(f"run {scenario_file}")
+        self.wait_for_predicate(self.check_scenario_clean_exit)
+
+    def check_active_count(self):
+        scenario_file = "test/data/scenario_buggy_failure.py"
+        self.log.info(f"Running scenario from: {scenario_file}")
+        self.warnet(f"run {scenario_file}")
+
+        def two_pass_one_fail():
+            deployed = scenarios_deployed()
+            if len([s for s in deployed if s["status"] == "succeeded"]) != 2:
+                return False
+            return len([s for s in deployed if s["status"] == "failed"]) == 1
+
+        self.wait_for_predicate(two_pass_one_fail)
+        table = self.warnet("status")
+        assert "Active Scenarios: 0" in table
 
 
 if __name__ == "__main__":

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,9 +10,9 @@
 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_deployed_scenarios as scenarios_deployed
 from warnet.status import _get_tank_status as network_status
 
 
@@ -126,12 +126,12 @@ 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:
-                exit_status = get_pod_exit_status(s)
-                self.log.debug(f"Scenario {s} exited with code {exit_status}")
+                exit_status = get_pod_exit_status(s["name"])
+                self.log.debug(f"Scenario {s['name']} exited with code {exit_status}")
                 if exit_status != 0:
                     return False
             return True