背景

当前DTS项目使用pytest测试框架运行自动化测试用例，但由于测试环境的资源限制，只为自动化测试项目部署了一个源数据库和一个目标数据库，这导致有一些测试用例无法开发。比如MYSQL->MYSQL的版本检查、源库binlog存在性检查、源库binlog是否开启检查、源库binlog影像类型检查、源库用户权限检查、源库连通性检查、MySQL参数lower_case_table_names一致性检查等等。

上面列举的测试用例只是开发了正向用例，因为只有一套数据库可用，无法做反向用例，如果通过更改数据库配置来满足反向用例的条件：

1、手工执行重启命令并指定启动参数，繁琐而且这也不是自动化测试了；

2、需要的时间长，拖慢自动化测试的进度；

3、更改重要的参数再改回来可能会影响其他一些测例的执行。

4、数据库版本检查的测例只能通过部署多套不同版本的数据库来完成。

综合以上的问题，目前DTS的自动化测试用例还是缺少很多，只能由测试人员手工执行，这消耗了测试人员的很多时间。

 

解决方案

容器以轻量、方便而著称，但是测试环境资源有限，测试用例多，启动多个容器实例所需的资源无法满足，手工管理也很麻烦。

如果通过编程语言远程启动docker容器来代替人为操作，需要时启动容器，用完及时释放资源，就可以大大节省人力和硬件资源。直接使用python的docker依赖来进行容器操作是一种可能的实施方案，如下启动一个MySQL容器。

    client = docker.from_env()
    # 启动MySQL容器
    container: Container = client.containers.run(
        'mysql:5.7',
        command='--lower_case_table_names=1',
        detach=True,
        name='mysql-container',
        ports={'3306/tcp': 13306},
        environment={
            'MYSQL_ROOT_PASSWORD': 'password',
            'MYSQL_USER': 'user',
            'MYSQL_PASSWORD': 'password',
            'MYSQL_DATABASE': 'mydb'
        }
    )
   # 删除容器 
   container.remove()

不过开源框架testcontainers已经封装提供了更为方便的实现。

 

TestContainers

官网地址：https://www.testcontainers.org

TestContainers是一个开源项目，它提供诸多可以在Docker容器中运行的组件，非常轻量、方便，需要做的只是安装docker服务，无需其他配置。它支持Java，Python，Rust，Go，.net等多种语言，可以提供测试所需的多种环境。

testcontainers支持众多常用的主流组件，以Java为例，支持如下组件

 

其中支持的Databases：

不是所用语言的库都支持如此多的组件，go、node.js的testcontainers库只支持寥寥几种组件。

 

 

使用介绍

使用前提：test-containers 基于 Docker，所以使用 test-container 前需要安装 Docker环境。

不同版本testcontainers-python的API差异很大，这里使用的相关依赖的版本如下

SQLAlchemy~=1.4.46
testcontainers~=3.7.0
urllib3~=1.25.11

 

官方示例

testcontainers-python文档地址：https://testcontainers-python.readthedocs.io/en/latest/README.html

testcontainers-python给出的官方示例代码都很短小

def test_docker_run_mysql():
    config = MySqlContainer('mysql:5.7.17')
    with config as mysql:
        engine = sqlalchemy.create_engine(mysql.get_connection_url())
        with engine.begin() as connection:
            result = connection.execute(sqlalchemy.text("select version()"))
            for row in result:
                assert row[0].startswith('5.7.17')

以上python代码启动了5.7.17版本的容器实例，并且使用python的ORM框架sqlalchemy去连接数据库，获取数据库连接后执行SQL语句并返回结果。

 

继承关系

事实上，所有的特性Container都直接或间接派生自DockerContainer。以下是继承关系：

DockerContainer
    |-KafkaContainer
    |-ElasticSearchContainer
    |-NginxContainer
    |-DbContainer
        |-MySqlContainer
        |-SqlServerContainer
        |-OracleDbContainer

DockerContainer封装了容器的通用操作，DbContainer封装了数据库的通用操作。而DockerContainer底层使用python的docker依赖来进行容器操作的。

 

