pytest-docker-tools

您已经编写了一个软件应用程序（任何语言）并将其打包成Docker镜像。现在您想在发布之前对构建的镜像进行烟熏测试或与其他容器进行一些集成测试。

• 您希望以类似docker-compose.yml的方式推理您的环境
• 希望环境在测试运行时自动创建和销毁
• 希望有选项在执行高频测试时重用先前创建的资源（例如容器）
• 不想编写大量样板代码来创建测试环境
• 希望能够并行运行测试
• 希望测试是可靠的

pytest-docker-tools 是一组用于创建 py.test 固定的有偏见的辅助工具，用于您的烟熏测试和集成测试。它努力保持您的环境定义声明性，就像docker-compose.yml一样。它拥抱 py.test 固定加载。 它试图不过于神奇。结果变得有点神奇，但也没有比 py.test 本身更神奇。

此库提供的主要接口是一组 'fixture 工厂'。它提供了一个 '最佳级别' 的固定实现，然后允许您将其作为模板 - 声明性地注入您自己的配置。您可以在您的 conftest.py 中定义您的固定，并在所有测试中访问它们，并且您可以在单个测试模块中根据需要覆盖它们。

该API简单直观，隐式地捕获了规范中固定件的依赖关系。例如，如果您正在构建一个微服务并想指向其DNS和模拟DNS服务器，它可能看起来像这样：

# conftest.py

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container

fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '127.0.0.1',
    }
)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}']
)


@pytest.fixture
def apiclient(apiserver):
    port = apiserver.ports['8080/tcp'][0]
    return HTTPConnection(f'localhost:{port}')

现在您可以创建一个测试来测试您的微服务

# test_smoketest.py

import socket

def test_my_frobulator(apiserver):
    sock = socket.socket()
    sock.connect(('127.0.0.1', apiserver.ports['8080/tcp'][0]))


def test_my_frobulator_works_after_restart(apiserver):
    apiserver.restart()

    sock = socket.socket()
    sock.connect(('127.0.0.1', apiserver.ports['8080/tcp'][0]))

在这个例子中，所有的依赖关系都将按顺序和每个会话仅解决一次

• 将获取最新的redis:latest
• 将基于db文件夹中的Dockerfile构建一个容器镜像。

然后每个测试一次

• 将创建一个新的卷
• 将创建一个新的“后端”容器，从redis:latest创建。它将附加到新的卷。
• 将创建一个新的“前端”容器，从新构建的容器创建。它将通过环境变量获得后端的IP。容器中的端口3679将公开为主机上的临时端口。

然后测试可以运行并通过其临时高端口访问容器。测试结束时，环境将被丢弃。

如果测试失败，将捕获每个容器的docker logs输出并添加到测试输出中。

在示例中，您会注意到我们定义了一个apiclient固定件。当然，如果您使用它，它将隐式地拉入两个服务器固定件，并且“直接工作”

# test_smoketest.py

import json


def test_api_server(apiclient):
    apiclient.request('GET', '/')
    response = apiclient.getresponse()
    assert response.status == 200
    assert json.loads(response.read()) == {'result': '127.0.0.1'}

范围

所有的固定件工厂都接受scope关键字。使用这些工厂创建的固定件将像任何具有该范围的py.test固定件一样表现。

在这个例子中，我们创建了一个session范围的memcache，另一个是module范围的。

# conftest.py

from pytest_docker_tools import container, fetch

memcache_image = fetch(repository='memcached:latest')

memcache_session = container(
    image='{memcache_image.id}',
    scope='session',
    ports={
        '11211/tcp': None,
    },
)

memcache_module = container(
    image='{memcache_image.id}',
    scope='module',
    ports={
        '11211/tcp': None,
    },
)

当test_scope_1.py运行时，两个容器都没有运行，因此将启动每个新实例。它们的范围比单个function长，因此为需要它们的下一个测试保留它们。

# test_scope_1.py

import socket

def test_session_1(memcache_session):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_session.ports['11211/tcp'][0]))
    sock.sendall(b'set mykey 0 600 4\r\ndata\r\n')
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024) == b'STORED\r\nVALUE mykey 0 4\r\ndata\r\nEND\r\n'
    sock.close()

