use pytest to manage docker container startup for tests

Author: d-v-b

.github/workflows/test.yaml

@@ -34,9 +34,7 @@ jobs:
           python-version: ${{matrix.py_ver}}
       - name: Integration test
         env:
-          PY_VER: ${{matrix.py_ver}}
           MYSQL_VER: ${{matrix.mysql_ver}}
-          # taking default variables set in docker-compose.yaml to sync with local test
         run: |
-          export HOST_UID=$(id -u)
-          docker compose --profile test up --quiet-pull --build --exit-code-from djtest djtest
+          pip install -e ".[test]"
+          pytest --cov-report term-missing --cov=datajoint tests

docker-compose.yaml

@@ -1,4 +1,5 @@
-# HOST_UID=$(id -u) PY_VER=3.11 DJ_VERSION=$(grep -oP '\d+\.\d+\.\d+' datajoint/version.py) docker compose --profile test up --build --exit-code-from djtest djtest
+# Development environment with MySQL and MinIO services
+# To run tests: pytest --cov-report term-missing --cov=datajoint tests
 services:
   db:
     image: datajoint/mysql:${MYSQL_VER:-8.0}
@@ -64,26 +65,3 @@ services:
     user: ${HOST_UID:-1000}:mambauser
     volumes:
       - .:/src
-  djtest:
-    extends:
-      service: app
-    profiles: ["test"]
-    command:
-      - sh
-      - -c
-      - |
-        set -e
-        pip install -q -e ".[test]"
-        pip freeze | grep datajoint
-        pytest --cov-report term-missing --cov=datajoint tests
-  djtest-zarr:
-    extends:
-      service: app
-    profiles: ["test"]
-    command:
-      - sh
-      - -c
-      - |
-        set -e
-        pip install -q -e ".[test]"
-        pytest --cov-report term-missing --cov=datajoint tests/test_zarr.py

pyproject.toml

@@ -21,10 +21,10 @@ dependencies = [
   "faker",
   "urllib3",
   "setuptools",
-  "zarr-python >=3.1.0",
+  "zarr >=3.1.0",
   "typing-extensions>=4.15.0"
 ]
