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
|
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from testinfra.modules.base import Module
class Docker(Module):
"""Test docker containers running on system.
Example:
>>> nginx = host.docker("app_nginx")
>>> nginx.is_running
True
>>> nginx.id
'7e67dc7495ca8f451d346b775890bdc0fb561ecdc97b68fb59ff2f77b509a8fe'
>>> nginx.name
'app_nginx'
"""
def __init__(self, name):
self._name = name
super().__init__()
def inspect(self):
output = self.check_output("docker inspect %s", self._name)
return json.loads(output)[0]
@property
def is_running(self):
return self.inspect()["State"]["Running"]
@property
def is_restarting(self):
return self.inspect()["State"]["Restarting"]
@property
def id(self):
return self.inspect()["Id"]
@property
def name(self):
return self.inspect()["Name"][1:] # get rid of slash in front
@classmethod
def client_version(cls):
"""Docker client version"""
return cls.version("{{.Client.Version}}")
@classmethod
def server_version(cls):
"""Docker server version"""
return cls.version("{{.Server.Version}}")
@classmethod
def version(cls, format=None):
"""Docker version_ with an optional format (Go template).
>>> host.docker.version()
Client: Docker Engine - Community
...
>>> host.docker.version("{{.Client.Context}}"))
default
.. _version: https://docs.docker.com/engine/reference/commandline/version/
"""
cmd = "docker version"
if format:
cmd = f"{cmd} --format '{format}'"
return cls.check_output(cmd)
@classmethod
def get_containers(cls, **filters):
"""Return a list of containers
By default return list of all containers, including non-running
containers.
Filtering can be done using filters keys defined on
https://docs.docker.com/engine/reference/commandline/ps/#filtering
Multiple filters for a given key is handled by giving a list of string
as value.
>>> host.docker.get_containers()
[<docker nginx>, <docker redis>, <docker app>]
# Get all running containers
>>> host.docker.get_containers(status="running")
[<docker app>]
# Get containers named "nginx"
>>> host.docker.get_containers(name="nginx")
[<docker nginx>]
# Get containers named "nginx" or "redis"
>>> host.docker.get_containers(name=["nginx", "redis"])
[<docker nginx>, <docker redis>]
"""
cmd = "docker ps --all --quiet --format '{{.Names}}'"
args = []
for key, value in filters.items():
values = value if isinstance(value, (list, tuple)) else [value]
for v in values:
cmd += " --filter %s=%s"
args += [key, v]
result = []
for docker_id in cls(None).check_output(cmd, *args).splitlines():
result.append(cls(docker_id))
return result
def __repr__(self):
return f"<docker {self._name}>"
|
