Convert unittest.TestCase tests to pytest

[AetherUnbound]

Description

Most of our tests are set up in the functional format that pytest expects. We have two exceptions:

• catalog/tests/dags/common/sensors/test_single_run_external_dags_sensor.py
• ingestion_server/test/integration_test.py

Both of these tests use unittest.TestCase instead of individual test functions, and both should be able to be converted to pytest-formatted tests. TestCase-based tests don't allow the use of fixtures, which hinders the simplicity of some tests (see #3361 (comment)).

Examples of other pytest-formatted tests can be found throughout the catalog/tests directory.

Additional context

This came up as part of #3361, and the workarounds created in that PR should be removed when the tests are converted.

[ngken0995]

@AetherUnbound Hello, Can this be assign to me?

[ngken0995]

@AetherUnbound There are some errors from running ingestion_server. I'm not sure if I'm setting up the right credentials to run test on ingestion-server. The initial runjust ingestion_server/test-local is giving me raise Empty from

self = <test.integration_test.TestIngestion testMethod=test_filtered_image_index_creation>

    @pytest.mark.order(after="test_promote_images")
    def test_filtered_image_index_creation(self):
>       self._create_and_populate_filtered_index("image", "integration", "integration")

test/integration_test.py:517: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
test/integration_test.py:369: in _create_and_populate_filtered_index
    assert self.__class__.cb_queue.get(timeout=60) == "CALLBACK!"
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <multiprocessing.queues.Queue object at 0x7f25c9635990>, block = True, timeout = 59.99999627300008

    def get(self, block=True, timeout=None):
        if self._closed:
            raise ValueError(f"Queue {self!r} is closed")
        if block and timeout is None:
            with self._rlock:
                res = self._recv_bytes()
            self._sem.release()
        else:
            if block:
                deadline = time.monotonic() + timeout
            if not self._rlock.acquire(block, timeout):
                raise Empty
            try:
                if block:
                    timeout = deadline - time.monotonic()
                    if not self._poll(timeout):
>                       raise Empty
E                       _queue.Empty

/usr/lib/python3.11/multiprocessing/queues.py:114: Empty

I cancel the run and re-run just ingestion_server/test-local and get 23 errors stating psycopg2.errors.DuplicateTable: relation "api_deletedaudio" already exists Below is one of the full detail of error:

cls = <class 'test.integration_test.TestIngestion'>

    @classmethod
    def setUpClass(cls) -> None:
        # Launch a Bottle server to receive and handle callbacks
        cb_queue = Queue()
        cls.cb_queue = cb_queue
        cb_process = Process(target=start_bottle, args=(cb_queue,))
        cls.cb_process = cb_process
        cb_process.start()
    
        # Orchestrate containers with Docker Compose
        compose_path = gen_integration_compose()
        cls.compose_path = compose_path
    
        cls._compose_cmd(["up", "-d"])
    
        # Wait for services to be ready
        cls.upstream_db, cls.downstream_db = cls._wait_for_dbs()
        cls._wait_for_es()
        cls._wait_for_ing()
    
        # Set up the base scenario for the tests
>       cls._load_schemas(
            cls.downstream_db,
            [
                "api_deletedaudio",
                "api_deletedimage",
                "api_matureaudio",
                "api_matureimage",
                "audio",
                "audioset",
                "image",
            ],
        )

test/integration_test.py:414: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'test.integration_test.TestIngestion'>
conn = <connection object at 0x7f97bfacd580; dsn: 'user=deploy password=xxx connect_timeout=5 dbname=openledger host=localhost port=65432', closed: 0>
schema_names = ['api_deletedaudio', 'api_deletedimage', 'api_matureaudio', 'api_matureimage', 'audio', 'audioset', ...]

    @classmethod
    def _load_schemas(cls, conn, schema_names):
        cur = conn.cursor()
        for schema_name in schema_names:
            schema_path = this_dir.joinpath("mock_schemas", f"{schema_name}.sql")
            with open(schema_path) as schema:
>               cur.execute(schema.read())
E               psycopg2.errors.DuplicateTable: relation "api_deletedaudio" already exists

test/integration_test.py:154: DuplicateTable

Are you able to reproduce these errors?

[AetherUnbound]

Hmm, I'm not seeing those errors when I run things locally 😮 can you try a few things?

• just ingestion_server/install
• just down -v
• docker-compose --profile=ingestion_server -f ingestion_server/test/integration-docker-compose.yml down -v

And then try just build and just ingestion_server/test-local again to see if that works?

[ngken0995]

@AetherUnbound Thank you for giving the suggestions. In the end, I got a macbook and the error doesn't exist anymore. It was an investment to make on getting a new laptop during the holiday.

ingestion_server/test/integration_test.py might be a bit more complicated than expected. cb_queue, upstream_db and downstream_db are declared in setUpClass and called throughout the TestIngestion class. unittest.TestCase used setUpClass to allow us to define instructions that will execute before the class runs it's test.

I tried various ways to use setUpClass as a pytest.fixture, call it only on the first test because there might be a duplicate error. I'm not sure how to use declare cb_queue, upstream_dbanddownstream_db` as global variables and used on other tests.

catalog/tests/dags/common/sensors/test_single_run_external_dags_sensor.py test are converted to pytest.

Please provide any guidance on ingestion_server/test/integration_test.py will be great.

[AetherUnbound]

Glad you were able to get that first issue worked out!

I tried various ways to use setUpClass as a pytest.fixture

I think you may want to look into fixture scope for this! I believe setting the scope to module for those fixtures should ensure they're only spun up once for the tests in that module. Alternatively, you could use package or session if they need to be available for other tests outside that module. Hopefully that helps, let me know if you might need any more information!

[ngken0995]

Thank you for the advice! I will look into it.