testcontainers-python/core/tests/test_reusable_containers.py at e84fe02f68f45df5e3556248276f0b150a45b2b4 · testcontainers/testcontainers-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
from time import sleep

from docker.models.containers import Container

from testcontainers.core.config import testcontainers_config
from testcontainers.core.container import DockerContainer
from testcontainers.core.docker_client import DockerClient
from testcontainers.core.waiting_utils import wait_for_logs
from testcontainers.core.container import Reaper


def test_docker_container_reuse_default():
    # Make sure Ryuk cleanup is not active from previous test runs
    Reaper.delete_instance()

    container = DockerContainer("hello-world").start()
    wait_for_logs(container, "Hello from Docker!")

    assert container._reuse == False
    assert testcontainers_config.tc_properties_testcontainers_reuse_enable() == False
    assert Reaper._socket is not None

    container.stop()
    containers = DockerClient().client.containers.list(all=True)
    assert container._container is not None
    assert container._container.id not in [container.id for container in containers]


def test_docker_container_with_reuse_reuse_disabled(caplog):
    # Make sure Ryuk cleanup is not active from previous test runs
    Reaper.delete_instance()

    container = DockerContainer("hello-world").with_reuse().start()
    wait_for_logs(container, "Hello from Docker!")

    assert container._reuse == True
    assert testcontainers_config.tc_properties_testcontainers_reuse_enable() == False
    assert ("Reuse was requested (`with_reuse`) but the environment does not support the ") in caplog.text
    assert Reaper._socket is not None

    container.stop()
    containers = DockerClient().client.containers.list(all=True)
    assert container._container is not None
    assert container._container.id not in [container.id for container in containers]


def test_docker_container_without_reuse_reuse_enabled(monkeypatch):
    # Make sure Ryuk cleanup is not active from previous test runs
    Reaper.delete_instance()

    tc_properties_mock = testcontainers_config.tc_properties | {"testcontainers.reuse.enable": "true"}
    monkeypatch.setattr(testcontainers_config, "tc_properties", tc_properties_mock)

    container = DockerContainer("hello-world").start()
    wait_for_logs(container, "Hello from Docker!")

    assert container._reuse == False
    assert testcontainers_config.tc_properties_testcontainers_reuse_enable() == True
    assert Reaper._socket is not None

    container.stop()
    containers = DockerClient().client.containers.list(all=True)
    assert container._container is not None
    assert container._container.id not in [container.id for container in containers]


def test_docker_container_with_reuse_reuse_enabled(monkeypatch):
    # Make sure Ryuk cleanup is not active from previous test runs
    Reaper.delete_instance()

    tc_properties_mock = testcontainers_config.tc_properties | {"testcontainers.reuse.enable": "true"}
    monkeypatch.setattr(testcontainers_config, "tc_properties", tc_properties_mock)

    container = DockerContainer("hello-world").with_reuse().start()
    wait_for_logs(container, "Hello from Docker!")

    assert Reaper._socket is None

    containers = DockerClient().client.containers.list(all=True)
    assert container._container is not None
    assert container._container.id in [container.id for container in containers]

    # Cleanup after keeping container alive (with_reuse)
    container.stop()


def test_docker_container_with_reuse_reuse_enabled_same_id(monkeypatch):
    # Make sure Ryuk cleanup is not active from previous test runs
    Reaper.delete_instance()

    tc_properties_mock = testcontainers_config.tc_properties | {"testcontainers.reuse.enable": "true"}
    monkeypatch.setattr(testcontainers_config, "tc_properties", tc_properties_mock)

    container_1 = DockerContainer("hello-world").with_reuse().start()
    assert container_1._container is not None
    id_1 = container_1._container.id
    container_2 = DockerContainer("hello-world").with_reuse().start()
    assert container_2._container is not None
    id_2 = container_2._container.id
    assert Reaper._socket is None
    assert id_1 == id_2
    # Cleanup after keeping container alive (with_reuse)
    container_1.stop()
    # container_2.stop() is not needed since it is the same as container_1


def test_docker_container_labels_hash_default():
    # w/out reuse
    with DockerContainer("hello-world") as container:
        assert container._container is not None
        assert "hash" not in container._container.labels.keys()


def test_docker_container_labels_hash(monkeypatch):
    tc_properties_mock = testcontainers_config.tc_properties | {"testcontainers.reuse.enable": "true"}
    monkeypatch.setattr(testcontainers_config, "tc_properties", tc_properties_mock)
    expected_hash = "1296172062"
    with DockerContainer("hello-world").with_reuse() as container:
        assert container._container is not None
        assert container._container.labels["hash"] == expected_hash


def test_docker_client_find_container_by_hash_not_existing():
    with DockerContainer("hello-world"):
        assert DockerClient().find_container_by_hash("foo") == None


def test_docker_client_find_container_by_hash_existing(monkeypatch):
    tc_properties_mock = testcontainers_config.tc_properties | {"testcontainers.reuse.enable": "true"}
    monkeypatch.setattr(testcontainers_config, "tc_properties", tc_properties_mock)
    with DockerContainer("hello-world").with_reuse() as container:
        assert container._container is not None
        hash_ = container._container.labels["hash"]
        found_container = DockerClient().find_container_by_hash(hash_)
        assert isinstance(found_container, Container)