forked from testcontainers/testcontainers-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcontainer.py
More file actions
132 lines (109 loc) · 4.63 KB
/
Copy pathcontainer.py
File metadata and controls
132 lines (109 loc) · 4.63 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
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
import blindspin
import crayons
from docker.models.containers import Container
from testcontainers.core.docker_client import DockerClient
from testcontainers.core.exceptions import ContainerStartException
from testcontainers.core.utils import inside_container
class DockerContainer(object):
def __init__(self, image, **kargs):
self.env = {}
self.ports = {}
self.volumes = {}
self.image = image
self._docker = DockerClient()
self._container = None
self._command = None
self._name = None
self._kargs = kargs
def with_env(self, key: str, value: str) -> 'DockerContainer':
self.env[key] = value
return self
def with_bind_ports(self, container: int,
host: int = None) -> 'DockerContainer':
self.ports[container] = host
return self
def with_exposed_ports(self, *ports) -> 'DockerContainer':
for port in list(ports):
self.ports[port] = None
return self
def with_kargs(self, **kargs) -> 'DockerContainer':
self._kargs = kargs
return self
def start(self):
print("")
print("{} {}".format(crayons.yellow("Pulling image"),
crayons.red(self.image)))
with blindspin.spinner():
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._kargs
)
print("")
print("Container started: ",
crayons.yellow(self._container.short_id, bold=True))
return self
def stop(self, force=True, delete_volume=True):
self.get_wrapped_container().remove(force=force, v=delete_volume)
def __enter__(self):
return self.start()
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
def __del__(self):
"""
Try to remove the container in all circumstances
"""
if self._container is not None:
try:
self.stop()
except: # noqa: E722
pass
def get_container_host_ip(self) -> str:
# infer from docker host
host = self.get_docker_client().host()
if not host:
return "localhost"
# check testcontainers itself runs inside docker container
if inside_container():
# If newly spawned container's gateway IP address from the docker
# "bridge" network is equal to detected host address, we should use
# container IP address, otherwise fall back to detected host
# address. Even it's inside container, we need to double check,
# because docker host might be set to docker:dind, usually in CI/CD environment
gateway_ip = self.get_docker_client().gateway_ip(self._container.id)
if gateway_ip == host:
return self.get_docker_client().bridge_ip(self._container.id)
return host
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
def with_command(self, command: str) -> 'DockerContainer':
self._command = command
return self
def with_name(self, name: str) -> 'DockerContainer':
self._name = name
return self
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
def get_wrapped_container(self) -> Container:
return self._container
def get_docker_client(self) -> DockerClient:
return self._docker
def exec(self, command):
if not self._container:
raise ContainerStartException("Container should be started before")
return self.get_wrapped_container().exec_run(command)