<s> """<STR_LIT>""" <EOL> from __future__ import print_function , unicode_literals <EOL> from tests . fixtures import temp_image_name <EOL> from atomic_reactor . core import DockerTasker <EOL> from atomic_reactor . util import ImageName , clone_git_repo <EOL> from tests . constants import LOCALHOST_REGISTRY , INPUT_IMAGE , DOCKERFILE_GIT , MOCK , COMMAND <EOL> from tests . util import requires_internet <EOL> import docker , docker . errors <EOL> from flexmock import flexmock <EOL> import pytest <EOL> if MOCK : <EOL> from tests . docker_mock import mock_docker <EOL> input_image_name = ImageName . parse ( INPUT_IMAGE ) <EOL> def setup_module ( module ) : <EOL> if MOCK : <EOL> return <EOL> d = docker . Client ( ) <EOL> try : <EOL> d . inspect_image ( INPUT_IMAGE ) <EOL> setattr ( module , '<STR_LIT>' , True ) <EOL> except docker . errors . APIError : <EOL> _ = [ x for x in d . pull ( INPUT_IMAGE , stream = True ) ] <EOL> setattr ( module , '<STR_LIT>' , False ) <EOL> def teardown_module ( module ) : <EOL> if MOCK : <EOL> return <EOL> if not getattr ( module , '<STR_LIT>' , False ) : <EOL> d = docker . Client ( ) <EOL> d . remove_image ( INPUT_IMAGE ) <EOL> def test_run ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> container_id = t . run ( input_image_name , command = "<STR_LIT:id>" ) <EOL> try : <EOL> t . wait ( container_id ) <EOL> finally : <EOL> t . remove_container ( container_id ) <EOL> def test_run_invalid_command ( ) : <EOL> if MOCK : <EOL> mock_docker ( should_raise_error = { '<STR_LIT:start>' : [ ] } ) <EOL> t = DockerTasker ( ) <EOL> try : <EOL> with pytest . raises ( docker . errors . APIError ) : <EOL> t . run ( input_image_name , command = COMMAND ) <EOL> finally : <EOL> containers = t . d . containers ( all = True ) <EOL> container_id = [ c for c in containers if c [ "<STR_LIT>" ] == COMMAND ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> t . remove_container ( container_id ) <EOL> def test_image_exists ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> assert t . image_exists ( input_image_name ) is True <EOL> def test_image_doesnt_exist ( ) : <EOL> image = "<STR_LIT>" <EOL> if MOCK : <EOL> mock_docker ( should_raise_error = { '<STR_LIT>' : [ image ] } ) <EOL> t = DockerTasker ( ) <EOL> assert t . image_exists ( image ) is False <EOL> def test_logs ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> container_id = t . run ( input_image_name , command = "<STR_LIT:id>" ) <EOL> try : <EOL> t . wait ( container_id ) <EOL> output = t . logs ( container_id , stderr = True , stream = False ) <EOL> assert "<STR_LIT:\n>" . join ( output ) . startswith ( "<STR_LIT>" ) <EOL> finally : <EOL> t . remove_container ( container_id ) <EOL> def test_remove_container ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> container_id = t . run ( input_image_name , command = "<STR_LIT:id>" ) <EOL> try : <EOL> t . wait ( container_id ) <EOL> finally : <EOL> t . remove_container ( container_id ) <EOL> def test_remove_image ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( inspect_should_fail = True ) <EOL> t = DockerTasker ( ) <EOL> container_id = t . run ( input_image_name , command = "<STR_LIT:id>" ) <EOL> t . wait ( container_id ) <EOL> image_id = t . commit_container ( container_id , image = temp_image_name ) <EOL> try : <EOL> t . remove_container ( container_id ) <EOL> finally : <EOL> t . remove_image ( image_id ) <EOL> assert not t . image_exists ( temp_image_name ) <EOL> def test_commit_container ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> container_id = t . run ( INPUT_IMAGE , command = "<STR_LIT:id>" ) <EOL> t . wait ( container_id ) <EOL> image_id = t . commit_container ( container_id , message = "<STR_LIT>" , image = temp_image_name ) <EOL> try : <EOL> assert t . image_exists ( image_id ) <EOL> finally : <EOL> t . remove_container ( container_id ) <EOL> t . remove_image ( image_id ) <EOL> def test_inspect_image ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> inspect_data = t . inspect_image ( input_image_name ) <EOL> assert isinstance ( inspect_data , dict ) <EOL> def test_tag_image ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> temp_image_name . registry = "<STR_LIT>" <EOL> temp_image_name . tag = "<STR_LIT:1>" <EOL> img = t . tag_image ( INPUT_IMAGE , temp_image_name ) <EOL> try : <EOL> assert t . image_exists ( temp_image_name ) <EOL> assert img == temp_image_name . to_str ( ) <EOL> finally : <EOL> t . remove_image ( temp_image_name ) <EOL> def test_tag_image_same_name ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> temp_image_name . registry = "<STR_LIT>" <EOL> temp_image_name . tag = "<STR_LIT:1>" <EOL> flexmock ( t . d ) . should_receive ( '<STR_LIT>' ) . never ( ) <EOL> img = t . tag_image ( temp_image_name , temp_image_name . copy ( ) ) <EOL> @ pytest . mark . parametrize ( ( '<STR_LIT>' , ) , [ <EOL> ( True , ) , <EOL> ( False , ) , <EOL> ] ) <EOL> def test_push_image ( temp_image_name , should_fail ) : <EOL> if MOCK : <EOL> mock_docker ( push_should_fail = should_fail ) <EOL> t = DockerTasker ( ) <EOL> temp_image_name . registry = LOCALHOST_REGISTRY <EOL> temp_image_name . tag = "<STR_LIT:1>" <EOL> t . tag_image ( INPUT_IMAGE , temp_image_name ) <EOL> if should_fail : <EOL> with pytest . raises ( RuntimeError ) as exc : <EOL> output = t . push_image ( temp_image_name , insecure = True ) <EOL> assert "<STR_LIT>" in str ( exc ) <EOL> else : <EOL> output = t . push_image ( temp_image_name , insecure = True ) <EOL> assert output is not None <EOL> t . remove_image ( temp_image_name ) <EOL> def test_tag_and_push ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> temp_image_name . registry = LOCALHOST_REGISTRY <EOL> temp_image_name . tag = "<STR_LIT:1>" <EOL> output = t . tag_and_push_image ( INPUT_IMAGE , temp_image_name , insecure = True ) <EOL> assert output is not None <EOL> assert t . image_exists ( temp_image_name ) <EOL> t . remove_image ( temp_image_name ) <EOL> def test_pull_image ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> local_img = input_image_name <EOL> remote_img = local_img . copy ( ) <EOL> remote_img . registry = LOCALHOST_REGISTRY <EOL> t . tag_and_push_image ( local_img , remote_img , insecure = True ) <EOL> got_image = t . pull_image ( remote_img , insecure = True ) <EOL> assert remote_img . to_str ( ) == got_image <EOL> assert len ( t . last_logs ) > <NUM_LIT:0> <EOL> t . remove_image ( remote_img ) <EOL> def test_get_image_info_by_id_nonexistent ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . get_image_info_by_image_id ( "<STR_LIT>" ) <EOL> assert response is None <EOL> def test_get_image_info_by_id ( ) : <EOL> if MOCK : <EOL> mock_docker ( provided_image_repotags = input_image_name . to_str ( ) ) <EOL> t = DockerTasker ( ) <EOL> image_id = t . get_image_info_by_image_name ( input_image_name ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> response = t . get_image_info_by_image_id ( image_id ) <EOL> assert isinstance ( response , dict ) <EOL> def test_get_image_info_by_name_tag_in_name ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . get_image_info_by_image_name ( input_image_name ) <EOL> assert len ( response ) == <NUM_LIT:1> <EOL> def test_get_image_info_by_name_tag_in_name_nonexisten ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . get_image_info_by_image_name ( temp_image_name ) <EOL> assert len ( response ) == <NUM_LIT:0> <EOL> @ requires_internet <EOL> def test_build_image_from_path ( tmpdir , temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> tmpdir_path = str ( tmpdir . realpath ( ) ) <EOL> clone_git_repo ( DOCKERFILE_GIT , tmpdir_path ) <EOL> df = tmpdir . join ( "<STR_LIT>" ) <EOL> assert df . check ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . build_image_from_path ( tmpdir_path , temp_image_name , use_cache = True ) <EOL> list ( response ) <EOL> assert response is not None <EOL> assert t . image_exists ( temp_image_name ) <EOL> t . remove_image ( temp_image_name ) <EOL> @ requires_internet <EOL> def test_build_image_from_git ( temp_image_name ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . build_image_from_git ( DOCKERFILE_GIT , temp_image_name , use_cache = True ) <EOL> list ( response ) <EOL> assert response is not None <EOL> assert t . image_exists ( temp_image_name ) <EOL> t . remove_image ( temp_image_name ) <EOL> def test_get_info ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . get_info ( ) <EOL> assert isinstance ( response , dict ) <EOL> def test_get_version ( ) : <EOL> if MOCK : <EOL> mock_docker ( ) <EOL> t = DockerTasker ( ) <EOL> response = t . get_info ( ) <EOL> assert isinstance ( response , dict ) <EOL> @ pytest . mark . parametrize ( ( '<STR_LIT>' , '<STR_LIT>' ) , [ <EOL> ( None , <NUM_LIT> ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ] ) <EOL> def test_timeout ( timeout , expected_timeout ) : <EOL> ( flexmock ( docker . Client ) <EOL> . should_receive ( '<STR_LIT>' ) <EOL> . with_args ( version = str , timeout = expected_timeout ) ) <EOL> kwargs = { } <EOL> if timeout is not None : <EOL> kwargs [ '<STR_LIT>' ] = timeout <EOL> t = DockerTasker ( ** kwargs ) </s>
