testcontainers-python/core/tests/test_network.py at 860d0f25065e7f2f19d754b01cf8d2e7758e8101 · 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
from http import HTTPStatus
from testcontainers.core.container import DockerContainer
from testcontainers.core.docker_client import DockerClient
from testcontainers.core.labels import LABEL_SESSION_ID
from testcontainers.core.network import Network

import docker.errors
import pytest

NGINX_ALPINE_SLIM_IMAGE = "nginx:1.25.4-alpine-slim"


def test_network_gets_created_and_cleaned_up():
    with Network() as network:
        docker = DockerClient()
        networks_list = docker.client.networks.list(network.name)
        assert networks_list[0].name == network.name
        assert networks_list[0].id == network.id
    assert not docker.client.networks.list(network.name)


def test_network_create_wo_cm():
    network = Network()
    network.create()
    docker = DockerClient()
    networks_list = docker.client.networks.list(network.name)
    assert networks_list[0].name == network.name
    assert networks_list[0].id == network.id

    network.remove()
    assert not docker.client.networks.list(network.name)


def test_network_create_errors():
    network = Network()
    network.create()

    # calling create the second time should raise an error
    with pytest.raises(docker.errors.APIError) as excinfo:
        network.create()

    assert excinfo.value.response.status_code == HTTPStatus.CONFLICT
    excinfo.match(f"network with name {network.name} already exists")
    network.remove()


def test_containers_can_communicate_over_network():
    with (
        Network() as network,
        DockerContainer(NGINX_ALPINE_SLIM_IMAGE)
        .with_name("alpine1")
        .with_network_aliases("alpine1-alias-1", "alpine1-alias-2")
        .with_network(network) as alpine1,
        DockerContainer(NGINX_ALPINE_SLIM_IMAGE)
        .with_name("alpine2")
        .with_network_aliases("alpine2-alias-1", "alpine2-alias-2")
        .with_network(network) as alpine2,
    ):
        assert_can_ping(alpine1, "alpine2")
        assert_can_ping(alpine1, "alpine2-alias-1")
        assert_can_ping(alpine1, "alpine2-alias-2")

        assert_can_ping(alpine2, "alpine1")
        assert_can_ping(alpine2, "alpine1-alias-1")
        assert_can_ping(alpine2, "alpine1-alias-2")

        assert_can_request(alpine1, "alpine2")
        assert_can_request(alpine1, "alpine2-alias-1")
        assert_can_request(alpine1, "alpine2-alias-2")

        assert_can_request(alpine2, "alpine1")
        assert_can_request(alpine2, "alpine1-alias-1")
        assert_can_request(alpine2, "alpine1-alias-2")


def assert_can_ping(container: DockerContainer, remote_name: str):
    status, output = container.exec("ping -c 1 %s" % remote_name)
    assert status == 0
    assert "64 bytes" in str(output)


def assert_can_request(container: DockerContainer, remote_name: str):
    status, output = container.exec(f"wget -qO- http://{remote_name}")
    assert status == 0
    assert "Welcome to nginx!" in output.decode()


def test_network_has_labels():
    network = Network()
    try:
        network.create()
        network = network._docker.client.networks.get(network_id=network.id)
        assert LABEL_SESSION_ID in network.attrs.get("Labels")
    finally:
        network.remove()