<s> import os <EOL> import six <EOL> import unittest <EOL> import yaml <EOL> from maestro import entities , exceptions , loader , lifecycle , maestro , plays <EOL> class EntityTest ( unittest . TestCase ) : <EOL> def test_get_name ( self ) : <EOL> self . assertEqual ( entities . Entity ( '<STR_LIT:foo>' ) . name , '<STR_LIT:foo>' ) <EOL> class ShipTest ( unittest . TestCase ) : <EOL> def test_simple_ship ( self ) : <EOL> ship = entities . Ship ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertEqual ( ship . name , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( ship . ip , '<STR_LIT>' ) <EOL> self . assertEqual ( ship . endpoint , '<STR_LIT>' ) <EOL> def test_ship_endpoint ( self ) : <EOL> ship = entities . Ship ( '<STR_LIT:foo>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( ship . name , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( ship . ip , '<STR_LIT>' ) <EOL> self . assertEqual ( ship . endpoint , '<STR_LIT>' ) <EOL> self . assertTrue ( ship . endpoint in ship . backend . base_url ) <EOL> class ServiceTest ( unittest . TestCase ) : <EOL> def test_get_image ( self ) : <EOL> service = entities . Service ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . assertEqual ( service . image , '<STR_LIT>' ) <EOL> class ContainerTest ( unittest . TestCase ) : <EOL> SERVICE = '<STR_LIT:foo>' <EOL> IMAGE = '<STR_LIT>' <EOL> CONTAINER = '<STR_LIT>' <EOL> SHIP = '<STR_LIT>' <EOL> SHIP_IP = '<STR_LIT>' <EOL> SCHEMA = { '<STR_LIT>' : <NUM_LIT:2> } <EOL> DOCKER_VERSION = '<STR_LIT>' <EOL> def _cntr ( service_name = SERVICE , service_env = None , image = IMAGE , <EOL> ship_name = SHIP , ship_ip = SHIP_IP , <EOL> container_name = CONTAINER , config = None , schema = SCHEMA , <EOL> api_version = DOCKER_VERSION ) : <EOL> service = entities . Service ( service_name , image , schema = schema , <EOL> env = service_env ) <EOL> return entities . Container ( container_name , <EOL> entities . Ship ( ship_name , ship_ip , <EOL> api_version = api_version ) , <EOL> service , config = config , schema = schema ) <EOL> def test_image_propagates_from_service ( self ) : <EOL> container = self . _cntr ( ) <EOL> self . assertEqual ( container . image , container . service . image ) <EOL> def test_get_image_details_basic ( self ) : <EOL> d = self . _cntr ( ) . get_image_details ( ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_get_image_details_notag ( self ) : <EOL> d = self . _cntr ( image = '<STR_LIT>' ) . get_image_details ( ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_get_image_details_custom_registry ( self ) : <EOL> d = self . _cntr ( image = '<STR_LIT>' ) . get_image_details ( ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_get_image_details_custom_port ( self ) : <EOL> d = self . _cntr ( image = '<STR_LIT>' ) . get_image_details ( ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_get_image_details_custom_port_notag ( self ) : <EOL> d = self . _cntr ( image = '<STR_LIT>' ) . get_image_details ( ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( d [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_env_propagates_from_service ( self ) : <EOL> service_env = { '<STR_LIT>' : '<STR_LIT:value>' } <EOL> container_env = { '<STR_LIT>' : '<STR_LIT>' } <EOL> container = self . _cntr ( service_env = service_env , <EOL> config = { '<STR_LIT>' : container_env } ) <EOL> for k , v in service_env . items ( ) : <EOL> self . assertIn ( k , container . env ) <EOL> self . assertEqual ( v , container . env [ k ] ) <EOL> for k , v in container_env . items ( ) : <EOL> self . assertIn ( k , container . env ) <EOL> self . assertEqual ( v , container . env [ k ] ) <EOL> def test_dns_option ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( container . dns , [ '<STR_LIT>' ] ) <EOL> def test_dns_as_list_option ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } ) <EOL> self . assertEqual ( container . dns , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_no_dns_option ( self ) : <EOL> self . assertIsNone ( self . _cntr ( ) . dns ) <EOL> def test_swap_limit_number ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> } } ) <EOL> self . assertEqual ( container . memswap_limit , <NUM_LIT> ) <EOL> def test_swap_limit_string_no_suffix ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( container . memswap_limit , <NUM_LIT> ) <EOL> def test_swap_limit_string_with_suffix ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( container . memswap_limit , <NUM_LIT> * <NUM_LIT> ) <EOL> def test_log_config_default ( self ) : <EOL> self . assertTrue ( "<STR_LIT>" not in self . _cntr ( ) . host_config ) <EOL> def test_log_config_syslog ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertTrue ( "<STR_LIT>" in container . host_config ) <EOL> self . assertEqual ( container . host_config [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { } } ) <EOL> def test_log_config_syslog_with_opts ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } } ) <EOL> self . assertTrue ( "<STR_LIT>" in container . host_config ) <EOL> self . assertEqual ( container . host_config [ '<STR_LIT>' ] , { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } } ) <EOL> def test_log_config_wrong_driver_type ( self ) : <EOL> self . assertRaises ( <EOL> exceptions . InvalidLogConfigurationException , <EOL> lambda : self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def test_log_config_wrong_opt_type ( self ) : <EOL> self . assertRaises ( <EOL> exceptions . InvalidLogConfigurationException , <EOL> lambda : self . _cntr ( <EOL> config = { '<STR_LIT>' : '<STR_LIT>' , "<STR_LIT>" : '<STR_LIT>' } ) ) <EOL> def test_log_config_converts_options_to_string ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:2> <EOL> } } ) <EOL> self . assertTrue ( "<STR_LIT>" in container . host_config ) <EOL> self . assertEqual ( len ( container . host_config [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <NUM_LIT:2> ) <EOL> for value in container . host_config [ '<STR_LIT>' ] [ '<STR_LIT>' ] . values ( ) : <EOL> self . assertTrue ( isinstance ( value , six . string_types ) ) <EOL> def test_restart_policy_default ( self ) : <EOL> self . assertEqual ( self . _cntr ( ) . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_no ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_always ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_onfailure ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_onfailure_with_max_retries ( self ) : <EOL> container = self . _cntr ( <EOL> config = { '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } ) <EOL> def test_restart_policy_wrong_type ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_missing_retries ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' } } ) <EOL> self . assertEqual ( container . restart_policy , <EOL> { '<STR_LIT:Name>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> def test_restart_policy_wrong_name ( self ) : <EOL> self . assertRaises ( <EOL> exceptions . InvalidRestartPolicyConfigurationException , <EOL> lambda : self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def test_volumes_simple_bind ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> self . assertTrue ( '<STR_LIT>' in container . volumes ) <EOL> self . assertEqual ( container . volumes , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> def test_volumes_dict_bind_no_mode ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : { '<STR_LIT:target>' : '<STR_LIT>' } } } ) <EOL> self . assertTrue ( '<STR_LIT>' in container . volumes ) <EOL> self . assertEqual ( container . volumes , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> def test_volumes_ro_bind ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:target>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' <EOL> } } } ) <EOL> self . assertTrue ( '<STR_LIT>' in container . volumes ) <EOL> self . assertEqual ( container . volumes , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> def test_volumes_bind_with_mode ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:target>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' <EOL> } } } ) <EOL> self . assertTrue ( '<STR_LIT>' in container . volumes ) <EOL> self . assertEqual ( container . volumes , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } ) <EOL> def test_volumes_multibind_throws ( self ) : <EOL> self . assertRaises ( <EOL> exceptions . InvalidVolumeConfigurationException , <EOL> lambda : self . _cntr ( config = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } } ) ) <EOL> def test_volumes_invalid_params_throws ( self ) : <EOL> self . assertRaises ( <EOL> exceptions . InvalidVolumeConfigurationException , <EOL> lambda : self . _cntr ( config = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } } ) ) <EOL> def test_volumes_old_schema ( self ) : <EOL> container = self . _cntr ( <EOL> config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } , <EOL> schema = { '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> self . assertEqual ( container . volumes , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : False } } ) <EOL> def test_workdir ( self ) : <EOL> container = self . _cntr ( config = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( container . workdir , '<STR_LIT>' ) <EOL> def test_volume_conflict_container ( self ) : <EOL> self . assertRaisesRegexp ( <EOL> exceptions . InvalidVolumeConfigurationException , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( ContainerTest . CONTAINER ) , <EOL> lambda : self . _cntr ( config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ) ) <EOL> class BaseConfigFileUsingTest ( unittest . TestCase ) : <EOL> def _get_config ( self , name ) : <EOL> path = os . path . join ( os . path . dirname ( __file__ ) , <EOL> '<STR_LIT>' . format ( name ) ) <EOL> return loader . load ( path ) <EOL> class ConductorTest ( BaseConfigFileUsingTest ) : <EOL> def test_duplicate_container_name ( self ) : <EOL> self . assertRaises ( <EOL> yaml . constructor . ConstructorError , <EOL> lambda : self . _get_config ( '<STR_LIT>' ) ) <EOL> def test_empty_registry_list ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> self . assertIsNot ( c . registries , None ) <EOL> self . assertEqual ( c . registries , { } ) <EOL> def test_volumes_parsing ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> instance1 = c . containers [ '<STR_LIT>' ] <EOL> instance2 = c . containers [ '<STR_LIT>' ] <EOL> self . assertEqual ( instance1 . get_volumes ( ) , <EOL> set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( instance2 . get_volumes ( ) , <EOL> set ( [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( instance2 . volumes_from , <EOL> set ( [ instance1 . name ] ) ) <EOL> def test_volume_conflict_volumes_from ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> self . assertRaisesRegexp ( <EOL> exceptions . InvalidVolumeConfigurationException , <EOL> '<STR_LIT>' , <EOL> lambda : maestro . Conductor ( config ) ) <EOL> def test_volumes_from_unknown ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> self . assertRaisesRegexp ( <EOL> exceptions . InvalidVolumeConfigurationException , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> lambda : maestro . Conductor ( config ) ) <EOL> def test_env_name ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> self . assertEqual ( c . env_name , '<STR_LIT:test>' ) <EOL> foo1 = c . containers [ '<STR_LIT>' ] <EOL> self . assertEqual ( foo1 . env [ '<STR_LIT>' ] , '<STR_LIT:test>' ) <EOL> def test_missing_env_name ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> self . assertRaisesRegexp ( <EOL> exceptions . EnvironmentConfigurationException , <EOL> '<STR_LIT>' , <EOL> lambda : maestro . Conductor ( config ) ) <EOL> class ConfigTest ( BaseConfigFileUsingTest ) : <EOL> def test_yaml_parsing_test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT:bar>' <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , config [ '<STR_LIT:foo>' ] ) <EOL> def test_ship_parsing ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> self . assertEqual ( c . ships [ '<STR_LIT>' ] . ip , '<STR_LIT>' ) <EOL> self . assertEqual ( c . ships [ '<STR_LIT>' ] . endpoint , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in c . ships [ '<STR_LIT>' ] . backend . base_url ) <EOL> self . assertEqual ( c . ships [ '<STR_LIT>' ] . ip , '<STR_LIT>' ) <EOL> self . assertEqual ( c . ships [ '<STR_LIT>' ] . endpoint , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in c . ships [ '<STR_LIT>' ] . backend . base_url ) <EOL> class LifecycleHelperTest ( unittest . TestCase ) : <EOL> def _get_container ( self ) : <EOL> ship = entities . Ship ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> service = entities . Service ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> return entities . Container ( <EOL> '<STR_LIT>' , ship , service , <EOL> config = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:data>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:foo>' : '<STR_LIT:bar>' , '<STR_LIT>' : <NUM_LIT> } } ) <EOL> def test_script_env_all_strings ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> env = c . _create_env ( ) <EOL> self . assertEqual ( type ( env [ '<STR_LIT>' ] ) , str ) <EOL> self . assertEqual ( env [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_parse_checker_exec ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertIsNot ( c , None ) <EOL> self . assertIsInstance ( c , lifecycle . ScriptExecutor ) <EOL> self . assertEqual ( c . command , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_parse_checker_tcp ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:port>' : '<STR_LIT>' } ) <EOL> self . assertIsInstance ( c , lifecycle . TCPPortPinger ) <EOL> self . assertEqual ( c . host , container . ship . ip ) <EOL> self . assertEqual ( c . port , <NUM_LIT> ) <EOL> self . assertEqual ( c . attempts , <EOL> lifecycle . TCPPortPinger . DEFAULT_MAX_ATTEMPTS ) <EOL> def test_parse_checker_tcp_with_max_wait ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:port>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:2> } ) <EOL> self . assertIsInstance ( c , lifecycle . TCPPortPinger ) <EOL> self . assertEqual ( c . host , container . ship . ip ) <EOL> self . assertEqual ( c . port , <NUM_LIT> ) <EOL> self . assertEqual ( c . attempts , <NUM_LIT:2> ) <EOL> def test_parse_checker_tcp_unknown_port ( self ) : <EOL> container = self . _get_container ( ) <EOL> self . assertRaises ( <EOL> exceptions . InvalidLifecycleCheckConfigurationException , <EOL> lifecycle . LifecycleHelperFactory . from_config , <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:port>' : '<STR_LIT>' } ) <EOL> def test_parse_checker_tcp_invalid_port ( self ) : <EOL> container = self . _get_container ( ) <EOL> self . assertRaises ( <EOL> exceptions . InvalidLifecycleCheckConfigurationException , <EOL> lifecycle . LifecycleHelperFactory . from_config , <EOL> container , { '<STR_LIT:type>' : '<STR_LIT>' , '<STR_LIT:port>' : '<STR_LIT:data>' } ) <EOL> def test_parse_unknown_checker_type ( self ) : <EOL> self . assertRaises ( <EOL> KeyError , <EOL> lifecycle . LifecycleHelperFactory . from_config , <EOL> self . _get_container ( ) , { '<STR_LIT:type>' : '<STR_LIT>' } ) <EOL> def test_parse_checker_http_defaults ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT:http>' , '<STR_LIT:port>' : '<STR_LIT>' } ) <EOL> self . assertIsInstance ( c , lifecycle . HttpRequestLifecycle ) <EOL> self . assertEqual ( c . host , container . ship . ip ) <EOL> self . assertEqual ( c . port , <NUM_LIT> ) <EOL> self . assertEqual ( c . max_wait , <EOL> lifecycle . HttpRequestLifecycle . DEFAULT_MAX_WAIT ) <EOL> self . assertEqual ( c . match_regex , None ) <EOL> self . assertEqual ( c . path , '<STR_LIT:/>' ) <EOL> self . assertEqual ( c . scheme , '<STR_LIT:http>' ) <EOL> self . assertEqual ( c . method , '<STR_LIT>' ) <EOL> self . assertEqual ( c . requests_options , { } ) <EOL> self . assertTrue ( c . _test_response ) <EOL> def test_parse_checker_http_explicits ( self ) : <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , <EOL> { <EOL> '<STR_LIT:type>' : '<STR_LIT:http>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : { '<STR_LIT>' : False } <EOL> } ) <EOL> self . assertIsInstance ( c , lifecycle . HttpRequestLifecycle ) <EOL> self . assertEqual ( c . host , container . ship . ip ) <EOL> self . assertEqual ( c . port , <NUM_LIT> ) <EOL> self . assertEqual ( c . max_wait , <NUM_LIT:2> ) <EOL> self . assertFalse ( c . match_regex . search ( '<STR_LIT>' ) ) <EOL> self . assertTrue ( c . match_regex . search ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( c . path , '<STR_LIT>' ) <EOL> self . assertEqual ( c . scheme , '<STR_LIT>' ) <EOL> self . assertEqual ( c . method , '<STR_LIT>' ) <EOL> self . assertEqual ( c . requests_options , { '<STR_LIT>' : False } ) <EOL> def test_parse_checker_http_status_match ( self ) : <EOL> class FakeEmptyResponse ( object ) : <EOL> status_code = <NUM_LIT:200> <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT:http>' , '<STR_LIT:port>' : '<STR_LIT>' } ) <EOL> self . assertTrue ( c . _test_response ( FakeEmptyResponse ( ) ) ) <EOL> def test_parse_checker_http_status_fail ( self ) : <EOL> class FakeEmptyResponse ( object ) : <EOL> status_code = <NUM_LIT> <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , { '<STR_LIT:type>' : '<STR_LIT:http>' , '<STR_LIT:port>' : '<STR_LIT>' } ) <EOL> self . assertFalse ( c . _test_response ( FakeEmptyResponse ( ) ) ) <EOL> def test_parse_checker_http_regex_match ( self ) : <EOL> class FakeEmptyResponse ( object ) : <EOL> text = '<STR_LIT>' <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , <EOL> { '<STR_LIT:type>' : '<STR_LIT:http>' , '<STR_LIT:port>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertTrue ( c . _test_response ( FakeEmptyResponse ( ) ) ) <EOL> def test_parse_checker_http_regex_fail ( self ) : <EOL> class FakeEmptyResponse ( object ) : <EOL> text = '<STR_LIT>' <EOL> container = self . _get_container ( ) <EOL> c = lifecycle . LifecycleHelperFactory . from_config ( <EOL> container , <EOL> { '<STR_LIT:type>' : '<STR_LIT:http>' , '<STR_LIT:port>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertFalse ( c . _test_response ( FakeEmptyResponse ( ) ) ) <EOL> class LoginTaskTest ( BaseConfigFileUsingTest ) : <EOL> def test_find_registry_for_container_by_name ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> container = c . containers [ '<STR_LIT>' ] <EOL> registry = plays . tasks . LoginTask . registry_for_container ( <EOL> container , c . registries ) <EOL> self . assertEqual ( registry , c . registries [ '<STR_LIT>' ] ) <EOL> def test_find_registry_for_container_by_fqdn ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> container = c . containers [ '<STR_LIT>' ] <EOL> registry = plays . tasks . LoginTask . registry_for_container ( <EOL> container , c . registries ) <EOL> self . assertEqual ( registry , c . registries [ '<STR_LIT>' ] ) <EOL> def test_find_registry_for_container_not_found ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> container = c . containers [ '<STR_LIT>' ] <EOL> registry = plays . tasks . LoginTask . registry_for_container ( <EOL> container , c . registries ) <EOL> self . assertEqual ( registry , None ) <EOL> def test_find_registry_by_image_name ( self ) : <EOL> config = self . _get_config ( '<STR_LIT>' ) <EOL> c = maestro . Conductor ( config ) <EOL> container = c . containers [ '<STR_LIT>' ] <EOL> registry = plays . tasks . LoginTask . registry_for_container ( <EOL> container , c . registries ) <EOL> self . assertEqual ( registry , c . registries [ '<STR_LIT>' ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