def test_session_2(memcache_session):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_session.ports['11211/tcp'][0]))
    sock.sendall(b'set mykey 0 600 4\r\ndata\r\n')
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024) == b'STORED\r\nVALUE mykey 0 4\r\ndata\r\nEND\r\n'
    sock.close()

def test_module_1(memcache_module):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_module.ports['11211/tcp'][0]))
    sock.sendall(b'set mykey 0 600 4\r\ndata\r\n')
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024) == b'STORED\r\nVALUE mykey 0 4\r\ndata\r\nEND\r\n'
    sock.close()

def test_module_2(memcache_module):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_module.ports['11211/tcp'][0]))
    sock.sendall(b'set mykey 0 600 4\r\ndata\r\n')
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024) == b'STORED\r\nVALUE mykey 0 4\r\ndata\r\nEND\r\n'
    sock.close()

当test_scope_2.py运行时，session范围的容器仍在运行，因此它将被重用。但我们现在处于一个新的模块中，所以module范围的容器将被销毁。将创建一个新实例。

# test_scope_2.py

import socket

def test_session_3(memcache_session):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_session.ports['11211/tcp'][0]))
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024).endswith(b'END\r\n')
    sock.close()

def test_module_3(memcache_module):
    sock = socket.socket()
    sock.connect(('127.0.0.1', memcache_module.ports['11211/tcp'][0]))
    sock.sendall(b'get mykey\r\n')
    assert sock.recv(1024) == b'END\r\n'
    sock.close()

并行性

集成和烟雾测试通常很慢，但大部分时间都花费在等待上。因此，并行运行测试是加快它们的好方法。pytest-docker-tools避免创建可能冲突的资源名称。它还使您不必关心服务绑定到的端口变得容易。这意味着它与pytest-xdist配合使用非常合适。

以下是一个仅测试在xdist下运行的100个redis固定件实例创建和销毁的简单示例。创建一个test_xdist.py插件

import pytest
from pytest_docker_tools import container, fetch

my_redis_image = fetch(repository='redis:latest')

my_redis = container(
    image='{my_redis_image.id}',
)


@pytest.mark.parametrize("i", list(range(100)))
def test_xdist(i, my_redis):
    assert my_redis.status == "running"

并使用以下方法调用它

pytest test_xdist.py -n auto

它将为每个核心创建一个工作进程，并并行运行测试

===================================== test session starts ======================================
platform darwin -- Python 3.6.5, pytest-3.6.3, py-1.5.4, pluggy-0.6.0
rootdir: ~/pytest-docker-tools, inifile:
plugins: xdist-1.22.2, forked-0.2, docker-tools-0.0.2
gw0 [100] / gw1 [100] / gw2 [100] / gw3 [100] / gw4 [100] / gw5 [100] / gw6 [100] / gw7 [100]
scheduling tests via LoadScheduling
......................................................................................... [ 82%]
...........                                                                              [100%]
================================= 100 passed in 70.08 seconds ==================================

工厂参考

容器

要测试中创建容器，请使用container固定件工厂。

from pytest_docker_tools import container

my_microservice_backend = container(image='redis:latest')

此工厂的默认范围是function。这意味着每个测试都会创建一个新的容器。

container固定件工厂支持可以传递给docker-py run方法的任何参数。有关所有参数的详细信息，请参阅此处。

任何字符串变量都将与其他定义的固定件进行插值。这意味着固定件可以依赖于其他固定件，并且它们将按顺序构建和运行。

例如

from pytest_docker_tools import container, fetch

redis_image = fetch(repository='redis:latest')
redis = container(image='{redis_image.id}')


def test_container_starts(redis):
    assert redis.status == "running"

这将首先获取最新的redis:latest，然后从拉取的确切镜像中运行容器。注意，如果您不使用build或fetch来准备Docker镜像，则您指定的标记或哈希必须已经存在于您运行测试的主机上。没有隐式获取Docker镜像。

