forked from testcontainers/testcontainers-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_core_ports.py
More file actions
99 lines (84 loc) · 3.04 KB
/
test_core_ports.py
File metadata and controls
99 lines (84 loc) · 3.04 KB
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
import pytest
from typing import Union, Optional
from testcontainers.core.container import DockerContainer
from docker.errors import APIError
@pytest.mark.parametrize(
"container_port, host_port",
[
("8080", "8080"),
("8125/udp", "8125/udp"),
("8092/udp", "8092/udp"),
("9000/tcp", "9000/tcp"),
("8080", "8080/udp"),
(8080, 8080),
(9000, None),
("9009", None),
("9000", ""),
("9000/udp", ""),
],
)
def test_docker_container_with_bind_ports(container_port: Union[str, int], host_port: Optional[Union[str, int]]):
container = DockerContainer("alpine:latest")
container.with_bind_ports(container_port, host_port)
container.start()
# prepare to inspect container
container_id = container._container.id
client = container._container.client
# assemble expected output to compare to container API
container_port = str(container_port)
host_port = str(host_port or "")
# if the port protocol is not specified, it will default to tcp
if "/" not in container_port:
container_port += "/tcp"
expected = {container_port: [{"HostIp": "", "HostPort": host_port}]}
# compare PortBindings to expected output
assert client.containers.get(container_id).attrs["HostConfig"]["PortBindings"] == expected
container.stop()
@pytest.mark.parametrize(
"container_port, host_port",
[
("0", "8080"),
("8080", "abc"),
(0, 0),
(-1, 8080),
(None, 8080),
],
)
def test_error_docker_container_with_bind_ports(container_port: Union[str, int], host_port: Optional[Union[str, int]]):
with pytest.raises(APIError):
container = DockerContainer("alpine:latest")
container.with_bind_ports(container_port, host_port)
container.start()
@pytest.mark.parametrize(
"ports, expected",
[
(("8125/udp",), {"8125/udp": {}}),
(("8092/udp", "9000/tcp"), {"8092/udp": {}, "9000/tcp": {}}),
(("8080", "8080/udp"), {"8080/tcp": {}, "8080/udp": {}}),
((9000,), {"9000/tcp": {}}),
((8080, 8080), {"8080/tcp": {}}),
(("9001", 9002), {"9001/tcp": {}, "9002/tcp": {}}),
(("9001", 9002, "9003/udp", 9004), {"9001/tcp": {}, "9002/tcp": {}, "9003/udp": {}, "9004/tcp": {}}),
],
)
def test_docker_container_with_exposed_ports(ports: tuple[Union[str, int], ...], expected: dict):
container = DockerContainer("alpine:latest")
container.with_exposed_ports(*ports)
container.start()
container_id = container._container.id
client = container._container.client
assert client.containers.get(container_id).attrs["Config"]["ExposedPorts"] == expected
container.stop()
@pytest.mark.parametrize(
"ports",
[
((9000, None)),
(("", 9000)),
("tcp", ""),
],
)
def test_error_docker_container_with_exposed_ports(ports: tuple[Union[str, int], ...]):
with pytest.raises(APIError):
container = DockerContainer("alpine:latest")
container.with_exposed_ports(*ports)
container.start()