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

Author: pinheadmz

src/warnet/status.py

@@ -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")
 

test/data/scenario_buggy_failure.py

@@ -0,0 +1,23 @@
+#!/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

@@ -21,6 +21,7 @@ def run_test(self):
             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()
 
@@ -76,6 +77,8 @@ def run_and_check_miner_scenario_from_file(self):
         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 run_and_check_scenario_from_file(self):
@@ -90,6 +93,22 @@ def check_regtest_recon(self):
         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
+            if len([s for s in deployed if s["status"] == "failed"]) != 1:
+                return False
+            return True
+        self.wait_for_predicate(two_pass_one_fail)
+        table = self.warnet("status")
+        assert "Active Scenarios: 0" in table
+
 
 if __name__ == "__main__":
     test = ScenariosTest()