容器将在测试开始时准备就绪，测试完成后将自动删除。

如果在超时期间（默认为30秒）由于某些原因容器未能就绪，则测试将失败。

timeout 可传递给 container 工厂

from pytest_docker_tools import container, fetch

redis_image = fetch(repository='redis:latest')
redis = container(image='{redis_image.id}', timeout=30)

def test_container_starts(redis):
    assert redis.status == "running"

通过代码定义 Dockerfile 创建容器

import io

from pytest_docker_tools import build, container

dockerfile = io.BytesIO(b"""
FROM alpine:3.12
RUN apk --no-cache add python3
CMD ["tail", "-f", "/dev/null"]
""")

image = build(fileobj=dockerfile)
container = container(image='{image.id}')

def test_container_starts(container):
    assert container.status == "running"

IP地址

如果您的容器仅连接到单个网络，您可以通过辅助属性获取其IP地址。给定此环境

# conftest.py

from pytest_docker_tools import container, fetch, network

redis_image = fetch(repository='redis:latest')
backend_network = network()

redis = container(
  image='{redis_image.id}',
  network='{backend_network.name}',
)

您可以通过容器辅助访问IP地址

import ipaddress

def test_get_service_ip(redis):
    # This will raise a ValueError if not a valid ip
    ipaddress.ip_address(redis.ips.primary)

如果您想按网络查找其IP地址，也可以更具体地访问它

import ipaddress

def test_get_service_ip(backend_network, redis):
    ipaddress.ip_address(redis.ips[backend_network])

端口

工厂采用与官方Python Docker API相同的端口参数。我们建议使用临时高端口的语法

# conftest.py

from pytest_docker_tools import container

apiserver = container(
  image='{apiserver_image.id}',
  ports={'8080/tcp': None}
)

Docker会将容器中的8080端口映射到主机的随机端口。为了从测试中访问它，您可以从容器实例获取已绑定端口

def test_connect_my_service(apiserver):
    assert apiserver.ports['8080/tcp'][0] != 8080

日志

您可以使用 logs 方法检查容器的日志

from pytest_docker_tools import container, fetch


redis_image = fetch(repository='redis:latest')
redis = container(
    image='{redis_image.id}',
    ports={'6379/tcp': None},
)

def test_logs(redis):
    assert 'oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo' in redis.logs()

镜像

要从默认仓库拉取镜像，请使用 fetch 固定工厂。要从本地源构建镜像，请使用 build 固定工厂。如果您正在对已在本地上构建的工件进行烟雾测试，可以使用 image 固定工厂来引用它。

from pytest_docker_tools import build, image, fetch

my_image = fetch(repository='redis:latest')

my_image_2 = build(
  path='db'
)

build 固定工厂支持可以传递给 docker-py build 方法的所有参数。有关所有参数，请参阅此处。 fetch 固定工厂支持可以传递给 docker-py pull 方法的所有参数。有关所有参数，请参阅此处。

此工厂的默认作用域为 session。这意味着该固定工厂将在每次 py.test 调用中只构建或获取一次。固定工厂将在测试（或其他固定工厂）尝试使用它之前不会触发。这意味着如果您没有运行使用它的测试，则不会浪费构建镜像的时间。

缓存

默认情况下，镜像会在调用之间保留。这可以大幅提高效率。但在对镜像进行增量开发时，可能会留下许多孤立的层。运行 docker image prune 将丢弃这些层，但由于这些都是未标记的镜像，它将包括对您当前项目仍然有效的那些。对于大型镜像，这可能会真正减慢您的测试周期。

为了避免这种情况，您需要将镜像标记为 docker image prune 默认不会丢弃标记的镜像。但要使这对于多阶段镜像有效，您还需要标记阶段。为了支持这一点，pytest-docker-tools采用 stages 参数。例如

from pytest_docker_tools import build, image, fetch

my_image = build(
  path='db',
  tag='localhost/myproject:latest',
  stages={
      'builder': 'localhost/myproject:builder'
  }
)

在内部，这将使pytest-docker-tools首先构建（并标记）builder 阶段。这就像运行

