1import contextlib
2import json
3import shutil
4import socket
5import tarfile
6import tempfile
7import threading
8
9import pytest
10import six
11from six.moves import BaseHTTPServer
12from six.moves import socketserver
13
14
15import docker
16
17from .. import helpers
18
19BUSYBOX = helpers.BUSYBOX
20
21
22class ListImagesTest(helpers.BaseTestCase):
23    def test_images(self):
24        res1 = self.client.images(all=True)
25        self.assertIn('Id', res1[0])
26        res10 = res1[0]
27        self.assertIn('Created', res10)
28        self.assertIn('RepoTags', res10)
29        distinct = []
30        for img in res1:
31            if img['Id'] not in distinct:
32                distinct.append(img['Id'])
33        self.assertEqual(len(distinct), self.client.info()['Images'])
34
35    def test_images_quiet(self):
36        res1 = self.client.images(quiet=True)
37        self.assertEqual(type(res1[0]), six.text_type)
38
39
40class PullImageTest(helpers.BaseTestCase):
41    def test_pull(self):
42        try:
43            self.client.remove_image('hello-world')
44        except docker.errors.APIError:
45            pass
46        res = self.client.pull('hello-world')
47        self.tmp_imgs.append('hello-world')
48        self.assertEqual(type(res), six.text_type)
49        self.assertGreaterEqual(
50            len(self.client.images('hello-world')), 1
51        )
52        img_info = self.client.inspect_image('hello-world')
53        self.assertIn('Id', img_info)
54
55    def test_pull_streaming(self):
56        try:
57            self.client.remove_image('hello-world')
58        except docker.errors.APIError:
59            pass
60        stream = self.client.pull('hello-world', stream=True)
61        self.tmp_imgs.append('hello-world')
62        for chunk in stream:
63            if six.PY3:
64                chunk = chunk.decode('utf-8')
65            json.loads(chunk)  # ensure chunk is a single, valid JSON blob
66        self.assertGreaterEqual(
67            len(self.client.images('hello-world')), 1
68        )
69        img_info = self.client.inspect_image('hello-world')
70        self.assertIn('Id', img_info)
71
72
73class CommitTest(helpers.BaseTestCase):
74    def test_commit(self):
75        container = self.client.create_container(BUSYBOX, ['touch', '/test'])
76        id = container['Id']
77        self.client.start(id)
78        self.tmp_containers.append(id)
79        res = self.client.commit(id)
80        self.assertIn('Id', res)
81        img_id = res['Id']
82        self.tmp_imgs.append(img_id)
83        img = self.client.inspect_image(img_id)
84        self.assertIn('Container', img)
85        self.assertTrue(img['Container'].startswith(id))
86        self.assertIn('ContainerConfig', img)
87        self.assertIn('Image', img['ContainerConfig'])
88        self.assertEqual(BUSYBOX, img['ContainerConfig']['Image'])
89        busybox_id = self.client.inspect_image(BUSYBOX)['Id']
90        self.assertIn('Parent', img)
91        self.assertEqual(img['Parent'], busybox_id)
92
93
94class RemoveImageTest(helpers.BaseTestCase):
95    def test_remove(self):
96        container = self.client.create_container(BUSYBOX, ['touch', '/test'])
97        id = container['Id']
98        self.client.start(id)
99        self.tmp_containers.append(id)
100        res = self.client.commit(id)
101        self.assertIn('Id', res)
102        img_id = res['Id']
103        self.tmp_imgs.append(img_id)
104        self.client.remove_image(img_id, force=True)
105        images = self.client.images(all=True)
106        res = [x for x in images if x['Id'].startswith(img_id)]
107        self.assertEqual(len(res), 0)
108
109
110class ImportImageTest(helpers.BaseTestCase):
111    '''Base class for `docker import` test cases.'''
112
113    TAR_SIZE = 512 * 1024
114
115    def write_dummy_tar_content(self, n_bytes, tar_fd):
116        def extend_file(f, n_bytes):
117            f.seek(n_bytes - 1)
118            f.write(bytearray([65]))
119            f.seek(0)
120
121        tar = tarfile.TarFile(fileobj=tar_fd, mode='w')
122
123        with tempfile.NamedTemporaryFile() as f:
124            extend_file(f, n_bytes)
125            tarinfo = tar.gettarinfo(name=f.name, arcname='testdata')
126            tar.addfile(tarinfo, fileobj=f)
127
128        tar.close()
129
130    @contextlib.contextmanager
131    def dummy_tar_stream(self, n_bytes):
132        '''Yields a stream that is valid tar data of size n_bytes.'''
133        with tempfile.NamedTemporaryFile() as tar_file:
134            self.write_dummy_tar_content(n_bytes, tar_file)
135            tar_file.seek(0)
136            yield tar_file
137
138    @contextlib.contextmanager
139    def dummy_tar_file(self, n_bytes):
140        '''Yields the name of a valid tar file of size n_bytes.'''
141        with tempfile.NamedTemporaryFile() as tar_file:
142            self.write_dummy_tar_content(n_bytes, tar_file)
143            tar_file.seek(0)
144            yield tar_file.name
145
146    def test_import_from_bytes(self):
147        with self.dummy_tar_stream(n_bytes=500) as f:
148            content = f.read()
149
150        # The generic import_image() function cannot import in-memory bytes
151        # data that happens to be represented as a string type, because
152        # import_image() will try to use it as a filename and usually then
153        # trigger an exception. So we test the import_image_from_data()
154        # function instead.
155        statuses = self.client.import_image_from_data(
156            content, repository='test/import-from-bytes')
157
158        result_text = statuses.splitlines()[-1]
159        result = json.loads(result_text)
160
161        self.assertNotIn('error', result)
162
163        img_id = result['status']
164        self.tmp_imgs.append(img_id)
165
166    def test_import_from_file(self):
167        with self.dummy_tar_file(n_bytes=self.TAR_SIZE) as tar_filename:
168            # statuses = self.client.import_image(
169            #     src=tar_filename, repository='test/import-from-file')
170            statuses = self.client.import_image_from_file(
171                tar_filename, repository='test/import-from-file')
172
173        result_text = statuses.splitlines()[-1]
174        result = json.loads(result_text)
175
176        self.assertNotIn('error', result)
177
178        self.assertIn('status', result)
179        img_id = result['status']
180        self.tmp_imgs.append(img_id)
181
182    def test_import_from_stream(self):
183        with self.dummy_tar_stream(n_bytes=self.TAR_SIZE) as tar_stream:
184            statuses = self.client.import_image(
185                src=tar_stream, repository='test/import-from-stream')
186            # statuses = self.client.import_image_from_stream(
187            #     tar_stream, repository='test/import-from-stream')
188        result_text = statuses.splitlines()[-1]
189        result = json.loads(result_text)
190
191        self.assertNotIn('error', result)
192
193        self.assertIn('status', result)
194        img_id = result['status']
195        self.tmp_imgs.append(img_id)
196
197    @contextlib.contextmanager
198    def temporary_http_file_server(self, stream):
199        '''Serve data from an IO stream over HTTP.'''
200
201        class Handler(BaseHTTPServer.BaseHTTPRequestHandler):
202            def do_GET(self):
203                self.send_response(200)
204                self.send_header('Content-Type', 'application/x-tar')
205                self.end_headers()
206                shutil.copyfileobj(stream, self.wfile)
207
208        server = socketserver.TCPServer(('', 0), Handler)
209        thread = threading.Thread(target=server.serve_forever)
210        thread.setDaemon(True)
211        thread.start()
212
213        yield 'http://%s:%s' % (socket.gethostname(), server.server_address[1])
214
215        server.shutdown()
216
217    @pytest.mark.skipif(True, reason="Doesn't work inside a container - FIXME")
218    def test_import_from_url(self):
219        # The crappy test HTTP server doesn't handle large files well, so use
220        # a small file.
221        tar_size = 10240
222
223        with self.dummy_tar_stream(n_bytes=tar_size) as tar_data:
224            with self.temporary_http_file_server(tar_data) as url:
225                statuses = self.client.import_image(
226                    src=url, repository='test/import-from-url')
227
228        result_text = statuses.splitlines()[-1]
229        result = json.loads(result_text)
230
231        self.assertNotIn('error', result)
232
233        self.assertIn('status', result)
234        img_id = result['status']
235        self.tmp_imgs.append(img_id)
236