catfish/tests/__init__.py

import functools
from multiprocessing import Process

from aiohttp.test_utils import unused_port
from aiounittest import AsyncTestCase
from click.testing import CliRunner

from catfish import worker
from catfish.__main__ import cli
from catfish.utils.processes import CURRENT_PROCESS, terminate_processes


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()