#NEWLINE# @ unittest . skipIf ( ( mock is None ) , 'mock package not present' ) #NEWLINE# @ mock . patch ( 'airflow.utils.file.mkdtemp' ) #NEWLINE# @ mock . patch ( 'airflow.operators.docker_operator.Client' ) #NEWLINE# def test_execute( self , client_class_mock , mkdtemp_mock ) : #NEWLINE# #INDENT# host_config = mock . Mock ( ) #NEWLINE# mkdtemp_mock . return_value = '/mkdtemp' #NEWLINE# client_mock = mock . Mock ( spec = Client ) #NEWLINE# client_mock . create_container . return_value = { #NEWLINE# #INDENT# 'Id' : 'some_id' , #NEWLINE# #UNINDENT# } #NEWLINE# client_mock . create_host_config . return_value = host_config #NEWLINE# client_mock . images . return_value = [ ] #NEWLINE# client_mock . logs . return_value = [ 'container log' ] #NEWLINE# client_mock . pull . return_value = [ '{"status":"pull log"}' ] #NEWLINE# client_mock . wait . return_value = 0 #NEWLINE# client_class_mock . return_value = self #NEWLINE# operator = DockerOperator ( api_version = '1.19' , command = 'env' , environment = { #NEWLINE# #INDENT# 'UNIT' : 'TEST' , #NEWLINE# #UNINDENT# } , image = 'ubuntu:latest' , network_mode = 'bridge' , owner = 'unittest' , task_id = 'unittest' , volumes = [ '/host/path:/container/path' ] ) #NEWLINE# operator . execute ( None ) #NEWLINE# client_class_mock . assert_called_with ( base_url = 'unix://var/run/docker.sock' , tls = None , version = '1.19' ) #NEWLINE# client_mock . create_container . assert_called_with ( command = 'env' , cpu_shares = 1024 , environment = { #NEWLINE# #INDENT# 'AIRFLOW_TMP_DIR' : '/tmp/airflow' , #NEWLINE# 'UNIT' : 'TEST' , #NEWLINE# #UNINDENT# } , host_config = host_config , image = 'ubuntu:latest' , mem_limit = None , user = None ) #NEWLINE# client_mock . create_host_config . assert_called_with ( binds = [ '/host/path:/container/path' , '/mkdtemp:/tmp/airflow' ] , network_mode = 'bridge' ) #NEWLINE# client_mock . images . assert_called_with ( name = 'ubuntu:latest' ) #NEWLINE# client_mock . logs . assert_called_with ( container = 'some_id' , stream = True ) #NEWLINE# client_mock . pull . assert_called_with ( 'ubuntu:latest' , stream = True ) #NEWLINE# client_mock . wait . assert_called_with ( 'some_id' )