docker build --target builder --tag localhost/myproject:builder .

然后在该阶段标记后，它将运行默认的目标，就像之前一样，这就像运行

docker build --tag localhost/myproject:latest .

这将重用先前构建中生成的层（如果适用）。

现在，当您运行 docker image prune 时，最新的镜像构建和它依赖的阶段的最新版本都将保留。

网络

默认情况下，您使用 container() 固定工厂创建的所有容器都将运行在您的默认Docker网络上。您可以使用 network() 固定工厂创建一个专用的测试网络。

from pytest_docker_tools import container, fetch, network

frontend_network = network()

redis_image = fetch(repository='redis:latest')
redis = container(
    image='{redis_image.id}',
    network='{frontend_network.name}',
)

network 固定工厂支持可以传递给 docker-py 网络 create 方法的所有参数。有关所有参数，请参阅此处。

该工厂的默认作用域是 function。这意味着每次执行测试时都会创建一个新的网络。

使用后的网络将在测试完成后被移除。

卷

在理想情况下，Docker容器实例是只读的。如果它要写入数据，则不会写入容器内部，而是写入卷。如果您正在测试您的服务能否以只读方式运行，则可能需要挂载一个可读写（rw）卷。您可以使用 volume() 固定工厂创建一个与您的测试生命周期绑定的Docker卷。

from pytest_docker_tools import volume

backend_storage = volume()

volume 固定工厂支持传递给 docker-py 卷 create 方法的所有参数。有关所有参数，请参阅这里。

此外，您可以指定一个 initial_content 字典。这允许您使用一组初始状态初始化卷。在以下示例中，我们将预先初始化一个minio服务，其中包含2个桶和1个位于其中一个桶中的对象。

from pytest_docker_tools import container, fetch, volume


minio_image = fetch(repository='minio/minio:latest')

minio_volume = volume(
    initial_content={
        'bucket-1': None,
        'bucket-2/example.txt': b'Test file 1',
    }
)

minio = container(
    image='{minio_image.id}',
    command=['server', '/data'],
    volumes={
        '{minio_volume.name}': {'bind': '/data'},
    },
    environment={
        'MINIO_ACCESS_KEY': 'minio',
        'MINIO_SECRET_KEY': 'minio123',
    },
)

def test_volume_is_seeded(minio):
    files = minio.get_files('/data')
    assert files['data/bucket-2/example.txt'] == b'Test file 1'
    assert files['data/bucket-1'] == None

minio_volume 容器将创建一个空文件夹（bucket-1）和名为 example.txt 的文本文件，位于另一个名为 bucket-2 的单独文件夹中。

该工厂的默认作用域是 function。这意味着每次执行测试时都会创建一个新的卷。该卷将在使用它的测试完成后被移除。

固定装置

docker_client

docker_client 固定装置返回官方docker客户端的一个实例。

def test_container_created(docker_client, fakedns):
    for c in docker_client.containers.list(ignore_removed=True):
        if c.id == fakedns.id:
            # Looks like we managed to start one!
            break
    else:
        assert False, 'Looks like we failed to start a container'

当直接使用 docker_client 时请小心

• 显然，通过API强制创建的资源在测试结束时不会自动移除
• 很容易破坏xdist兼容性
  • 始终使用 ignore_removed 与 docker_client.containers.list() 一起使用 - 没有它则是不稳定的
  • 很容易找到您正在处理的资源的其他实例（在其他工作者中创建的）。请注意这一点！
• 不要采取破坏性行动 - 可能有人在运行带有其他（非测试）容器的机器上运行测试，附带损害很容易发生，应该避免。

这是我们的固定工厂使用的固定装置。这意味着如果您定义了自己的 docker_client 固定装置，则测试将使用该装置。

技巧和窍门

测试构建工件

我们经常发现自己使用一组在测试时构建的容器进行测试（使用 build()），但随后又想使用相同的测试与在CI平台上生成的工件（使用 image()）一起使用。这最终看起来像这样

if not os.environ.get('IMAGE_ID', ''):
    image = build(path='examples/resolver-service/dns')
