lint

Author: pinheadmz

test/test_base.py

@@ -2,7 +2,6 @@
 import logging
 import logging.config
 import os
-import pexpect
 import re
 import sys
 import threading
@@ -11,6 +10,8 @@
 from tempfile import mkdtemp
 from time import sleep
 
+import pexpect
+
 from warnet import SRC_DIR
 from warnet.k8s import get_pod_exit_status
 from warnet.network import _connected as network_connected
@@ -47,7 +48,7 @@ def cleanup(self, signum=None, frame=None):
         try:
             self.log.info("Stopping network")
             if self.network:
-                session = pexpect.spawn(f"warnet down", encoding="utf-8")
+                session = pexpect.spawn("warnet down", encoding="utf-8")
                 session.logfile = sys.stdout
                 session.expect("Do you want to bring down the running Warnet?", timeout=30)
                 session.sendline("y")

test/wargames_test.py

@@ -64,7 +64,9 @@ def check_scenario_permissions(self):
 
         self.log.info("Switch to wargames player context")
         self.log.info(self.warnet(f"admin create-kubeconfigs --kubeconfig-dir={self.kconfigdir}"))
-        clicker = pexpect.spawn(f"warnet auth {self.kconfigdir}/warnet-user-wargames-red-kubeconfig")
+        clicker = pexpect.spawn(
+            f"warnet auth {self.kconfigdir}/warnet-user-wargames-red-kubeconfig"
+        )
         while clicker.expect(["Overwrite", "Updated kubeconfig"]) == 0:
             print(clicker.before, clicker.after)
             clicker.sendline("y")
@@ -85,11 +87,13 @@ def check_scenario_permissions(self):
         assert self.warnet("bitcoin rpc armada getblockcount") == "6"
 
         self.log.info("Check pod limit per namespace")
-        for i in range(10):
+        for _ in range(10):
             stream_command(
                 f"warnet run {self.scen_test_dir / 'nothing.py'} --source_dir={self.scen_src_dir}"
             )
-        assert len(scenarios_deployed()) == 2, f"Unexpected scenarios deployed:{scenarios_deployed()}"
+        assert len(scenarios_deployed()) == 2, (
+            f"Unexpected scenarios deployed:{scenarios_deployed()}"
+        )
 
         self.log.info("Restore admin context")
         stream_command(f"kubectl config use-context {self.initial_context}")