🐛 Flaky integration tests relying on migration as a service

packages/pytest-simcore/src/pytest_simcore/helpers/utils_postgres.py

@@ -1,15 +1,63 @@
+import logging
 from contextlib import contextmanager
 from copy import deepcopy
-from typing import Any, Iterator
+from typing import Any, Final, Iterator
 
+import docker
 import simcore_postgres_database.cli
 import sqlalchemy as sa
+from docker.models.services import Service
 from simcore_postgres_database.models.base import metadata
+from tenacity import TryAgain, retry
+from tenacity.before_sleep import before_sleep_log
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_fixed
+
+log = logging.getLogger(__name__)
+
+
+_MINUTE: Final[int] = 60
+_LOG_HEAD_MIGRATION: Final[str] = "Migration service"
+
+
+def _get_migration_service(docker_client: docker.DockerClient) -> Service | None:
+    service: Service
+    for service in docker_client.services.list():
+        if "migration" in service.name:
+            return service
+
+    return None
+
+
+def _was_migration_service_started(docker_client: docker.DockerClient) -> bool:
+    service: Service | None = _get_migration_service(docker_client)
+    return service is not None
+
+
+def _did_migration_service_finished_postgres_migration(
+    docker_client: docker.DockerClient,
+) -> bool:
+    service: Service | None = _get_migration_service(docker_client)
+    assert service is not None
+
+    logs = [x.decode() for x in service.logs(stdout=True)]
+    return "Migration Done. Wait forever ..." in "\n".join(logs)
+
+
+@retry(
+    wait=wait_fixed(0.5),
+    stop=stop_after_delay(2 * _MINUTE),
+    before_sleep=before_sleep_log(log, logging.WARNING),
+    reraise=True,
+)
+def wait_for_migration_service(docker_client: docker.DockerClient) -> None:
+    if not _did_migration_service_finished_postgres_migration(docker_client):
+        raise TryAgain(f"{_LOG_HEAD_MIGRATION} did not finish Postgres migration")
 
 
 @contextmanager
 def migrated_pg_tables_context(
-    postgres_config: dict[str, str],
+    docker_client: docker.DockerClient, postgres_config: dict[str, str]
 ) -> Iterator[dict[str, Any]]:
     """
     Within the context, tables are created and dropped
@@ -23,6 +71,15 @@ def migrated_pg_tables_context(
         )
     )
 
+    # NOTE: if migration service was also started we should wait for the service
+    # to finish migrating Postgres, before trying to run the migrations again
+    if _was_migration_service_started(docker_client):
+        log.info("%s is running, attending for it to be idle", _LOG_HEAD_MIGRATION)
+        wait_for_migration_service(docker_client)
+        log.info("%s is now idle!", _LOG_HEAD_MIGRATION)
+    else:
+        log.info("%s is not present", _LOG_HEAD_MIGRATION)
+
     simcore_postgres_database.cli.discover.callback(**postgres_config)
     simcore_postgres_database.cli.upgrade.callback("head")
 

packages/pytest-simcore/src/pytest_simcore/postgres_service.py

@@ -4,6 +4,7 @@
 
 from typing import AsyncIterator, Final, Iterator, TypedDict
 
+import docker
 import pytest
 import sqlalchemy as sa
 import tenacity
@@ -188,10 +189,11 @@ def postgres_dsn_url(postgres_dsn: PostgresTestConfig) -> str:
 def postgres_db(
     postgres_dsn: PostgresTestConfig,
     postgres_engine: sa.engine.Engine,
+    docker_client: docker.DockerClient,
 ) -> Iterator[sa.engine.Engine]:
     """An postgres database init with empty tables and an sqlalchemy engine connected to it"""
 
-    with migrated_pg_tables_context(postgres_dsn.copy()):
+    with migrated_pg_tables_context(docker_client, postgres_dsn.copy()):
         yield postgres_engine