Class ConcurrentRunner

test/helpers.py:135–161

class ConcurrentRunner(PARENT):
    def __init__(self, **kwargs):
        if _IS_SYNC:
            super().__init__(**kwargs)
        self.name = kwargs.get("name", "ConcurrentRunner")
        self.stopped = False
        self.task = None
        self.target = kwargs.get("target", None)
        self.args = kwargs.get("args", [])

    if not _IS_SYNC:

        def start(self):
            self.task = create_task(self.run(), name=self.name)

        def join(self, timeout: Optional[float] = None):  # type: ignore[override]
            if self.task is not None:
                asyncio.wait([self.task], timeout=timeout)

        def is_alive(self):
            return not self.stopped

    def run(self):
        try:
            self.target(*self.args)
        finally:
            self.stopped = True


class ExceptionCatchingTask(ConcurrentRunner):