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