DockerContainer

这是最灵活也是不太方便的容器类型， 此容器允许使用启动任何Docker镜像。

 

DockerContainer的构造函数

    def __init__(self, image, docker_client_kw: dict = None, **kwargs):
        self.env = {}
        self.ports = {}
        self.volumes = {}
        self.image = image
        self._docker = DockerClient(**(docker_client_kw or {}))
        self._container = None
        self._command = None
        self._name = None
        self._kwargs = kwargs

with_env方法用于设置容器的环境变量

    def with_env(self, key: str, value: str) -> 'DockerContainer':
        self.env[key] = value
        return self

with_command方法指定容器启动时的参数

    def with_command(self, command: str) -> 'DockerContainer':
        self._command = command
        return self

with_bind_ports用于设置一对绑定的端口

    def with_bind_ports(self, container: int,
                        host: int = None) -> 'DockerContainer':
        self.ports[container] = host
        return self

with_exposed_ports方法用于暴露一个内部端口，绑定的宿主机端口是随机的

    def with_exposed_ports(self, *ports) -> 'DockerContainer':
        for port in list(ports):
            self.ports[port] = None
        return self

get_exposed_port方法用来获取指定的内部端口所绑定的宿主机端口

    @wait_container_is_ready()
    def get_exposed_port(self, port) -> str:
        mapped_port = self.get_docker_client().port(self._container.id, port)
        if inside_container():
            gateway_ip = self.get_docker_client().gateway_ip(self._container.id)
            host = self.get_docker_client().host()
​
            if gateway_ip == host:
                return port
        return mapped_port

with_volume_mapping方法用于挂载数据卷

    def with_volume_mapping(self, host: str, container: str,
                            mode: str = 'ro') -> 'DockerContainer':
        # '/home/user1/': {'bind': '/mnt/vol2', 'mode': 'rw'}
        mapping = {'bind': container, 'mode': mode}
        self.volumes[host] = mapping
        return self

exec方法用于在容器内部执行指令

    def exec(self, command):
        if not self._container:
            raise ContainerStartException("Container should be started before")
        return self.get_wrapped_container().exec_run(command)

get_wrapped_container方法返回container对象，这才真正对应着一个里面有所有的容器信息

    def get_wrapped_container(self) -> Container:
        return self._container

 

可以使用DockerContainer启动任何类型的容器，比如启动一个MySQL容器

    container = DockerContainer("mysql:5.7")
    container.with_env("MYSQL_ROOT_PASSWORD", "afcer554KCJ5")
    container.with_command("--lower_case_table_names=1")
    container.with_exposed_ports(3306)
    container.start()

事实上，MysqlContainer与DockerContainer差异很小，MysqlContainer只是多做了一点事情(设置环境变量、检查容器是否启动就绪)，自动化测试直接使用DockerContainer也能满足需求，当然使用MysqlContainer会更加方便一些。

 

MysqlContainer

MySqlContainer继承自DbContainer，DbContainer继承自DockerContainer。

MysqlContainer构造函数

    def __init__(self,
                 image="mysql:latest",
                 MYSQL_USER=None,
                 MYSQL_ROOT_PASSWORD=None,
                 MYSQL_PASSWORD=None,
                 MYSQL_DATABASE=None,
                 **kwargs):
        super(MySqlContainer, self).__init__(image, **kwargs)
        self.port_to_expose = 3306
        self.with_exposed_ports(self.port_to_expose)
        self.MYSQL_USER = MYSQL_USER or environ.get('MYSQL_USER', 'test')
        self.MYSQL_ROOT_PASSWORD = MYSQL_ROOT_PASSWORD or environ.get('MYSQL_ROOT_PASSWORD', 'test')
        self.MYSQL_PASSWORD = MYSQL_PASSWORD or environ.get('MYSQL_PASSWORD', 'test')
        self.MYSQL_DATABASE = MYSQL_DATABASE or environ.get('MYSQL_DATABASE', 'test')
