[PYTHON] Korrigieren Sie die Datenbank mit pytest-docker

Das Folgende ist als "Beispiel einer Pytest-Umgebung zum Reparieren einer Datenbank mit Docker" vorbereitet.

Dieser Artikel war MySQL, aber ich habe ihn auch mit PostgreSQL überprüft.

Als Bibliothek der Datenbankverbindungsschnittstelle

Für MySQL: PyMySQL
Für PostgreSQL: psycopg2

Ich habe verwendet, aber ich denke, es ist fast das gleiche, wenn ich andere Dinge benutze.

Fix Fixture Beispiel in conftest.py

In der Stichprobe von pytest-docker, die auf den Start von [httpbin] wartet (https://httpbin.org)

def is_responsive(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return True
    except ConnectionError:
        return False


@pytest.fixture(scope="session")
def http_service(docker_ip, docker_services):
    """Ensure that HTTP service is up and responsive."""

    # `port_for` takes a container port and returns the corresponding host port
    port = docker_services.port_for("httpbin", 80)
    url = "http://{}:{}".format(docker_ip, port)
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_responsive(url)
    )
    return url

Datenbankversion davon. Auf der Anwendungsseite habe ich versucht, selbst bei Verwendung von ORM wie SQLAlchemy eine Verbindung mit der Direktverbindungsbibliothek herzustellen. Ich bin mir nicht sicher, ob es das Timing von SQLAlchemys create_engine () über ORM ist. Es ist so, als würde man klarstellen, dass Sie eine Verbindung über die Treiberbibliothek herstellen und den Start überprüfen.

Für MySQL-Container

Warten Sie, bis der MySQL-Container gestartet wird und es keine Ausnahmen für connect () gibt.

def is_mysqld_ready(docker_ip):
    try:
        pymysql.connect(
            host=docker_ip,
            user=os.getenv('MYSQL_USER', ''),
            password=os.getenv('MYSQL_PASSWORD', ''),
            db=os.getenv('MYSQL_DATABASE', '')
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_mysqld_ready(docker_ip)
    )
    return

Für PostgreSQL-Container

Es ist das gleiche.

def is_postgresql_ready(docker_ip):
    try:
        psycopg2.connect(
            "postgresql://{user}:{password}@{host}/{db}".format(
                user=os.getenv('POSTGRES_USER', ''),
                password=os.getenv('POSTGRES_PASSWORD', ''),
                host=docker_ip,
                db=os.getenv('POSTGRES_DB', '')
            )
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_postgresql_ready(docker_ip)
    )
    return

docker-compose.yml

#Für MySQL
version: "3"

services:
  database:
    image: mysql:5.7
    ports:
      - 3306:3306
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
      - MYSQL_USER=${MYSQL_USER}
      - MYSQL_PASSWORD=${MYSQL_PASSWORD}
      - MYSQL_DATABASE=${MYSQL_DATABASE}

#Für PostgreSQL
version: "3"

services:
  database:
    image: postgres:13
    ports:
      - 5432:5432
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}

Erstellung von Schema- und Testdaten

Danach mounten Sie das Volume mit Docker-Compose und initialisieren es. Legen Sie das SQL- und Initialisierungs-Shell-Skript usw. in den Ordner . / Initdb.d von ./initdb.d: / docker-entrypoint-initdb.d. Wenn Sie dies tun, können Sie es testen.

Qiitas Artikel enthält Daten initialisieren, wenn MySQL mit Docker startet.