Skip to content

Reproducible test case for issue 707 #729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions modules/mysql/tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from pathlib import Path
import pytest
import docker


# build a custom MySQL image
@pytest.fixture(scope="session")
def mysql_custom_image() -> str:
tag = "mysql-custom:8.0"
client = docker.from_env()
DOCKERFILE_PATH = (Path(__file__).parent / "seeds").absolute().as_posix()
image, _ = client.images.build(path=DOCKERFILE_PATH, tag=tag)
return tag
28 changes: 28 additions & 0 deletions modules/mysql/tests/seeds/Dockerfile
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
#
# custom mysql image
#
ARG MYSQL_VERSION="8.0"
FROM mysql:${MYSQL_VERSION} as builder

# That file does the DB initialization but also runs mysql daemon, by removing the last line it will only init
RUN ["sed", "-i", "s/exec \"$@\"/echo \"not running $@\"/", "/usr/local/bin/docker-entrypoint.sh"]

# needed for intialization
ENV MYSQL_ROOT_PASSWORD=root
ENV MYSQL_DATABASE=test MYSQL_USER=test MYSQL_PASSWORD=test

COPY *.sql /docker-entrypoint-initdb.d/

# Need to change the datadir to something other than /var/lib/mysql because the parent docker file defines it as a volume.
# https://docs.docker.com/engine/reference/builder/#volume :
# Changing the volume from within the Dockerfile: If any build steps change the data within the volume after
# it has been declared, those changes will be discarded.
RUN ["/usr/local/bin/docker-entrypoint.sh", "mysqld", "--datadir", "/initialized-db" ]

#
# create final container
#
FROM mysql:${MYSQL_VERSION}

# copy datadir from builder
COPY --from=builder /initialized-db /var/lib/mysql
44 changes: 44 additions & 0 deletions modules/mysql/tests/test_mysql_custom_image.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
import pytest
import re
import sqlalchemy

from testcontainers.core.utils import is_arm
from testcontainers.mysql import MySqlContainer

# imports for issue 707
from testcontainers.core.waiting_utils import wait_for_logs
from types import MethodType


# Testcontainers-Python 4.8.2 / main branch: This method gets the following error:
# E TimeoutError: Container did not emit logs satisfying predicate in 120.000 seconds
@pytest.mark.skipif(is_arm(), reason="mysql container not available for ARM")
def test_docker_run_mysql_8_custom(mysql_custom_image):
config = MySqlContainer(mysql_custom_image)
with config as mysql:
engine = sqlalchemy.create_engine(mysql.get_connection_url())
with engine.begin() as connection:
result = connection.execute(sqlalchemy.text("select * from stuff"))
assert len(list(result)) == 4, "Should have gotten all the stuff"


# Testcontainers-Python 4.8.2 / main branch: This method works
@pytest.mark.skipif(is_arm(), reason="mysql container not available for ARM")
def test_docker_run_mysql_8_custom_overwrite_connect_method(mysql_custom_image):
config = MySqlContainer(mysql_custom_image)

# 20241025 patch the _connect method to change the wait_for_logs regex
def _connect(self) -> None:
wait_for_logs(
self,
re.compile(".* ready for connections.* ready for connections.*", flags=re.DOTALL | re.MULTILINE).search,
)

config._connect = MethodType(_connect, config)

with config as mysql:
mysql_url = mysql.get_connection_url()
engine = sqlalchemy.create_engine(mysql_url)
with engine.begin() as connection:
result = connection.execute(sqlalchemy.text("select * from stuff"))
assert len(list(result)) == 4, "Should have gotten all the stuff"