else:
    image = image(name=os.environ['IMAGE_ID'])

但现在您只需这样做

from pytest_docker_tools import image_or_build

image = image_or_build(
    environ_key='IMAGE_ID',
    path='examples/resolver-service/dns',
)

def test_image(image):
    assert image.attrs['Os'] == 'linux'

开发和CI环境之间的网络差异

您开发和CI环境之间的另一个常见差异可能是，您的测试在CI上运行在Docker中。如果您绑定挂载 docker.sock，则您的测试可能最终会在与您正在测试的容器相同的容器网络上运行，并且无法访问您映射到主机盒子的任何端口。换句话说

• 在您的开发机器上，您的测试可能通过localhost:8000访问您的测试实例（端口映射到主机）
• 在您的CI机器上，它们可能需要通过172.16.0.5:8000访问您的测试实例

容器对象有一个 get_addr 辅助函数，它将根据其所在的环境返回正确的内容。

from pytest_docker_tools import container

apiserver = container(
  image='{apiserver_image.id}',
  ports={'8080/tcp': None}
)

def test_connect_my_service(apiserver):
    ip, port = apiserver.get_addr('8080/tcp')
    # ... connect to ip:port ...

动态作用域

pytest 固定装置装饰器现在允许您指定一个回调来决定固定装置的作用域。这被称为 动态作用域。您可以使用此功能来决定是否每个测试或每个测试运行都应有容器实例。例如

# conftest.py
from pytest_docker_tools import container, fetch

def determine_scope(fixture_name, config):
    if config.getoption("--keep-containers", None):
        return "session"
    return "function"

memcache_image = fetch(repository='memcached:latest')

memcache = container(
    image='{memcache_image.id}',
    scope=determine_scope,
    ports={
        '11211/tcp': None,
    },
)

您的测试可以与之前完全相同

def test_connect_my_service(memcache):
    ip, port = memcache.get_addr('11211/tcp')
    # ... connect to ip:port ...

客户端固定装置

您可能需要为正在测试的服务创建一个API客户端。虽然我们已经在README中做了这件事，但仍有必要指出。您可以定义一个客户端固定值，使其依赖于您的Docker容器，然后只需在测试中引用客户端即可。

# conftest.py

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container

fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '127.0.0.1',
    }
)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}']
)


@pytest.fixture
def apiclient(apiserver):
    port = apiserver.ports['8080/tcp'][0]
    return HTTPConnection(f'localhost:{port}')

然后在测试中引用它

# test_the_test_client.py

import json


def test_api_server(apiclient):
    apiclient.request('GET', '/')
    response = apiclient.getresponse()
    assert response.status == 200
    result = json.loads(response.read())
    assert result['result'] == '127.0.0.1'

在这个例子中，任何使用hpfeeds_client固定值的测试都会获得一个正确配置的客户端，该客户端连接到在Docker容器上运行在临时高端口的代理。当测试完成后，客户端将干净地断开连接，并且Docker容器将被丢弃。

固定值重载

可以使用固定值工厂定义复杂的环境。它们形成一个有向无环图。通过使用固定值重载，可以在单个测试模块的上下文中替换依赖图中该节点，而无需重新定义整个环境。

在不重新定义其依赖项的情况下替换容器固定值

您可以在您的conftest.py中定义固定值

# conftest.py

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container

fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '127.0.0.1',
    }
)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}']
)


@pytest.fixture
def apiclient(apiserver):
    port = apiserver.ports['8080/tcp'][0]
    return HTTPConnection(f'localhost:{port}')

然后您可以在测试模块中重载这些固定值。例如，如果redis有一个魔法复制功能，并且您想测试API的边缘情况，您可以在您的test_smoketest_alternate.py中这样做

# test_smoketest_alternate.py

import json

from pytest_docker_tools import container

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '192.168.192.168',
    }
)

def test_api_server(apiclient):
    apiclient.request('GET', '/')
    response = apiclient.getresponse()
    assert response.status == 200
    result = json.loads(response.read())
    assert result['result'] == '192.168.192.168'

