<s> import contextlib <EOL> import json <EOL> import shutil <EOL> import socket <EOL> import tarfile <EOL> import tempfile <EOL> import threading <EOL> import pytest <EOL> import six <EOL> from six . moves import BaseHTTPServer <EOL> from six . moves import socketserver <EOL> import docker <EOL> from . . import helpers <EOL> BUSYBOX = helpers . BUSYBOX <EOL> class ListImagesTest ( helpers . BaseTestCase ) : <EOL> def test_images ( self ) : <EOL> res1 = self . client . images ( all = True ) <EOL> self . assertIn ( '<STR_LIT>' , res1 [ <NUM_LIT:0> ] ) <EOL> res10 = res1 [ <NUM_LIT:0> ] <EOL> self . assertIn ( '<STR_LIT>' , res10 ) <EOL> self . assertIn ( '<STR_LIT>' , res10 ) <EOL> distinct = [ ] <EOL> for img in res1 : <EOL> if img [ '<STR_LIT>' ] not in distinct : <EOL> distinct . append ( img [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( distinct ) , self . client . info ( ) [ '<STR_LIT>' ] ) <EOL> def test_images_quiet ( self ) : <EOL> res1 = self . client . images ( quiet = True ) <EOL> self . assertEqual ( type ( res1 [ <NUM_LIT:0> ] ) , six . text_type ) <EOL> class PullImageTest ( helpers . BaseTestCase ) : <EOL> def test_pull ( self ) : <EOL> try : <EOL> self . client . remove_image ( '<STR_LIT>' ) <EOL> except docker . errors . APIError : <EOL> pass <EOL> res = self . client . pull ( '<STR_LIT>' ) <EOL> self . tmp_imgs . append ( '<STR_LIT>' ) <EOL> self . assertEqual ( type ( res ) , six . text_type ) <EOL> self . assertGreaterEqual ( <EOL> len ( self . client . images ( '<STR_LIT>' ) ) , <NUM_LIT:1> <EOL> ) <EOL> img_info = self . client . inspect_image ( '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , img_info ) <EOL> def test_pull_streaming ( self ) : <EOL> try : <EOL> self . client . remove_image ( '<STR_LIT>' ) <EOL> except docker . errors . APIError : <EOL> pass <EOL> stream = self . client . pull ( '<STR_LIT>' , stream = True ) <EOL> self . tmp_imgs . append ( '<STR_LIT>' ) <EOL> for chunk in stream : <EOL> if six . PY3 : <EOL> chunk = chunk . decode ( '<STR_LIT:utf-8>' ) <EOL> json . loads ( chunk ) <EOL> self . assertGreaterEqual ( <EOL> len ( self . client . images ( '<STR_LIT>' ) ) , <NUM_LIT:1> <EOL> ) <EOL> img_info = self . client . inspect_image ( '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , img_info ) <EOL> class CommitTest ( helpers . BaseTestCase ) : <EOL> def test_commit ( self ) : <EOL> container = self . client . create_container ( BUSYBOX , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> id = container [ '<STR_LIT>' ] <EOL> self . client . start ( id ) <EOL> self . tmp_containers . append ( id ) <EOL> res = self . client . commit ( id ) <EOL> self . assertIn ( '<STR_LIT>' , res ) <EOL> img_id = res [ '<STR_LIT>' ] <EOL> self . tmp_imgs . append ( img_id ) <EOL> img = self . client . inspect_image ( img_id ) <EOL> self . assertIn ( '<STR_LIT>' , img ) <EOL> self . assertTrue ( img [ '<STR_LIT>' ] . startswith ( id ) ) <EOL> self . assertIn ( '<STR_LIT>' , img ) <EOL> self . assertIn ( '<STR_LIT>' , img [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( BUSYBOX , img [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> busybox_id = self . client . inspect_image ( BUSYBOX ) [ '<STR_LIT>' ] <EOL> self . assertIn ( '<STR_LIT>' , img ) <EOL> self . assertEqual ( img [ '<STR_LIT>' ] , busybox_id ) <EOL> def test_commit_with_changes ( self ) : <EOL> cid = self . client . create_container ( BUSYBOX , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . tmp_containers . append ( cid ) <EOL> self . client . start ( cid ) <EOL> img_id = self . client . commit ( <EOL> cid , changes = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> self . tmp_imgs . append ( img_id ) <EOL> img = self . client . inspect_image ( img_id ) <EOL> assert '<STR_LIT>' in img <EOL> assert img [ '<STR_LIT>' ] . startswith ( cid [ '<STR_LIT>' ] ) <EOL> assert '<STR_LIT>' in img [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> assert img [ '<STR_LIT>' ] [ '<STR_LIT>' ] == [ '<STR_LIT>' ] <EOL> class RemoveImageTest ( helpers . BaseTestCase ) : <EOL> def test_remove ( self ) : <EOL> container = self . client . create_container ( BUSYBOX , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> id = container [ '<STR_LIT>' ] <EOL> self . client . start ( id ) <EOL> self . tmp_containers . append ( id ) <EOL> res = self . client . commit ( id ) <EOL> self . assertIn ( '<STR_LIT>' , res ) <EOL> img_id = res [ '<STR_LIT>' ] <EOL> self . tmp_imgs . append ( img_id ) <EOL> self . client . remove_image ( img_id , force = True ) <EOL> images = self . client . images ( all = True ) <EOL> res = [ x for x in images if x [ '<STR_LIT>' ] . startswith ( img_id ) ] <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:0> ) <EOL> class ImportImageTest ( helpers . BaseTestCase ) : <EOL> '''<STR_LIT>''' <EOL> TAR_SIZE = <NUM_LIT> * <NUM_LIT> <EOL> def write_dummy_tar_content ( self , n_bytes , tar_fd ) : <EOL> def extend_file ( f , n_bytes ) : <EOL> f . seek ( n_bytes - <NUM_LIT:1> ) <EOL> f . write ( bytearray ( [ <NUM_LIT> ] ) ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> tar = tarfile . TarFile ( fileobj = tar_fd , mode = '<STR_LIT:w>' ) <EOL> with tempfile . NamedTemporaryFile ( ) as f : <EOL> extend_file ( f , n_bytes ) <EOL> tarinfo = tar . gettarinfo ( name = f . name , arcname = '<STR_LIT>' ) <EOL> tar . addfile ( tarinfo , fileobj = f ) <EOL> tar . close ( ) <EOL> @ contextlib . contextmanager <EOL> def dummy_tar_stream ( self , n_bytes ) : <EOL> '''<STR_LIT>''' <EOL> with tempfile . NamedTemporaryFile ( ) as tar_file : <EOL> self . write_dummy_tar_content ( n_bytes , tar_file ) <EOL> tar_file . seek ( <NUM_LIT:0> ) <EOL> yield tar_file <EOL> @ contextlib . contextmanager <EOL> def dummy_tar_file ( self , n_bytes ) : <EOL> '''<STR_LIT>''' <EOL> with tempfile . NamedTemporaryFile ( ) as tar_file : <EOL> self . write_dummy_tar_content ( n_bytes , tar_file ) <EOL> tar_file . seek ( <NUM_LIT:0> ) <EOL> yield tar_file . name <EOL> def test_import_from_bytes ( self ) : <EOL> with self . dummy_tar_stream ( n_bytes = <NUM_LIT> ) as f : <EOL> content = f . read ( ) <EOL> statuses = self . client . import_image_from_data ( <EOL> content , repository = '<STR_LIT>' ) <EOL> result_text = statuses . splitlines ( ) [ - <NUM_LIT:1> ] <EOL> result = json . loads ( result_text ) <EOL> self . assertNotIn ( '<STR_LIT:error>' , result ) <EOL> img_id = result [ '<STR_LIT:status>' ] <EOL> self . tmp_imgs . append ( img_id ) <EOL> def test_import_from_file ( self ) : <EOL> with self . dummy_tar_file ( n_bytes = self . TAR_SIZE ) as tar_filename : <EOL> statuses = self . client . import_image_from_file ( <EOL> tar_filename , repository = '<STR_LIT>' ) <EOL> result_text = statuses . splitlines ( ) [ - <NUM_LIT:1> ] <EOL> result = json . loads ( result_text ) <EOL> self . assertNotIn ( '<STR_LIT:error>' , result ) <EOL> self . assertIn ( '<STR_LIT:status>' , result ) <EOL> img_id = result [ '<STR_LIT:status>' ] <EOL> self . tmp_imgs . append ( img_id ) <EOL> def test_import_from_stream ( self ) : <EOL> with self . dummy_tar_stream ( n_bytes = self . TAR_SIZE ) as tar_stream : <EOL> statuses = self . client . import_image ( <EOL> src = tar_stream , repository = '<STR_LIT>' ) <EOL> result_text = statuses . splitlines ( ) [ - <NUM_LIT:1> ] <EOL> result = json . loads ( result_text ) <EOL> self . assertNotIn ( '<STR_LIT:error>' , result ) <EOL> self . assertIn ( '<STR_LIT:status>' , result ) <EOL> img_id = result [ '<STR_LIT:status>' ] <EOL> self . tmp_imgs . append ( img_id ) <EOL> @ contextlib . contextmanager <EOL> def temporary_http_file_server ( self , stream ) : <EOL> '''<STR_LIT>''' <EOL> class Handler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> def do_GET ( self ) : <EOL> self . send_response ( <NUM_LIT:200> ) <EOL> self . send_header ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> shutil . copyfileobj ( stream , self . wfile ) <EOL> server = socketserver . TCPServer ( ( '<STR_LIT>' , <NUM_LIT:0> ) , Handler ) <EOL> thread = threading . Thread ( target = server . serve_forever ) <EOL> thread . setDaemon ( True ) <EOL> thread . start ( ) <EOL> yield '<STR_LIT>' % ( socket . gethostname ( ) , server . server_address [ <NUM_LIT:1> ] ) <EOL> server . shutdown ( ) <EOL> @ pytest . mark . skipif ( True , reason = "<STR_LIT>" ) <EOL> def test_import_from_url ( self ) : <EOL> tar_size = <NUM_LIT> <EOL> with self . dummy_tar_stream ( n_bytes = tar_size ) as tar_data : <EOL> with self . temporary_http_file_server ( tar_data ) as url : <EOL> statuses = self . client . import_image ( <EOL> src = url , repository = '<STR_LIT>' ) <EOL> result_text = statuses . splitlines ( ) [ - <NUM_LIT:1> ] <EOL> result = json . loads ( result_text ) <EOL> self . assertNotIn ( '<STR_LIT:error>' , result ) <EOL> self . assertIn ( '<STR_LIT:status>' , result ) <EOL> img_id = result [ '<STR_LIT:status>' ] <EOL> self . tmp_imgs . append ( img_id ) </s>
