test: Run negative ping tests in parallel

Rather than having to wait for each timeout sequentially, run each
test in parallel.
This commit is contained in:
Tobias Waldekranz
2024-05-17 23:14:42 +02:00
parent 4cf54d422e
commit 9e7e30ccb0
5 changed files with 37 additions and 9 deletions
+1 -1
View File
@@ -6,7 +6,7 @@ from .furl import Furl
from .netns import IsolatedMacVlan
from .sniffer import Sniffer
from .tap import Test
from .util import until
from .util import parallel, until
def std_topology(name):
return os.path.realpath(
+28
View File
@@ -1,7 +1,35 @@
import time
import threading
import infamy.neigh
import infamy.netconf as netconf
class ParallelFn(threading.Thread):
def __init__(self, group=None, target=None, name=None,
args=(), kwargs={}, Verbose=None):
threading.Thread.__init__(self, group, target, name, args, kwargs)
self._exc, self._return = None, None
def run(self):
if self._target is not None:
try:
self._return = self._target(*self._args,
**self._kwargs)
except Exception as e:
self._exc = e
def join(self, *args):
threading.Thread.join(self, *args)
if self._exc:
raise self._exc
return self._return
def parallel(*fns):
ths = [ParallelFn(target=fn) for fn in fns]
[th.start() for th in ths]
return [th.join() for th in ths]
def until(fn, attempts=10, interval=1):
for attempt in range(attempts):
if fn():