Fixes formatting errors

Author: dimeko

deltascan/cli/cmd.py

@@ -16,9 +16,10 @@
 from rich.columns import Columns
 import threading
 import signal
-import select 
+import select
 import sys
 
+
 def interactive_shell(_app, _ui, _is_interactive):
     """
     Starts an interactive shell for the application.
@@ -33,15 +34,15 @@ def interactive_shell(_app, _ui, _is_interactive):
     shell = Shell(_app)
     __only_first_time = True
 
-    while _app.cleaning_up == False:
-        if _app.is_running == False and _is_interactive == True and __only_first_time == True:
+    while _app.cleaning_up is False:
+        if _app.is_running is False and _is_interactive is True and __only_first_time is True:
             __only_first_time = False
             pass
         else:
             # Setting inpout with timeout in order not to block in case of cancel
             # and _app.cleaning_up is re-checked
             a, b, c = select.select([sys.stdin], [], [], 2)
-            if a ==[] and b == [] and c == []:
+            if a == [] and b == [] and c == []:
                 continue
 
         _ui["ui_live"].stop()
@@ -135,7 +136,7 @@ def do_scan(self, v):
             return
         v1, v2 = v.split(" ")
         _r = self._app.add_scan(v1, v2)
-        if self._app.is_running == False:
+        if self._app.is_running is False:
             _dscan_thread = threading.Thread(target=self._app.scan)
             _dscan_thread.start()
         if _r is False:
@@ -220,10 +221,12 @@ def do_exit(self, _):
         Exit Deltascan"""
         os._exit(0)
 
+
 def signal_handler(signal, frame):
     print("Exiting without cleanup :-(")
     os._exit(1)
 
+
 def run():
     """
     Entry point for the command line interface.
@@ -362,8 +365,8 @@ def run():
             _dscan_thread.start()
             _shell_thread.start()
             _dscan_thread.join()
-        
-            if _dscan.is_interactive or clargs.interactive == True:
+
+            if _dscan.is_interactive or clargs.interactive is True:
                 _shell_thread.join()
             else:
                 print("No scans left in the queue... Exiting.")
@@ -389,12 +392,12 @@ def run():
             output = CliOutput(_r)
             output.display()
         else:
-            if clargs.interactive == True:
+            if clargs.interactive is True:
                 print("No action provided. Starting interactive shell.")
             else:
                 print("Invalid action. Exiting...")
 
-        if clargs.interactive == True:
+        if clargs.interactive is True:
             _shell_thread = threading.Thread(
                 target=interactive_shell, args=(_dscan, ui_context, clargs.interactive,))
             _shell_thread.start()
@@ -406,8 +409,9 @@ def run():
         signal.signal(signal.SIGINT, signal_handler)
         _dscan.cleanup()
         print("Cancelling running scans and closing ...")
-    except Exception:
+    except Exception as e:
         print(f"Unknown error occurred: {str(e)}")
 
+
 if __name__ == "__main__":
     run()

deltascan/core/deltascan.py

@@ -174,15 +174,15 @@ def scan(self):
         Returns:
             None
         """
-        if self._is_running == True:
+        if self._is_running is True:
             return
         self._T = Thread(target=self._scan_orchestrator)
         self._T.start()
         self._T.join()
 
     def _scan_orchestrator(self):
         self._scans_to_wait = {}
-        
+
         self._is_running = True
         while True:
             self._remove_finished_scan_from_list()
@@ -206,24 +206,24 @@ def _scan_orchestrator(self):
             time.sleep(1)
 
     def _remove_finished_scan_from_list(self):
-            """
-            Removes the finished scans from the list of scans to wait for completion.
+        """
+        Removes the finished scans from the list of scans to wait for completion.
 
-            This method iterates over the `_scans_to_wait` dictionary and checks if each thread is alive.
-            If a thread is not alive, it is removed from the dictionary.
+        This method iterates over the `_scans_to_wait` dictionary and checks if each thread is alive.
+        If a thread is not alive, it is removed from the dictionary.
 
-            Args:
-                None
+        Args:
+            None
 
-            Returns:
-                None
-            """
-            threads_to_remove = []
-            for _n, _th in self._scans_to_wait.items():
-                if _th["_thr"].is_alive() is False:
-                    threads_to_remove.append(_n)
-            for _n in threads_to_remove:
-                del self._scans_to_wait[_n]
+        Returns:
+            None
+        """
+        threads_to_remove = []
+        for _n, _th in self._scans_to_wait.items():
+            if _th["_thr"].is_alive() is False:
+                threads_to_remove.append(_n)
+        for _n in threads_to_remove:
+            del self._scans_to_wait[_n]
 
     def _get_profile(self, _profile):
         try:
@@ -834,7 +834,7 @@ def is_running(self):
     @property
     def scans_to_wait(self):
         return len(self._scans_to_wait.keys())
-    
+
     @property
     def scans_to_execute(self):
         return len(self._scan_list)