docker-mavproxy/conftest.py at master · auturgy/docker-mavproxy · 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
136
137
138
139
140
141
142
143
144
145
146
147
148
# conftest.py for docker image testing

import docker
import pytest


class DockerContainerError(AssertionError):
    pass


class DockerContainerExecError(DockerContainerError):
    pass

class DockerContainerFile():

    def __init__(self, docker_container, path):
        self._docker_container = docker_container
        self._path = path

    @property
    def exists(self):
        if self.is_file:
            return True
        if self.is_directory:
            return True
        return False

    @property
    def is_directory(self):
        try:
            self._docker_container.check_output('test -d {}'.format(self._path))
            return True
        except DockerContainerExecError as e:
            return False

    @property
    def is_executable(self):
        try:
            self._docker_container.check_output('test  -x {}'.format(self._path))
            return True
        except DockerContainerExecError as e:
            return False

    @property
    def is_file(self):
        try:
            self._docker_container.check_output('test -f {}'.format(self._path))
            return True
        except DockerContainerExecError as e:
            return False

class DockerContainerImage():

    def __init__(self, container_image):
        self._container_image = container_image

    @property
    def name(self):
        image_name = self._container_image.tags[0]
        image_name = image_name.split(':')[0]
        return image_name

    @property
    def tag(self):
        tag_name = self._container_image.tags[0]
        tag_name = tag_name.split(':')[1]
        return tag_name

class DockerContainer():

    def __init__(self, container_image):
        self._image = container_image
        self._client = docker.from_env()
        self._container = None

    def check_output(self, command):
        ret =  self._container.exec_run(command)
        if ret.exit_code != 0:
            raise DockerContainerExecError(ret.output)
        return ret.output

    def exists(self, command):
        """
        Implement command exists like testinfra: http://testinfra.readthedocs.io/en/latest/modules.html#testinfra.modules.file.File
        :param command:
        :return:
        """
        if command.startswith('/'):
            command_file = self.file(command)
            if command_file.is_executable:
                return True
        else:
            for path in self.get_env('PATH').split(':'):
                command_path = '{}/{}'.format(path, command)
                command_file = self.file(command_path)
                if command_file.is_executable:
                    return True
        return False

    def file(self, path):
        return DockerContainerFile(self, path)

    @property
    def default_command(self):
        return '/bin/sh -c "sleep 10000"'

    def get_env(self, env_name):
        """
        Get environment variable content
        :param env_name:
        :return:
        """
        return self.check_output('/bin/sh -c "echo ${}"'.format(env_name)).decode().strip()

    @property
    def id(self):
        return self._container.id

    @property
    def name(self):
        return self._container.name

    @property
    def image(self):
        return DockerContainerImage(self._container.image)

    def kill(self):
        self._container.kill()

    def run(self):
        self._container = self._client.containers.run(self._image, self.default_command, detach=True)

    def run_expect(self, command, expected):
        if not isinstance(expected, list):
            expected = [expected]
        expected = [int(e) for e in expected]
        ret =  self._container.exec_run(command)
        exit_code = ret.exit_code
        if ret.exit_code not in expected:
            raise DockerContainerExecError('Command "{}" exit code {}, expected exit code(s): {}'.format(command, exit_code, expected))
        return True

@pytest.fixture(scope='module')
def docker_container(container_image):
    docker_container = DockerContainer(container_image)
    docker_container.run()
    yield docker_container
    docker_container.kill()