-requires-python = ">=3.9,<4.0"
+requires-python = ">=3.11,<4.0"
 authors = [
   {name = "Dimitri Yatsenko", email = "[email protected]"},
   {name = "Thinh Nguyen", email = "[email protected]"},
@@ -80,11 +80,15 @@ Repository = "https://github.com/datajoint/datajoint-python"
 dj = "datajoint.cli:cli"
 datajoint = "datajoint.cli:cli"
 
-[project.optional-dependencies]
+[dependency-groups]
 test = [
   "pytest",
   "pytest-cov",
+  "docker",
+  "requests",
 ]
+
+[project.optional-dependencies]
 dev = [
   "pre-commit",
   "black==24.2.0",

tests/conftest.py

@@ -1,14 +1,20 @@
+import atexit
 import json
+import logging
 import os
 import shutil
+import signal
+import time
 from os import environ, remove
 from pathlib import Path
 from typing import Dict, List
 
 import certifi
+import docker
 import minio
 import networkx as nx
 import pytest
+import requests
 import urllib3
 from packaging import version
 
@@ -24,6 +30,240 @@
 from . import schema_uuid as schema_uuid_module
 
 
+# Configure logging for container management
+logger = logging.getLogger(__name__)
+
+
+
+
+# Global container registry for cleanup
+_active_containers = set()
+_docker_client = None
+
+
+def _get_docker_client():
+    """Get or create docker client"""
+    global _docker_client
+    if _docker_client is None:
+        _docker_client = docker.from_env()
+    return _docker_client
+
+
+def _cleanup_containers():
+    """Clean up any remaining containers"""
+    if _active_containers:
+        logger.info(f"Emergency cleanup: {len(_active_containers)} containers to clean up")
+        try:
+            client = _get_docker_client()
+            for container_id in list(_active_containers):
+                try:
+                    container = client.containers.get(container_id)
+                    container.remove(force=True)
+                    logger.info(f"Emergency cleanup: removed container {container_id[:12]}")
+                except docker.errors.NotFound:
+                    logger.debug(f"Container {container_id[:12]} already removed")
+                except Exception as e:
+                    logger.error(f"Error cleaning up container {container_id[:12]}: {e}")
+                finally:
+                    _active_containers.discard(container_id)
+        except Exception as e:
+            logger.error(f"Error during emergency cleanup: {e}")
+    else:
+        logger.debug("No containers to clean up")
+
+
+def _register_container(container):
+    """Register a container for cleanup"""
+    _active_containers.add(container.id)
+    logger.debug(f"Registered container {container.id[:12]} for cleanup")
+
+
+def _unregister_container(container):
+    """Unregister a container from cleanup"""
+    _active_containers.discard(container.id)
+    logger.debug(f"Unregistered container {container.id[:12]} from cleanup")
+
+
+# Register cleanup functions
+atexit.register(_cleanup_containers)
+
+
+def _signal_handler(signum, frame):
+    """Handle signals to ensure container cleanup"""
+    logger.warning(f"Received signal {signum}, performing emergency container cleanup...")
+    _cleanup_containers()
+
+    # Restore default signal handler and re-raise the signal
+    # This allows pytest to handle the cancellation normally
+    signal.signal(signum, signal.SIG_DFL)
+    os.kill(os.getpid(), signum)
+
+
+# Register signal handlers for graceful cleanup, but only for non-interactive scenarios
+# In pytest, we'll rely on fixture teardown and atexit handlers primarily
+try:
+    import pytest
+    # If we're here, pytest is available, so only register SIGTERM (for CI/batch scenarios)
+    signal.signal(signal.SIGTERM, _signal_handler)
+    # Don't intercept SIGINT (Ctrl+C) to allow pytest's normal cancellation behavior
+except ImportError:
+    # If pytest isn't available, register both handlers
+    signal.signal(signal.SIGINT, _signal_handler)
+    signal.signal(signal.SIGTERM, _signal_handler)
+
+
+@pytest.fixture(scope="session")
+def docker_client():
+    """Docker client for managing containers."""
+    return _get_docker_client()
+
+
+@pytest.fixture(scope="session")
+def mysql_container(docker_client):
+    """Start MySQL container and wait for it to be healthy."""
+    mysql_ver = os.environ.get("MYSQL_VER", "8.0")
+    container_name = f"datajoint_test_mysql_{os.getpid()}"
+
+    logger.info(f"Starting MySQL container {container_name} with version {mysql_ver}")
+
+    # Remove existing container if it exists
+    try:
+        existing = docker_client.containers.get(container_name)
+        logger.info(f"Removing existing MySQL container {container_name}")
+        existing.remove(force=True)
+    except docker.errors.NotFound:
+        logger.debug(f"No existing MySQL container {container_name} found")
+
+    # Start MySQL container
+    container = docker_client.containers.run(
+        f"datajoint/mysql:{mysql_ver}",
+        name=container_name,
+        environment={
+            "MYSQL_ROOT_PASSWORD": "password"
+        },
+        command="mysqld --default-authentication-plugin=mysql_native_password",
+        ports={"3306/tcp": None},  # Let Docker assign random port
+        detach=True,
+        remove=True,
+        healthcheck={
+            "test": ["CMD", "mysqladmin", "ping", "-h", "localhost"],
+            "timeout": 30000000000,  # 30s in nanoseconds
+            "retries": 5,
+            "interval": 15000000000,  # 15s in nanoseconds
+        }
+    )
+
+    # Register container for cleanup
+    _register_container(container)
+    logger.info(f"MySQL container {container_name} started with ID {container.id[:12]}")
+
+    # Wait for health check
+    max_wait = 120  # 2 minutes
+    start_time = time.time()
+    logger.info(f"Waiting for MySQL container {container_name} to become healthy (max {max_wait}s)")
+
+    while time.time() - start_time < max_wait:
+        container.reload()
+        health_status = container.attrs["State"]["Health"]["Status"]
+        logger.debug(f"MySQL container {container_name} health status: {health_status}")
+        if health_status == "healthy":
+            break
+        time.sleep(2)
+    else:
+        logger.error(f"MySQL container {container_name} failed to become healthy within {max_wait}s")
+        container.remove(force=True)
+        raise RuntimeError("MySQL container failed to become healthy")
+
+    # Get the mapped port
+    port_info = container.attrs["NetworkSettings"]["Ports"]["3306/tcp"]
+    if port_info:
+        host_port = port_info[0]["HostPort"]
+        logger.info(f"MySQL container {container_name} is healthy and accessible on localhost:{host_port}")
+    else:
+        raise RuntimeError("Failed to get MySQL port mapping")
+
+    yield container, "localhost", int(host_port)
+
+    # Cleanup
+    logger.info(f"Cleaning up MySQL container {container_name}")
+    _unregister_container(container)
+    container.remove(force=True)
+    logger.info(f"MySQL container {container_name} removed")
+
+
+@pytest.fixture(scope="session")
+def minio_container(docker_client):
+    """Start MinIO container and wait for it to be healthy."""
+    minio_ver = os.environ.get("MINIO_VER", "RELEASE.2025-02-28T09-55-16Z")
+    container_name = f"datajoint_test_minio_{os.getpid()}"
+
+    logger.info(f"Starting MinIO container {container_name} with version {minio_ver}")
+
+    # Remove existing container if it exists
+    try:
+        existing = docker_client.containers.get(container_name)
+        logger.info(f"Removing existing MinIO container {container_name}")
+        existing.remove(force=True)
+    except docker.errors.NotFound:
+        logger.debug(f"No existing MinIO container {container_name} found")
+
+    # Start MinIO container
+    container = docker_client.containers.run(
+        f"minio/minio:{minio_ver}",
+        name=container_name,
+        environment={
+            "MINIO_ACCESS_KEY": "datajoint",
+            "MINIO_SECRET_KEY": "datajoint"
+        },
+        command=['server', '--address', ':9000', '/data'],
+        ports={"9000/tcp": None},  # Let Docker assign random port
+        detach=True,
+        remove=True
+    )
+
+    # Register container for cleanup
+    _register_container(container)
+    logger.info(f"MinIO container {container_name} started with ID {container.id[:12]}")
+
+    # Get the mapped port
+    container.reload()
+    port_info = container.attrs["NetworkSettings"]["Ports"]["9000/tcp"]
+    if port_info:
+        host_port = port_info[0]["HostPort"]
+        logger.info(f"MinIO container {container_name} mapped to localhost:{host_port}")
+    else:
+        raise RuntimeError("Failed to get MinIO port mapping")
+
+    # Wait for MinIO to be ready
+    minio_url = f"http://localhost:{host_port}"
+    max_wait = 60
+    start_time = time.time()
+    logger.info(f"Waiting for MinIO container {container_name} to become ready (max {max_wait}s)")
+
+    while time.time() - start_time < max_wait:
+        try:
+            response = requests.get(f"{minio_url}/minio/health/live", timeout=5)
+            if response.status_code == 200:
+                logger.info(f"MinIO container {container_name} is ready and accessible at {minio_url}")
+                break
+        except requests.exceptions.RequestException:
+            logger.debug(f"MinIO container {container_name} not ready yet, retrying...")
+            pass
+        time.sleep(2)
+    else:
+        logger.error(f"MinIO container {container_name} failed to become ready within {max_wait}s")
+        container.remove(force=True)
+        raise RuntimeError("MinIO container failed to become ready")
+
+    yield container, "localhost", int(host_port)
+
+    # Cleanup
+    logger.info(f"Cleaning up MinIO container {container_name}")
+    _unregister_container(container)
+    container.remove(force=True)
+    logger.info(f"MinIO container {container_name} removed")
+
+
 @pytest.fixture(scope="session")
 def prefix():
     return os.environ.get("DJ_TEST_DB_PREFIX", "djtest")
@@ -56,18 +296,20 @@ def enable_filepath_feature(monkeypatch):
 
 
 @pytest.fixture(scope="session")
-def db_creds_test() -> Dict:
+def db_creds_test(mysql_container) -> Dict:
+    _, host, port = mysql_container
     return dict(
-        host=os.getenv("DJ_TEST_HOST", "db"),
+        host=f"{host}:{port}",
         user=os.getenv("DJ_TEST_USER", "datajoint"),
         password=os.getenv("DJ_TEST_PASSWORD", "datajoint"),
     )
 
 
 @pytest.fixture(scope="session")
-def db_creds_root() -> Dict:
+def db_creds_root(mysql_container) -> Dict:
+    _, host, port = mysql_container
     return dict(
-        host=os.getenv("DJ_HOST", "db"),
+        host=f"{host}:{port}",
         user=os.getenv("DJ_USER", "root"),
         password=os.getenv("DJ_PASS", "password"),
     )
@@ -190,9 +432,10 @@ def connection_test(connection_root, prefix, db_creds_test):
 
 
 @pytest.fixture(scope="session")
-def s3_creds() -> Dict:
+def s3_creds(minio_container) -> Dict:
+    _, host, port = minio_container
     return dict(
-        endpoint=os.environ.get("S3_ENDPOINT", "minio:9000"),
+        endpoint=f"{host}:{port}",
         access_key=os.environ.get("S3_ACCESS_KEY", "datajoint"),
         secret_key=os.environ.get("S3_SECRET_KEY", "datajoint"),
         bucket=os.environ.get("S3_BUCKET", "datajoint.test"),

tests/test_json.py

@@ -7,8 +7,18 @@
 import datajoint as dj
 from datajoint.declare import declare
 
-if Version(dj.conn().query("select @@version;").fetchone()[0]) < Version("8.0.0"):
-    pytest.skip("These tests require MySQL >= v8.0.0", allow_module_level=True)
+
+def mysql_version_check(connection):
+    """Check if MySQL version is >= 8.0.0"""
+    version_str = connection.query("select @@version;").fetchone()[0]
+    if Version(version_str) < Version("8.0.0"):
+        pytest.skip("These tests require MySQL >= v8.0.0")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def check_mysql_version(connection_root):
+    """Automatically check MySQL version for all tests in this module"""
+    mysql_version_check(connection_root)
 
 
 class Team(dj.Lookup):