在这里，我们在test_smoketest_alternate中重新定义了fakedns容器。它能够使用我们在conftest.py中定义的fakedns_image固定值。更重要的是，在test_smoketest_alternate.py中，当我们使用核心apiclient固定值时，它实际上会拉入本地定义的fakedns，而不是来自conftest.py的版本！您不需要重新定义任何其他内容。它只是正常工作。

通过固定值注入固定值配置

您也可以从固定值工厂中拉入正常的pytest固定值。这意味着我们可以使用固定值重载并传递配置。在您的conftest.py中

# conftest.py

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container

fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '{example_com_a}',
    }
)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}']
)


@pytest.fixture
def apiclient(apiserver):
    port = apiserver.ports['8080/tcp'][0]
    return HTTPConnection(f'localhost:{port}')


@pytest.fixture
def example_com_a():
    return '127.0.0.1'

现在，当测试使用apiclient固定值时，他们将获得正常配置的fakedns容器。然而，您可以在测试模块中重新定义固定值 - 而其他固定值仍然会尊重它。例如

# test_smoketest_alternate.py

import json

import pytest


@pytest.fixture
def example_com_a():
    return '192.168.192.168'


def test_api_server(apiclient):
    apiclient.request('GET', '/')
    response = apiclient.getresponse()
    assert response.status == 200
    result = json.loads(response.read())
    assert result['result'] == '192.168.192.168'

您的api_server容器（及其redis后端）将按正常方式构建，但在这个测试模块中，它将使用其sqlite后端。

固定值参数化

您可以创建参数化固定值。也许您想运行所有api_server测试以针对您的两个身份验证后端。也许您有一个要测试多个配置的假象。

在您的conftest.py中

# conftest.py

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container

fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns_localhost = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '127.0.0.1',
    }
)

fakedns_alternate = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '192.168.192.168',
    }
)

@pytest.fixture(scope='function', params=['fakedns_localhost', 'fakedns_alternate'])
def fakedns(request):
      return request.getfixturevalue(request.param)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}']
)


@pytest.fixture
def apiclient(apiserver):
    port = apiserver.ports['8080/tcp'][0]
    return HTTPConnection(f'localhost:{port}')

测试与第一个例子相同，但现在它将对2个不同的假配置进行测试。

# test_smoketest.py

import ipaddress
import json


def test_api_server(apiclient):
    apiclient.request('GET', '/')
    response = apiclient.getresponse()
    assert response.status == 200
    result = json.loads(response.read())
    ipaddress.ip_address(result['result'])

此测试将调用两次 - 一次针对内存后端，一次针对sqlite后端。

固定值包装器

您可以使用wrapper_class包装您的固定值。这允许您为测试中的固定值添加辅助方法。在container固定值工厂的情况下，您还可以实现ready()以添加额外的容器就绪性检查。

在前面的测试中，我们创建了一个完整的测试客户端固定值。使用wrapper_class，我们可以将此便利方法挂载在固定值本身上

# test_fixture_wrappers.py

import ipaddress
import json
import random

from http.client import HTTPConnection

import pytest
from pytest_docker_tools import build, container
from pytest_docker_tools import wrappers


class Container(wrappers.Container):

    def ready(self):
        # This is called until it returns True - its a great hook for e.g.
        # waiting until a log message appears or a pid file is created etc
        if super().ready():
            return random.choice([True, False])
        return False

    def client(self):
        port = self.ports['8080/tcp'][0]
        return HTTPConnection(f'localhost:{port}')


fakedns_image = build(
    path='examples/resolver-service/dns',
)

fakedns = container(
    image='{fakedns_image.id}',
    environment={
        'DNS_EXAMPLE_COM__A': '127.0.0.1',
    }
)

apiserver_image = build(
    path='examples/resolver-service/api',
)

apiserver = container(
    image='{apiserver_image.id}',
    ports={
        '8080/tcp': None,
    },
    dns=['{fakedns.ips.primary}'],
    wrapper_class=Container,
)


def test_container_wrapper_class(apiserver):
    client = apiserver.client()
    client.request('GET', '/')
    response = client.getresponse()
    assert response.status == 200
    result = json.loads(response.read())
    ipaddress.ip_address(result['result'])

