Minor cleanup and add utility function for Automotive Scanner

Author: polybassa

scapy/contrib/automotive/scanner/executor.py

@@ -33,8 +33,11 @@
     Callable,
     Type,
     cast,
+    TypeVar,
 )
 
+T = TypeVar("T")
+
 
 class AutomotiveTestCaseExecutor(metaclass=abc.ABCMeta):
     """
@@ -415,6 +418,10 @@ def show_testcases_status(self):
                 data += [(repr(s), t.__class__.__name__, t.has_completed(s))]
         make_lined_table(data, lambda *tup: (tup[0], tup[1], tup[2]))
 
+    def get_test_cases_by_class(self, cls):
+        # type: (Type[T]) -> List[T]
+        return [x for x in self.configuration.test_cases if isinstance(x, cls)]
+
     @property
     def supported_responses(self):
         # type: () -> List[EcuResponse]

scapy/contrib/isotp/isotp_soft_socket.py

@@ -228,6 +228,8 @@ class TimeoutScheduler:
     # use heapq functions on _handles!
     _handles = []  # type: List[TimeoutScheduler.Handle]
 
+    logger = logging.getLogger("scapy.contrib.automotive.timeout_scheduler")
+
     @classmethod
     def schedule(cls, timeout, callback):
         # type: (float, Callable[[], None]) -> TimeoutScheduler.Handle
@@ -316,13 +318,13 @@ def _wait(cls, handle):
         # Wait until the next timeout,
         # or until event.set() gets called in another thread.
         if to_wait > 0:
-            log_isotp.debug("TimeoutScheduler Thread going to sleep @ %f " +
-                            "for %fs", now, to_wait)
+            cls.logger.debug("Thread going to sleep @ %f " +
+                             "for %fs", now, to_wait)
             interrupted = cls._event.wait(to_wait)
             new = cls._time()
-            log_isotp.debug("TimeoutScheduler Thread awake @ %f, slept for" +
-                            " %f, interrupted=%d", new, new - now,
-                            interrupted)
+            cls.logger.debug("Thread awake @ %f, slept for" +
+                             " %f, interrupted=%d", new, new - now,
+                             interrupted)
 
         # Clear the event so that we can wait on it again,
         # Must be done before doing the callbacks to avoid losing a set().
@@ -335,7 +337,7 @@ def _task(cls):
         start when the first timeout is added and stop when the last timeout
         is removed or executed."""
 
-        log_isotp.debug("TimeoutScheduler Thread spawning @ %f", cls._time())
+        cls.logger.debug("Thread spawning @ %f", cls._time())
 
         time_empty = None
 
@@ -357,7 +359,7 @@ def _task(cls):
         finally:
             # Worst case scenario: if this thread dies, the next scheduled
             # timeout will start a new one
-            log_isotp.debug("TimeoutScheduler Thread died @ %f", cls._time())
+            cls.logger.debug("Thread died @ %f", cls._time())
             cls._thread = None
 
     @classmethod
@@ -379,7 +381,7 @@ def _poll(cls):
                     callback = handle._cb
                     handle._cb = True
 
-            # Call the callback here, outside of the mutex
+            # Call the callback here, outside the mutex
             if callable(callback):
                 try:
                     callback()