​
        if self.MYSQL_USER == 'root':
            self.MYSQL_ROOT_PASSWORD = self.MYSQL_PASSWORD

镜像默认是latest，最新版。

 

启动容器

创建一个MysqlContainer对象后，只需简单调用start方法即可启动容器

    # 启动容器
    mysqlContainer.start()

start方法的定义

class DbContainer(DockerContainer):
​
    @wait_container_is_ready(*ADDITIONAL_TRANSIENT_ERRORS)
    def _connect(self):
        import sqlalchemy
        engine = sqlalchemy.create_engine(self.get_connection_url())
        engine.connect()
​
    def start(self):
        # 配置环境变量
        self._configure()
        # 这一步就是把容器启动起来
        super().start()
        # 验证容器服务是否可用
        self._connect()
        return self
​

在start方法里的最后调用_connect()方法是为了验证容器服务是否可用。_connect方法使用SQLAlchemy来连接数据库，SQLAlchemy通过容器的get_connection_url()方法获取数据库地址进行连接。_connect方法添加了@wait_container_is_ready注解，会一直等待直到数据库连接成功，超时时间120s。

SQLAlchemy是一个开源的Python ORM（对象关系映射）框架。

 

在windows上进行测试时，MysqlContainer的get_connection_url方法返回如下的URL对象

mysql+pymysql://test_asd:afrvte54657hnngf@localnpipe:65303/test

PostgresContainer的get_connection_url方法返回如下的URL对象

postgresql+psycopg2://postgres:***@localnpipe:64211/postgres

这里显然是通过named pipe来连接数据库，然而连接失败次数达到指定值，容器启动失败。

Windows命名管道只能用于Windows主机上的进程间通信，WSL上运行的Docker容器被视为独立的进程空间，因此无法通过命名管道进行通信。与WSL2中运行的Docker容器进行通信，需要使用网络通信协议，如TCP协议。

即使我在Windows本地启动MySQL服务，使用SQLAlchemy连接以上的URL对象也是失败。

 

阅读了testcontainers-python源码，通过继承MySqlContainer重写其get_connection_url()来解决。

from testcontainers.mysql import MySqlContainer
​
​
class CustomMysqlContainer(MySqlContainer):
    def get_connection_url(self):
        return 'mysql+pymysql://{0}:{1}@127.0.0.1:{2}/{3}'.format('root',
                                                                  self.MYSQL_ROOT_PASSWORD,
                                                                  self.get_exposed_port(3306),
                                                                  self.MYSQL_DATABASE)

将host写死为127.0.0.1，这样即可解决。

 

而testcontainers-java则是使用localhost连接的：

