lint

src/warnet/control.py

@@ -27,6 +27,7 @@
 
 console = Console()
 
+
 @click.command()
 @click.argument("scenario_name", required=False)
 def stop(scenario_name):

test/data/scenario_buggy_failure.py

@@ -7,6 +7,7 @@
 except Exception:
     from resources.scenarios.commander import Commander
 
+
 class Failure(Commander):
     def set_test_params(self):
         self.num_nodes = 1

test/scenarios_test.py

@@ -102,9 +102,8 @@ 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
+            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

test/test_base.py

@@ -12,8 +12,8 @@
 from warnet import SRC_DIR
 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
+from warnet.status import _get_tank_status as network_status
 
 
 class TestBase:
@@ -130,8 +130,8 @@ def check_scenarios():
             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