引用非字符串返回的固定值

您可以通过调用提供的工厂并传递参数来定义资源

from pytest_docker_tools import container

cache = container(
    name='my_cache_container',
    image='1838567e84867e8498695403067879',
    environment={
        'foo': 'bar',
        'mode': 'prod'
        }
    )

如前所述示例所示，通过以字符串模板的方式引用固定值（'{<fixture_name>}'），可以动态地解析工厂参数

from pytest_docker_tools import container, fetch

cache_image = fetch(repository='memcached:latest')

cache = container(
    name='my_cache_container',
    image='{cache_image.id}',
    environment={
        'foo': 'bar',
        'mode': 'prod'
        }
    )

在这个例子中，图像ID将从fetch()提供的图像包装器对象中获取。然而，这仅允许检索类似于字符串的值。例如，不可能使用类似字符串模板的语法动态获取用于envinronment参数的字典对象。这样做只会得到字符串化的字典。

要从测试夹具中获取非字符串返回值，有两种选择。首先，您可以在同一文件中定义另一个测试夹具或导入它。之后，您需要按照以下方式引用它：

import pytest
from pytest_docker_tools import container, fetch, fxtr

cache_image = fetch(repository='memcached:latest')

@pytest.fixture()
def memcached_env():
    yield {'foo': 'bar',
           'mode': 'prod'}

cache = container(
    name='my_cache_container',
    image='{cache_image.id}',
    environment=memcached_env
    )

然而，通常在使用pytest时，不需要首先导入夹具。这就是fxtr辅助函数可以发挥作用的地方。

import pytest
from pytest_docker_tools import container, fetch, fxtr

cache_image = fetch(repository='memcached:latest')

@pytest.fixture()
def memcached_env():
    yield {'foo': 'bar',
           'mode': 'prod'}

cache = container(
    name='my_cache_container',
    image='{cache_image.id}',
    environment=fxtr('memcached_env')
    )

在两个示例中，都向容器函数传递了一个合适的字典对象。对于容器资源，这有助于根据夹具动态设置环境或卷。

可重用容器

默认情况下，pytest-docker-tools的容器夹具工厂将在pytest被调用时创建每个定义的容器，并在会话结束时清理它们。这确保了您的测试环境是干净的，并且测试不会因为容器中留下的某些状态而通过。

有时这种行为可能不是您想要的。当您在迭代开发并多次运行测试时，“测试周期”（修复错误并重新运行测试所需的时间）的速度变得很重要。当使用--reuse-containers命令行参数时，pytest-docker-tools不会自动删除它创建的容器。它将尝试在pytest调用之间重用它们。pytest-docker-tools还会跟踪容器、卷或网络是否已过时（例如，如果您更改了镜像版本），并自动替换它。

注意：当使用--reuse-containers时，您必须为所有pytest-docker-tools夹具设置name属性。如果您不使用--reuse-containers，则不需要设置name属性。

关于使用可重用容器的说明

• 使用--reuse-containers参数创建的资源（容器、网络、卷）将没有终结器，因此范围可能不会像通常那样表现。这取决于测试作者确保没有两个不同的夹具共享相同的名称。
• 当重用资源时，您负责清理它们（例如，数据库、卷数据），因为测试期间写入的数据在完成后不会被删除。
• pytest-docker-tools创建的每个资源容器都会获得以下标签：creator: pytest-docker-tools。如有需要，这可以用于搜索遗留资源。例如，可以通过执行以下命令手动清理容器：docker ps -aq --filter "label=creator=pytest-docker-tools" | xargs docker rm -f

黑客攻击

该项目使用poetry。在为pytest-docker-tools设置开发环境之前，您需要一个工作状态的诗意和Python 3环境。当您这样做时，只需

poetry install

要运行所有linters和测试，请在poetry中运行./scripts/test.sh

poetry run ./scripts/tests.sh

这将运行pyupgrade、isort和black，这将就地修改您的更改以匹配我们使用的代码样式。