from aiounittest import AsyncTestCase from aiohttp.test_utils import unused_port from click.testing import CliRunner from catfish.__main__ import cli from catfish import worker from catfish.utils.processes import terminate_processes, CURRENT_PROCESS import functools from multiprocessing import Process class BaseTestCase(AsyncTestCase): def setUp(self): worker.stop_worker() self.cli_runner = CliRunner() self.cli = cli self.run_cli = functools.partial(self.cli_runner.invoke, self.cli) def tearDown(self): worker.stop_worker() terminate_processes(CURRENT_PROCESS.children(recursive=True)) class BaseWorkerTestCase(BaseTestCase): def setUp(self): super().setUp() self.unused_port = unused_port() self.router_process = Process( target=worker.run, args=(self.unused_port,), daemon=True ) self.router_process.start() def tearDown(self): self.router_process.terminate() super().tearDown()