19:09:15.663 [main] INFO 🐳 [mysql:5.7.34] - Container mysql:5.7.34 is starting: 18f8abf99e470a467d86d5b37c09be2b7cfd94e5b270d7e633b001fbb3db2ae2
19:09:16.095 [main] INFO 🐳 [mysql:5.7.34] - Waiting for database connection to become available at jdbc:mysql://localhost:64513/test using query 'SELECT 1'
19:09:24.280 [main] INFO 🐳 [mysql:5.7.34] - Container is started (JDBC URL: jdbc:mysql://localhost:64513/test)
19:09:24.281 [main] INFO 🐳 [mysql:5.7.34] - Container mysql:5.7.34 started in PT8.7804708S

 

 

 

创建mysql账号

MysqlContainer构造函数中，可以指定参数MYSQL_USER、MYSQL_PASSWORD、MYSQL_ROOT_PASSWORD、MYSQL_DATABASE来创建账户相关信息。

    mysqlContainer = MysqlContainer(image='mysql:5.7',
                              MYSQL_USER='test_asd',
                              MYSQL_ROOT_PASSWORD='afrvte54657hnngf',
                              MYSQL_PASSWORD='afrvte54657hnngf',
                              MYSQL_DATABASE='test')

在容器启动过程中看到如下日志

[Warning] root@localhost is created with an empty password ! Please consider switching off the --initialize-insecure option.

MySQL初始化时root@localhost用户是空密码。如果设置了MYSQL_ROOT_PASSWORD，容器会在MySQL服务运行后立即设置root@localhost用户密码，并创建一个新用户root@%，两者密码一致。

 

指定端口

MySQL容器内的MySQL服务默认运行在3306号端口，绑定的宿主机端口是随机的。

可以在容器启动前将其绑定到指定的宿主机端口，例如

    mysqlContainer = MysqlContainer(image='mysql:5.7',
                              MYSQL_USER='test_asd',
                              MYSQL_ROOT_PASSWORD='afrvte54657hnngf',
                              MYSQL_PASSWORD='afrvte54657hnngf',
                              MYSQL_DATABASE='test')
    mysqlContainer.with_bind_ports(3306, 26788)

而with_exposed_ports方法也用于暴露指定的内部端口，但绑定的宿主机端口是随机的。

使用with_bind_ports方法绑定指定的宿主机端口需要确保宿主机该端口空闲，而with_exposed_ports会自动寻找宿主机上空闲的一个可用端口。

 

指定启动参数

如果想要指定MySQL的运行参数，可以在容器启动前使用with_command方法来指定MySQL参数，方法的参数格式如下

    mysqlContainer = MysqlContainer(image='mysql:5.7',
                              MYSQL_USER='test_asd',
                              MYSQL_ROOT_PASSWORD='afrvte54657hnngf',
                              MYSQL_PASSWORD='afrvte54657hnngf',
                              MYSQL_DATABASE='test')
    mysqlContainer.with_command('--lower-case-table-names=1')
    # 将把上面设置的command覆盖
    mysqlContainer.with_command('--character_set_server=utf8mb4')
    # 想要同时设置多个MySQL服务启动参数，按如下格式传递参数
    mysqlContainer.with_command('--lower-case-table-names=1 --character_set_server=utf8mb4')

进入容器查看

 

 

容器操作

DockerContainer提供了start和stop方法，每次调用start方法就会启动一个新容器实例并返回，stop方法不是停止容器而是删除容器。

    def start(self):
        logger.info("Pulling image %s", self.image)
        docker_client = self.get_docker_client()
        self._container = docker_client.run(self.image,
                                            command=self._command,
                                            detach=True,
                                            environment=self.env,
                                            ports=self.ports,
                                            name=self._name,
                                            volumes=self.volumes,
                                            **self._kwargs
                                            )
        logger.info("Container started: %s", self._container.short_id)
        return self
​
    def stop(self, force=True, delete_volume=True):
        self.get_wrapped_container().remove(force=force, v=delete_volume)

为了避免容器一直运行，容器使用完后一定要调用DockerContainer的stop方法删除容器实例。

DockerContainer封装了具体的容器实例，提供了一些工具方法如with_env、with_command，它的实例变量_container才对应着一个具体的容器实例。

_container是Container类型的对象，是python的docker依赖提供的。

class Container(Model):
    """ Local representation of a container object. Detailed configuration may
        be accessed through the :py:attr:`attrs` attribute. Note that local
        attributes are cached; users may call :py:meth:`reload` to
        query the Docker daemon for the current properties, causing
        :py:attr:`attrs` to be refreshed.
    """
    
    //...
    
    def pause(self):
        """
        Pauses all processes within this container..
        """
        return self.client.api.pause(self.id)
​
    
    def remove(self, **kwargs):
        """
        Remove this container. Similar to the ``docker rm`` command.
        """
        return self.client.api.remove_container(self.id, **kwargs)
    
    
    def start(self, **kwargs):
        """
        Start this container. Similar to the ``docker start`` command, but
        """
        return self.client.api.start(self.id, **kwargs)
    
    
    def stop(self, **kwargs):
        """
        Stops a container. Similar to the ``docker stop`` command.
        """
        return self.client.api.stop(self.id, **kwargs)
    
    
    def unpause(self):
        """
        Unpause all processes within the container.
        """
        return self.client.api.unpause(self.id)

想要对容器实例进行其他操作，可以通过DockerContainer的get_wrapped_container方法获取_container实例变量。

    container = mysqlContainer.get_wrapped_container()
    #停止活动
    container.pause()
    #继续活动
    container.unpause()

我们可以调用pause方法来模拟数据库不可用状态。

 

性能表现和资源占用

测试环境：Windows WSL2+Windows docker desktop

启动容器的start方法是同步的，在等待容器完全启动后才会返回，这里的容器启动时间7.7s

内存占用

使用testcontainers-python创建容器时，推荐使用官方的Docker镜像，是经过优化和配置的，因此占用的内存资源是相对较少的。

第一次启动容器需要拉取镜像耗时会比较久。

 

 

控制容器启动超时时间

在start方法里，testcontainers通过连接数据库来判断服务是否可用。testcontainers默认设置的最大连接重试次数为120次，间隔1s。

os.environ.setdefault("TC_MAX_TRIES", '120')

调试阶段可以适当调小此值。

虽然单个容器启动速度很快，测试发现，如果连续启动4个MySQL容器，容器的启动时间超过甚至会超过120s导致测试失败(偶尔)。

 

 

PostgresContainer

指定参数

postgresContainer.with_command("-c wal_level=logical -c max_connections=134 -c shared_buffers=2GB")

资源占用

 

 

docker-compose

testcontainers也支持使用docker-compose。

 

创建docker-compose.yml文件，启动一个MySQL和一个postgres。

version: '3'
services:
  db:
    image: mysql:5.7
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: test_db
      MYSQL_USER: test_user
      MYSQL_PASSWORD: test_password
    ports:
      - "13306:3306"
  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB: test_db
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_password
    ports:
      - "15432:5432"

创建DockerCompose，指定docker-compose文件在当前目录下，启动DockerCompose

from testcontainers.compose import DockerCompose
​
​
def test_compose():
    compose = DockerCompose('.')
    compose.start()
​
    mysql_host = compose.get_service_host('db', 3306)
    print(mysql_host)
    mysql_port = compose.get_service_port('db', 3306)
    print(mysql_port)
​
    '''
        Returns tuple[str, str, int] stdout, stderr, return code
    '''
    return_tuple = compose.exec_in_container('db', ['ls'])
    print(return_tuple)
​
    time.sleep(1)
​
    compose.stop()

 

 

 

testcontainers-java

@Testcontainers: 用于启用 Testcontainers 支持。它可以用在类或方法级别，让 JUnit 在测试执行前启动 Docker 容器。例如：

@Testcontainers
public class MyTestClass {
  // ...
}

@Container: 用于将 Docker 容器作为测试类的静态字段启动。这个注解可以与 DockerComposeContainer、GenericContainer 和其他扩展类一起使用。例如：

@Container
public static PostgreSQLContainer postgreSQLContainer = new PostgreSQLContainer();

 

 

@Testcontainers
public class MyTest {
    
    @Container
    private static final MySQLContainer mySQLContainer = (MySQLContainer) new MySQLContainer("mysql:5.7.34")
            .withDatabaseName("test")
            .withUsername("dds")
            .withPassword("asfrer54765dsc")
            .withInitScript("mysqlScript.sql")
            .withEnv("name","value")
            .withCommand("--lower-case-table-names=1 --character_set_server=utf8mb4");
    
    @Test
    public void testMysql() throws SQLException {
        System.out.println(mySQLContainer.getMappedPort(3306));
        try (Connection conn = DriverManager.getConnection(mySQLContainer.getJdbcUrl(), mySQLContainer.getUsername(), mySQLContainer.getPassword())) {
            String sql = "SELECT 2 + 2";
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                try (ResultSet rs = stmt.executeQuery()) {
                    rs.next();
                    int result = rs.getInt(1);
                    Assertions.assertEquals(4, result);
                }
            }
        }
    }

withPassword方法设置的密码也是root用户的密码；

withInitScript可指定MySQL的初始化脚本；

withCommand可指定MySQL参数；

withEnv设置容器环境变量；

getJdbcUrl返回JDBC连接urljdbc:mysql://localhost:51319/test；

getMappedPort方法返回容器内部端口所绑定的宿主机端口。