<s> import collections <EOL> import copy <EOL> import sys <EOL> import uuid <EOL> import eventlet <EOL> import mock <EOL> from neutron_lib import constants as const <EOL> from neutron_lib import exceptions <EOL> from oslo_config import cfg <EOL> import oslo_messaging <EOL> import testtools <EOL> from neutron . agent . common import config <EOL> from neutron . agent . dhcp import agent as dhcp_agent <EOL> from neutron . agent . dhcp import config as dhcp_config <EOL> from neutron . agent import dhcp_agent as entry <EOL> from neutron . agent . linux import dhcp <EOL> from neutron . agent . linux import interface <EOL> from neutron . common import config as common_config <EOL> from neutron . common import constants as n_const <EOL> from neutron . common import utils <EOL> from neutron import context <EOL> from neutron . tests import base <EOL> HOSTNAME = '<STR_LIT>' <EOL> dev_man = dhcp . DeviceManager <EOL> rpc_api = dhcp_agent . DhcpPluginApi <EOL> DEVICE_MANAGER = '<STR_LIT>' % ( dev_man . __module__ , dev_man . __name__ ) <EOL> DHCP_PLUGIN = '<STR_LIT>' % ( rpc_api . __module__ , rpc_api . __name__ ) <EOL> fake_tenant_id = '<STR_LIT>' <EOL> fake_subnet1_allocation_pools = dhcp . DictModel ( dict ( id = '<STR_LIT>' , start = '<STR_LIT>' , <EOL> end = '<STR_LIT>' ) ) <EOL> fake_subnet1 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , enable_dhcp = True , name = '<STR_LIT>' , <EOL> tenant_id = fake_tenant_id , <EOL> gateway_ip = '<STR_LIT>' , host_routes = [ ] , <EOL> dns_nameservers = [ ] , ip_version = <NUM_LIT:4> , <EOL> ipv6_ra_mode = None , ipv6_address_mode = None , <EOL> allocation_pools = fake_subnet1_allocation_pools ) ) <EOL> fake_subnet2_allocation_pools = dhcp . DictModel ( dict ( id = '<STR_LIT>' , start = '<STR_LIT>' , <EOL> end = '<STR_LIT>' ) ) <EOL> fake_subnet2 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , enable_dhcp = False , name = '<STR_LIT>' , <EOL> tenant_id = fake_tenant_id , gateway_ip = '<STR_LIT>' , <EOL> host_routes = [ ] , dns_nameservers = [ ] , ip_version = <NUM_LIT:4> , <EOL> allocation_pools = fake_subnet2_allocation_pools ) ) <EOL> fake_subnet3 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , enable_dhcp = True ) ) <EOL> fake_ipv6_subnet = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , enable_dhcp = True , <EOL> tenant_id = fake_tenant_id , <EOL> gateway_ip = '<STR_LIT>' , ip_version = <NUM_LIT:6> , <EOL> ipv6_ra_mode = '<STR_LIT>' , ipv6_address_mode = None ) ) <EOL> fake_meta_subnet = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> cidr = '<STR_LIT>' , <EOL> gateway_ip = '<STR_LIT>' , <EOL> enable_dhcp = True ) ) <EOL> fake_fixed_ip1 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , subnet_id = fake_subnet1 . id , <EOL> ip_address = '<STR_LIT>' ) ) <EOL> fake_fixed_ip2 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , subnet_id = fake_subnet1 . id , <EOL> ip_address = '<STR_LIT>' ) ) <EOL> fake_fixed_ipv6 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , subnet_id = fake_ipv6_subnet . id , <EOL> ip_address = '<STR_LIT>' ) ) <EOL> fake_meta_fixed_ip = dhcp . DictModel ( dict ( id = '<STR_LIT>' , subnet = fake_meta_subnet , <EOL> ip_address = '<STR_LIT>' ) ) <EOL> fake_allocation_pool_subnet1 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , start = '<STR_LIT>' , <EOL> end = '<STR_LIT>' ) ) <EOL> fake_port1 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> device_id = '<STR_LIT>' , <EOL> device_owner = '<STR_LIT>' , <EOL> allocation_pools = fake_subnet1_allocation_pools , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_fixed_ip1 ] ) ) <EOL> fake_dhcp_port = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> device_id = '<STR_LIT>' , <EOL> device_owner = const . DEVICE_OWNER_DHCP , <EOL> allocation_pools = fake_subnet1_allocation_pools , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_fixed_ip2 ] ) ) <EOL> fake_port2 = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> device_id = '<STR_LIT>' , <EOL> device_owner = '<STR_LIT>' , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_fixed_ip2 ] ) ) <EOL> fake_ipv6_port = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> device_owner = '<STR_LIT>' , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_fixed_ipv6 ] ) ) <EOL> fake_meta_port = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> device_owner = const . DEVICE_OWNER_ROUTER_INTF , <EOL> device_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_meta_fixed_ip ] ) ) <EOL> fake_meta_dvr_port = dhcp . DictModel ( fake_meta_port . copy ( ) ) <EOL> fake_meta_dvr_port . device_owner = const . DEVICE_OWNER_DVR_INTERFACE <EOL> fake_dist_port = dhcp . DictModel ( dict ( id = '<STR_LIT>' , <EOL> mac_address = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , <EOL> device_owner = const . DEVICE_OWNER_DVR_INTERFACE , <EOL> device_id = '<STR_LIT>' , <EOL> fixed_ips = [ fake_meta_fixed_ip ] ) ) <EOL> FAKE_NETWORK_UUID = '<STR_LIT>' <EOL> FAKE_NETWORK_DHCP_NS = "<STR_LIT>" % FAKE_NETWORK_UUID <EOL> fake_network = dhcp . NetModel ( dict ( id = FAKE_NETWORK_UUID , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 , fake_subnet2 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> fake_network_ipv6 = dhcp . NetModel ( dict ( <EOL> id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_ipv6_subnet ] , <EOL> ports = [ fake_ipv6_port ] ) ) <EOL> fake_network_ipv6_ipv4 = dhcp . NetModel ( dict ( <EOL> id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_ipv6_subnet , fake_subnet1 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> isolated_network = dhcp . NetModel ( <EOL> dict ( <EOL> id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> nonisolated_dist_network = dhcp . NetModel ( <EOL> dict ( <EOL> id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ fake_port1 , fake_port2 ] ) ) <EOL> empty_network = dhcp . NetModel ( <EOL> dict ( <EOL> id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ ] ) ) <EOL> fake_meta_network = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_meta_subnet ] , <EOL> ports = [ fake_meta_port ] ) ) <EOL> fake_meta_dvr_network = dhcp . NetModel ( fake_meta_network . copy ( ) ) <EOL> fake_meta_dvr_network . ports = [ fake_meta_dvr_port ] <EOL> fake_dist_network = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ fake_meta_subnet ] , <EOL> ports = [ fake_meta_port , fake_dist_port ] ) ) <EOL> fake_down_network = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = False , <EOL> subnets = [ ] , <EOL> ports = [ ] ) ) <EOL> class TestDhcpAgent ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDhcpAgent , self ) . setUp ( ) <EOL> entry . register_options ( cfg . CONF ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . driver_cls_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . driver = mock . Mock ( name = '<STR_LIT>' ) <EOL> self . driver . existing_dhcp_networks . return_value = [ ] <EOL> self . driver_cls = self . driver_cls_p . start ( ) <EOL> self . driver_cls . return_value = self . driver <EOL> self . mock_makedirs_p = mock . patch ( "<STR_LIT>" ) <EOL> self . mock_makedirs = self . mock_makedirs_p . start ( ) <EOL> self . mock_ip_wrapper_p = mock . patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . mock_ip_wrapper = self . mock_ip_wrapper_p . start ( ) <EOL> def test_init_host ( self ) : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp , '<STR_LIT>' ) as sync_state : <EOL> dhcp . init_host ( ) <EOL> sync_state . assert_called_once_with ( ) <EOL> def test_dhcp_agent_manager ( self ) : <EOL> state_rpc_str = '<STR_LIT>' <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> with mock . patch . object ( dhcp_agent . DhcpAgentWithStateReport , <EOL> '<STR_LIT>' , <EOL> autospec = True ) as mock_sync_state : <EOL> with mock . patch . object ( dhcp_agent . DhcpAgentWithStateReport , <EOL> '<STR_LIT>' , <EOL> autospec = True ) as mock_periodic_resync : <EOL> with mock . patch ( state_rpc_str ) as state_rpc : <EOL> with mock . patch . object ( sys , '<STR_LIT>' ) as sys_argv : <EOL> sys_argv . return_value = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> base . etcdir ( '<STR_LIT>' ) ] <EOL> cfg . CONF . register_opts ( dhcp_config . DHCP_AGENT_OPTS ) <EOL> config . register_interface_driver_opts_helper ( cfg . CONF ) <EOL> config . register_agent_state_opts_helper ( cfg . CONF ) <EOL> cfg . CONF . register_opts ( interface . OPTS ) <EOL> common_config . init ( sys . argv [ <NUM_LIT:1> : ] ) <EOL> agent_mgr = dhcp_agent . DhcpAgentWithStateReport ( <EOL> '<STR_LIT>' ) <EOL> eventlet . greenthread . sleep ( <NUM_LIT:1> ) <EOL> agent_mgr . after_start ( ) <EOL> mock_sync_state . assert_called_once_with ( agent_mgr ) <EOL> mock_periodic_resync . assert_called_once_with ( agent_mgr ) <EOL> state_rpc . assert_has_calls ( <EOL> [ mock . call ( mock . ANY ) , <EOL> mock . call ( ) . report_state ( mock . ANY , mock . ANY , <EOL> mock . ANY ) ] ) <EOL> def test_dhcp_agent_main_agent_manager ( self ) : <EOL> logging_str = '<STR_LIT>' <EOL> launcher_str = '<STR_LIT>' <EOL> with mock . patch ( logging_str ) : <EOL> with mock . patch . object ( sys , '<STR_LIT>' ) as sys_argv : <EOL> with mock . patch ( launcher_str ) as launcher : <EOL> sys_argv . return_value = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> base . etcdir ( '<STR_LIT>' ) ] <EOL> entry . main ( ) <EOL> launcher . assert_has_calls ( <EOL> [ mock . call ( cfg . CONF ) , <EOL> mock . call ( ) . launch_service ( mock . ANY ) , <EOL> mock . call ( ) . wait ( ) ] ) <EOL> def test_run_completes_single_pass ( self ) : <EOL> with mock . patch ( DEVICE_MANAGER ) : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> attrs_to_mock = dict ( <EOL> [ ( a , mock . DEFAULT ) for a in <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> with mock . patch . multiple ( dhcp , ** attrs_to_mock ) as mocks : <EOL> dhcp . run ( ) <EOL> mocks [ '<STR_LIT>' ] . assert_called_once_with ( ) <EOL> mocks [ '<STR_LIT>' ] . assert_called_once_with ( ) <EOL> def test_call_driver ( self ) : <EOL> network = mock . Mock ( ) <EOL> network . id = '<STR_LIT:1>' <EOL> dhcp = dhcp_agent . DhcpAgent ( cfg . CONF ) <EOL> self . assertTrue ( dhcp . call_driver ( '<STR_LIT:foo>' , network ) ) <EOL> self . driver . assert_called_once_with ( cfg . CONF , <EOL> mock . ANY , <EOL> mock . ANY , <EOL> mock . ANY , <EOL> mock . ANY ) <EOL> def _test_call_driver_failure ( self , exc = None , <EOL> trace_level = '<STR_LIT>' , expected_sync = True ) : <EOL> network = mock . Mock ( ) <EOL> network . id = '<STR_LIT:1>' <EOL> self . driver . return_value . foo . side_effect = exc or Exception <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp , <EOL> '<STR_LIT>' ) as schedule_resync : <EOL> self . assertIsNone ( dhcp . call_driver ( '<STR_LIT:foo>' , network ) ) <EOL> self . driver . assert_called_once_with ( cfg . CONF , <EOL> mock . ANY , <EOL> mock . ANY , <EOL> mock . ANY , <EOL> mock . ANY ) <EOL> self . assertEqual ( expected_sync , schedule_resync . called ) <EOL> def test_call_driver_ip_address_generation_failure ( self ) : <EOL> error = oslo_messaging . RemoteError ( <EOL> exc_type = '<STR_LIT>' ) <EOL> self . _test_call_driver_failure ( exc = error , expected_sync = False ) <EOL> def test_call_driver_failure ( self ) : <EOL> self . _test_call_driver_failure ( ) <EOL> def test_call_driver_remote_error_net_not_found ( self ) : <EOL> self . _test_call_driver_failure ( <EOL> exc = oslo_messaging . RemoteError ( exc_type = '<STR_LIT>' ) , <EOL> trace_level = '<STR_LIT>' ) <EOL> def test_call_driver_network_not_found ( self ) : <EOL> self . _test_call_driver_failure ( <EOL> exc = exceptions . NetworkNotFound ( net_id = '<STR_LIT:1>' ) , <EOL> trace_level = '<STR_LIT>' ) <EOL> def test_call_driver_conflict ( self ) : <EOL> self . _test_call_driver_failure ( <EOL> exc = exceptions . Conflict ( ) , <EOL> trace_level = '<STR_LIT>' , <EOL> expected_sync = False ) <EOL> def _test_sync_state_helper ( self , known_net_ids , active_net_ids ) : <EOL> active_networks = set ( mock . Mock ( id = netid ) for netid in active_net_ids ) <EOL> with mock . patch ( DHCP_PLUGIN ) as plug : <EOL> mock_plugin = mock . Mock ( ) <EOL> mock_plugin . get_active_networks_info . return_value = active_networks <EOL> plug . return_value = mock_plugin <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> attrs_to_mock = dict ( [ ( a , mock . DEFAULT ) <EOL> for a in [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ] ) <EOL> with mock . patch . multiple ( dhcp , ** attrs_to_mock ) as mocks : <EOL> mocks [ '<STR_LIT>' ] . get_network_ids . return_value = known_net_ids <EOL> dhcp . sync_state ( ) <EOL> diff = set ( known_net_ids ) - set ( active_net_ids ) <EOL> exp_disable = [ mock . call ( net_id ) for net_id in diff ] <EOL> mocks [ '<STR_LIT>' ] . assert_has_calls ( [ mock . call . get_network_ids ( ) ] ) <EOL> mocks [ '<STR_LIT>' ] . assert_has_calls ( exp_disable ) <EOL> def test_sync_state_initial ( self ) : <EOL> self . _test_sync_state_helper ( [ ] , [ '<STR_LIT:a>' ] ) <EOL> def test_sync_state_same ( self ) : <EOL> self . _test_sync_state_helper ( [ '<STR_LIT:a>' ] , [ '<STR_LIT:a>' ] ) <EOL> def test_sync_state_disabled_net ( self ) : <EOL> self . _test_sync_state_helper ( [ '<STR_LIT:b>' ] , [ '<STR_LIT:a>' ] ) <EOL> def test_sync_state_waitall ( self ) : <EOL> with mock . patch . object ( dhcp_agent . eventlet . GreenPool , '<STR_LIT>' ) as w : <EOL> active_net_ids = [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' , '<STR_LIT:4>' , '<STR_LIT:5>' ] <EOL> known_net_ids = [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' , '<STR_LIT:4>' , '<STR_LIT:5>' ] <EOL> self . _test_sync_state_helper ( known_net_ids , active_net_ids ) <EOL> w . assert_called_once_with ( ) <EOL> def test_sync_state_for_all_networks_plugin_error ( self ) : <EOL> with mock . patch ( DHCP_PLUGIN ) as plug : <EOL> mock_plugin = mock . Mock ( ) <EOL> mock_plugin . get_active_networks_info . side_effect = Exception <EOL> plug . return_value = mock_plugin <EOL> with mock . patch . object ( dhcp_agent . LOG , '<STR_LIT>' ) as log : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp , <EOL> '<STR_LIT>' ) as schedule_resync : <EOL> dhcp . sync_state ( ) <EOL> self . assertTrue ( log . called ) <EOL> self . assertTrue ( schedule_resync . called ) <EOL> def test_sync_state_for_one_network_plugin_error ( self ) : <EOL> with mock . patch ( DHCP_PLUGIN ) as plug : <EOL> mock_plugin = mock . Mock ( ) <EOL> exc = Exception ( ) <EOL> mock_plugin . get_active_networks_info . side_effect = exc <EOL> plug . return_value = mock_plugin <EOL> with mock . patch . object ( dhcp_agent . LOG , '<STR_LIT>' ) as log : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp , <EOL> '<STR_LIT>' ) as schedule_resync : <EOL> dhcp . sync_state ( [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( log . called ) <EOL> schedule_resync . assert_called_with ( exc , '<STR_LIT>' ) <EOL> def test_periodic_resync ( self ) : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp_agent . eventlet , '<STR_LIT>' ) as spawn : <EOL> dhcp . periodic_resync ( ) <EOL> spawn . assert_called_once_with ( dhcp . _periodic_resync_helper ) <EOL> def test_report_state_revival_logic ( self ) : <EOL> dhcp = dhcp_agent . DhcpAgentWithStateReport ( HOSTNAME ) <EOL> with mock . patch . object ( dhcp . state_rpc , <EOL> '<STR_LIT>' ) as report_state , mock . patch . object ( dhcp , "<STR_LIT>" ) : <EOL> report_state . return_value = n_const . AGENT_ALIVE <EOL> dhcp . _report_state ( ) <EOL> self . assertEqual ( { } , dhcp . needs_resync_reasons ) <EOL> report_state . return_value = n_const . AGENT_REVIVED <EOL> dhcp . _report_state ( ) <EOL> self . assertEqual ( dhcp . needs_resync_reasons [ None ] , <EOL> [ '<STR_LIT>' ] ) <EOL> def test_periodic_resync_helper ( self ) : <EOL> with mock . patch . object ( dhcp_agent . eventlet , '<STR_LIT>' ) as sleep : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> resync_reasons = collections . OrderedDict ( <EOL> ( ( '<STR_LIT:a>' , '<STR_LIT>' ) , ( '<STR_LIT:b>' , '<STR_LIT>' ) ) ) <EOL> dhcp . needs_resync_reasons = resync_reasons <EOL> with mock . patch . object ( dhcp , '<STR_LIT>' ) as sync_state : <EOL> sync_state . side_effect = RuntimeError <EOL> with testtools . ExpectedException ( RuntimeError ) : <EOL> dhcp . _periodic_resync_helper ( ) <EOL> sync_state . assert_called_once_with ( resync_reasons . keys ( ) ) <EOL> sleep . assert_called_once_with ( dhcp . conf . resync_interval ) <EOL> self . assertEqual ( len ( dhcp . needs_resync_reasons ) , <NUM_LIT:0> ) <EOL> def test_populate_cache_on_start_without_active_networks_support ( self ) : <EOL> self . driver . existing_dhcp_networks . side_effect = NotImplementedError <EOL> with mock . patch . object ( dhcp_agent . LOG , '<STR_LIT>' ) as log : <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> self . driver . existing_dhcp_networks . assert_called_once_with ( <EOL> dhcp . conf , <EOL> ) <EOL> self . assertFalse ( dhcp . cache . get_network_ids ( ) ) <EOL> self . assertTrue ( log . called ) <EOL> def test_populate_cache_on_start ( self ) : <EOL> networks = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . driver . existing_dhcp_networks . return_value = networks <EOL> dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> self . driver . existing_dhcp_networks . assert_called_once_with ( <EOL> dhcp . conf , <EOL> ) <EOL> self . assertEqual ( set ( networks ) , set ( dhcp . cache . get_network_ids ( ) ) ) <EOL> def test_none_interface_driver ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , None ) <EOL> self . assertRaises ( SystemExit , dhcp . DeviceManager , <EOL> cfg . CONF , None ) <EOL> def test_nonexistent_interface_driver ( self ) : <EOL> self . driver_cls_p . stop ( ) <EOL> self . addCleanup ( self . driver_cls_p . start ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertRaises ( SystemExit , dhcp . DeviceManager , <EOL> cfg . CONF , None ) <EOL> class TestLogArgs ( base . BaseTestCase ) : <EOL> def test_log_args_without_log_dir_and_file ( self ) : <EOL> conf_dict = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> conf = dhcp . DictModel ( conf_dict ) <EOL> expected_args = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> args = config . get_log_args ( conf , '<STR_LIT>' ) <EOL> self . assertEqual ( expected_args , args ) <EOL> def test_log_args_without_log_file ( self ) : <EOL> conf_dict = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> conf = dhcp . DictModel ( conf_dict ) <EOL> expected_args = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> args = config . get_log_args ( conf , '<STR_LIT>' ) <EOL> self . assertEqual ( expected_args , args ) <EOL> def test_log_args_with_log_dir_and_file ( self ) : <EOL> conf_dict = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> conf = dhcp . DictModel ( conf_dict ) <EOL> expected_args = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> args = config . get_log_args ( conf , '<STR_LIT>' ) <EOL> self . assertEqual ( expected_args , args ) <EOL> def test_log_args_without_log_dir ( self ) : <EOL> conf_dict = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> conf = dhcp . DictModel ( conf_dict ) <EOL> expected_args = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> args = config . get_log_args ( conf , '<STR_LIT>' ) <EOL> self . assertEqual ( expected_args , args ) <EOL> def test_log_args_with_filelog_and_syslog ( self ) : <EOL> conf_dict = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> conf = dhcp . DictModel ( conf_dict ) <EOL> expected_args = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> args = config . get_log_args ( conf , '<STR_LIT>' ) <EOL> self . assertEqual ( expected_args , args ) <EOL> class TestDhcpAgentEventHandler ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDhcpAgentEventHandler , self ) . setUp ( ) <EOL> config . register_interface_driver_opts_helper ( cfg . CONF ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> entry . register_options ( cfg . CONF ) <EOL> self . plugin_p = mock . patch ( DHCP_PLUGIN ) <EOL> plugin_cls = self . plugin_p . start ( ) <EOL> self . plugin = mock . Mock ( ) <EOL> plugin_cls . return_value = self . plugin <EOL> self . cache_p = mock . patch ( '<STR_LIT>' ) <EOL> cache_cls = self . cache_p . start ( ) <EOL> self . cache = mock . Mock ( ) <EOL> cache_cls . return_value = self . cache <EOL> self . mock_makedirs_p = mock . patch ( "<STR_LIT>" ) <EOL> self . mock_makedirs = self . mock_makedirs_p . start ( ) <EOL> self . mock_init_p = mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . mock_init = self . mock_init_p . start ( ) <EOL> self . dhcp = dhcp_agent . DhcpAgent ( HOSTNAME ) <EOL> self . call_driver_p = mock . patch . object ( self . dhcp , '<STR_LIT>' ) <EOL> self . call_driver = self . call_driver_p . start ( ) <EOL> self . schedule_resync_p = mock . patch . object ( self . dhcp , <EOL> '<STR_LIT>' ) <EOL> self . schedule_resync = self . schedule_resync_p . start ( ) <EOL> self . external_process_p = mock . patch ( <EOL> '<STR_LIT>' <EOL> ) <EOL> self . external_process = self . external_process_p . start ( ) <EOL> def _process_manager_constructor_call ( self , ns = FAKE_NETWORK_DHCP_NS ) : <EOL> return mock . call ( conf = cfg . CONF , <EOL> uuid = FAKE_NETWORK_UUID , <EOL> namespace = ns , <EOL> default_cmd_callback = mock . ANY ) <EOL> def _enable_dhcp_helper ( self , network , enable_isolated_metadata = False , <EOL> is_isolated_network = False ) : <EOL> self . dhcp . _process_monitor = mock . Mock ( ) <EOL> if enable_isolated_metadata : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . plugin . get_network_info . return_value = network <EOL> self . dhcp . enable_dhcp_helper ( network . id ) <EOL> self . plugin . assert_has_calls ( [ <EOL> mock . call . get_network_info ( network . id ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , network ) <EOL> self . cache . assert_has_calls ( [ mock . call . put ( network ) ] ) <EOL> if is_isolated_network and enable_isolated_metadata : <EOL> self . external_process . assert_has_calls ( [ <EOL> self . _process_manager_constructor_call ( ) , <EOL> mock . call ( ) . enable ( ) ] ) <EOL> elif not enable_isolated_metadata : <EOL> self . external_process . assert_has_calls ( [ <EOL> self . _process_manager_constructor_call ( ns = None ) , <EOL> mock . call ( ) . disable ( ) ] ) <EOL> else : <EOL> self . assertFalse ( self . external_process . call_count ) <EOL> def test_enable_dhcp_helper_enable_metadata_isolated_network ( self ) : <EOL> self . _enable_dhcp_helper ( isolated_network , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = True ) <EOL> def test_enable_dhcp_helper_enable_metadata_no_gateway ( self ) : <EOL> isolated_network_no_gateway = copy . deepcopy ( isolated_network ) <EOL> isolated_network_no_gateway . subnets [ <NUM_LIT:0> ] . gateway_ip = None <EOL> self . _enable_dhcp_helper ( isolated_network_no_gateway , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = True ) <EOL> def test_enable_dhcp_helper_enable_metadata_nonisolated_network ( self ) : <EOL> nonisolated_network = copy . deepcopy ( isolated_network ) <EOL> nonisolated_network . ports [ <NUM_LIT:0> ] . device_owner = ( <EOL> const . DEVICE_OWNER_ROUTER_INTF ) <EOL> nonisolated_network . ports [ <NUM_LIT:0> ] . fixed_ips [ <NUM_LIT:0> ] . ip_address = '<STR_LIT>' <EOL> self . _enable_dhcp_helper ( nonisolated_network , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = False ) <EOL> def test_enable_dhcp_helper_enable_metadata_nonisolated_dist_network ( self ) : <EOL> nonisolated_dist_network . ports [ <NUM_LIT:0> ] . device_owner = ( <EOL> const . DEVICE_OWNER_ROUTER_INTF ) <EOL> nonisolated_dist_network . ports [ <NUM_LIT:0> ] . fixed_ips [ <NUM_LIT:0> ] . ip_address = '<STR_LIT>' <EOL> nonisolated_dist_network . ports [ <NUM_LIT:1> ] . device_owner = ( <EOL> const . DEVICE_OWNER_DVR_INTERFACE ) <EOL> nonisolated_dist_network . ports [ <NUM_LIT:1> ] . fixed_ips [ <NUM_LIT:0> ] . ip_address = '<STR_LIT>' <EOL> self . _enable_dhcp_helper ( nonisolated_dist_network , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = False ) <EOL> def test_enable_dhcp_helper_enable_metadata_empty_network ( self ) : <EOL> self . _enable_dhcp_helper ( empty_network , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = True ) <EOL> def test_enable_dhcp_helper_enable_metadata_ipv6_ipv4_network ( self ) : <EOL> self . _enable_dhcp_helper ( fake_network_ipv6_ipv4 , <EOL> enable_isolated_metadata = True , <EOL> is_isolated_network = True ) <EOL> def test_enable_dhcp_helper_driver_failure_ipv6_ipv4_network ( self ) : <EOL> self . plugin . get_network_info . return_value = fake_network_ipv6_ipv4 <EOL> self . call_driver . return_value = False <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> with mock . patch . object ( <EOL> self . dhcp , '<STR_LIT>' ) as enable_metadata : <EOL> self . dhcp . enable_dhcp_helper ( fake_network_ipv6_ipv4 . id ) <EOL> self . plugin . assert_has_calls ( <EOL> [ mock . call . get_network_info ( fake_network_ipv6_ipv4 . id ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , <EOL> fake_network_ipv6_ipv4 ) <EOL> self . assertFalse ( self . cache . called ) <EOL> self . assertFalse ( enable_metadata . called ) <EOL> self . assertFalse ( self . external_process . called ) <EOL> def test_enable_dhcp_helper ( self ) : <EOL> self . _enable_dhcp_helper ( fake_network ) <EOL> def test_enable_dhcp_helper_ipv6_network ( self ) : <EOL> self . _enable_dhcp_helper ( fake_network_ipv6 ) <EOL> def test_enable_dhcp_helper_down_network ( self ) : <EOL> self . plugin . get_network_info . return_value = fake_down_network <EOL> self . dhcp . enable_dhcp_helper ( fake_down_network . id ) <EOL> self . plugin . assert_has_calls ( <EOL> [ mock . call . get_network_info ( fake_down_network . id ) ] ) <EOL> self . assertFalse ( self . call_driver . called ) <EOL> self . assertFalse ( self . cache . called ) <EOL> self . assertFalse ( self . external_process . called ) <EOL> def test_enable_dhcp_helper_network_none ( self ) : <EOL> self . plugin . get_network_info . return_value = None <EOL> self . dhcp . enable_dhcp_helper ( '<STR_LIT>' ) <EOL> self . plugin . assert_has_calls ( <EOL> [ mock . call . get_network_info ( '<STR_LIT>' ) ] ) <EOL> self . assertFalse ( self . call_driver . called ) <EOL> self . assertFalse ( self . dhcp . schedule_resync . called ) <EOL> def test_enable_dhcp_helper_exception_during_rpc ( self ) : <EOL> self . plugin . get_network_info . side_effect = Exception <EOL> with mock . patch . object ( dhcp_agent . LOG , '<STR_LIT>' ) as log : <EOL> self . dhcp . enable_dhcp_helper ( fake_network . id ) <EOL> self . plugin . assert_has_calls ( <EOL> [ mock . call . get_network_info ( fake_network . id ) ] ) <EOL> self . assertFalse ( self . call_driver . called ) <EOL> self . assertTrue ( log . called ) <EOL> self . assertTrue ( self . schedule_resync . called ) <EOL> self . assertFalse ( self . cache . called ) <EOL> self . assertFalse ( self . external_process . called ) <EOL> def test_enable_dhcp_helper_driver_failure ( self ) : <EOL> self . plugin . get_network_info . return_value = fake_network <EOL> self . call_driver . return_value = False <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . dhcp . enable_dhcp_helper ( fake_network . id ) <EOL> self . plugin . assert_has_calls ( <EOL> [ mock . call . get_network_info ( fake_network . id ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , fake_network ) <EOL> self . assertFalse ( self . cache . called ) <EOL> self . assertFalse ( self . external_process . called ) <EOL> def _disable_dhcp_helper_known_network ( self , isolated_metadata = False ) : <EOL> if isolated_metadata : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . dhcp . disable_dhcp_helper ( fake_network . id ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( fake_network . id ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , fake_network ) <EOL> if isolated_metadata : <EOL> self . external_process . assert_has_calls ( [ <EOL> self . _process_manager_constructor_call ( ns = None ) , <EOL> mock . call ( ) . disable ( ) ] ) <EOL> else : <EOL> self . assertFalse ( self . external_process . call_count ) <EOL> def test_disable_dhcp_helper_known_network_isolated_metadata ( self ) : <EOL> self . _disable_dhcp_helper_known_network ( isolated_metadata = True ) <EOL> def test_disable_dhcp_helper_known_network ( self ) : <EOL> self . _disable_dhcp_helper_known_network ( ) <EOL> def test_disable_dhcp_helper_unknown_network ( self ) : <EOL> self . cache . get_network_by_id . return_value = None <EOL> self . dhcp . disable_dhcp_helper ( '<STR_LIT>' ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( '<STR_LIT>' ) ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . call_driver . call_count ) <EOL> self . assertFalse ( self . external_process . called ) <EOL> def _disable_dhcp_helper_driver_failure ( self , isolated_metadata = False ) : <EOL> if isolated_metadata : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . call_driver . return_value = False <EOL> self . dhcp . disable_dhcp_helper ( fake_network . id ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( fake_network . id ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , fake_network ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( fake_network . id ) ] ) <EOL> if isolated_metadata : <EOL> self . external_process . assert_has_calls ( [ <EOL> self . _process_manager_constructor_call ( ns = None ) , <EOL> mock . call ( ) . disable ( ) <EOL> ] ) <EOL> else : <EOL> self . assertFalse ( self . external_process . call_count ) <EOL> def test_disable_dhcp_helper_driver_failure_isolated_metadata ( self ) : <EOL> self . _disable_dhcp_helper_driver_failure ( isolated_metadata = True ) <EOL> def test_disable_dhcp_helper_driver_failure ( self ) : <EOL> self . _disable_dhcp_helper_driver_failure ( ) <EOL> def test_enable_isolated_metadata_proxy ( self ) : <EOL> self . dhcp . _process_monitor = mock . Mock ( ) <EOL> self . dhcp . enable_isolated_metadata_proxy ( fake_network ) <EOL> self . external_process . assert_has_calls ( [ <EOL> self . _process_manager_constructor_call ( ) , <EOL> mock . call ( ) . enable ( ) <EOL> ] ) <EOL> def test_disable_isolated_metadata_proxy ( self ) : <EOL> method_path = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with mock . patch ( method_path ) as destroy : <EOL> self . dhcp . disable_isolated_metadata_proxy ( fake_network ) <EOL> destroy . assert_called_once_with ( self . dhcp . _process_monitor , <EOL> fake_network . id , <EOL> cfg . CONF ) <EOL> def _test_enable_isolated_metadata_proxy ( self , network ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , False ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> method_path = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with mock . patch ( method_path ) as spawn : <EOL> self . dhcp . enable_isolated_metadata_proxy ( network ) <EOL> spawn . assert_called_once_with ( self . dhcp . _process_monitor , <EOL> network . namespace , <EOL> dhcp . METADATA_PORT , <EOL> cfg . CONF , <EOL> router_id = '<STR_LIT>' ) <EOL> def test_enable_isolated_metadata_proxy_with_metadata_network ( self ) : <EOL> self . _test_enable_isolated_metadata_proxy ( fake_meta_network ) <EOL> def test_enable_isolated_metadata_proxy_with_metadata_network_dvr ( self ) : <EOL> self . _test_enable_isolated_metadata_proxy ( fake_meta_dvr_network ) <EOL> def test_enable_isolated_metadata_proxy_with_dist_network ( self ) : <EOL> self . _test_enable_isolated_metadata_proxy ( fake_dist_network ) <EOL> def _test_disable_isolated_metadata_proxy ( self , network ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> method_path = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with mock . patch ( method_path ) as destroy : <EOL> self . dhcp . enable_isolated_metadata_proxy ( network ) <EOL> self . dhcp . disable_isolated_metadata_proxy ( network ) <EOL> destroy . assert_called_once_with ( self . dhcp . _process_monitor , <EOL> '<STR_LIT>' , <EOL> cfg . CONF ) <EOL> def test_disable_isolated_metadata_proxy_with_metadata_network ( self ) : <EOL> self . _test_disable_isolated_metadata_proxy ( fake_meta_network ) <EOL> def test_disable_isolated_metadata_proxy_with_metadata_network_dvr ( self ) : <EOL> self . _test_disable_isolated_metadata_proxy ( fake_meta_dvr_network ) <EOL> def test_disable_isolated_metadata_proxy_with_dist_network ( self ) : <EOL> self . _test_disable_isolated_metadata_proxy ( fake_dist_network ) <EOL> def test_network_create_end ( self ) : <EOL> payload = dict ( network = dict ( id = fake_network . id ) ) <EOL> with mock . patch . object ( self . dhcp , '<STR_LIT>' ) as enable : <EOL> self . dhcp . network_create_end ( None , payload ) <EOL> enable . assert_called_once_with ( fake_network . id ) <EOL> def test_network_update_end_admin_state_up ( self ) : <EOL> payload = dict ( network = dict ( id = fake_network . id , admin_state_up = True ) ) <EOL> with mock . patch . object ( self . dhcp , '<STR_LIT>' ) as enable : <EOL> self . dhcp . network_update_end ( None , payload ) <EOL> enable . assert_called_once_with ( fake_network . id ) <EOL> def test_network_update_end_admin_state_down ( self ) : <EOL> payload = dict ( network = dict ( id = fake_network . id , admin_state_up = False ) ) <EOL> with mock . patch . object ( self . dhcp , '<STR_LIT>' ) as disable : <EOL> self . dhcp . network_update_end ( None , payload ) <EOL> disable . assert_called_once_with ( fake_network . id ) <EOL> def test_network_delete_end ( self ) : <EOL> payload = dict ( network_id = fake_network . id ) <EOL> with mock . patch . object ( self . dhcp , '<STR_LIT>' ) as disable : <EOL> self . dhcp . network_delete_end ( None , payload ) <EOL> disable . assert_called_once_with ( fake_network . id ) <EOL> def test_refresh_dhcp_helper_no_dhcp_enabled_networks ( self ) : <EOL> network = dhcp . NetModel ( dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ ] , <EOL> ports = [ ] ) ) <EOL> self . cache . get_network_by_id . return_value = network <EOL> self . plugin . get_network_info . return_value = network <EOL> with mock . patch . object ( self . dhcp , '<STR_LIT>' ) as disable : <EOL> self . dhcp . refresh_dhcp_helper ( network . id ) <EOL> disable . assert_called_once_with ( network . id ) <EOL> self . assertFalse ( self . cache . called ) <EOL> self . assertFalse ( self . call_driver . called ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( '<STR_LIT>' ) ] ) <EOL> def test_refresh_dhcp_helper_exception_during_rpc ( self ) : <EOL> network = dhcp . NetModel ( dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> subnets = [ ] , <EOL> ports = [ ] ) ) <EOL> self . cache . get_network_by_id . return_value = network <EOL> self . plugin . get_network_info . side_effect = Exception <EOL> with mock . patch . object ( dhcp_agent . LOG , '<STR_LIT>' ) as log : <EOL> self . dhcp . refresh_dhcp_helper ( network . id ) <EOL> self . assertFalse ( self . call_driver . called ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( '<STR_LIT>' ) ] ) <EOL> self . assertTrue ( log . called ) <EOL> self . assertTrue ( self . dhcp . schedule_resync . called ) <EOL> def test_subnet_update_end ( self ) : <EOL> payload = dict ( subnet = dict ( network_id = fake_network . id ) ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . plugin . get_network_info . return_value = fake_network <EOL> self . dhcp . subnet_update_end ( None , payload ) <EOL> self . cache . assert_has_calls ( [ mock . call . put ( fake_network ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , <EOL> fake_network ) <EOL> def test_subnet_update_end_restart ( self ) : <EOL> new_state = dhcp . NetModel ( dict ( id = fake_network . id , <EOL> tenant_id = fake_network . tenant_id , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 , fake_subnet3 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> payload = dict ( subnet = dict ( network_id = fake_network . id ) ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . plugin . get_network_info . return_value = new_state <EOL> self . dhcp . subnet_update_end ( None , payload ) <EOL> self . cache . assert_has_calls ( [ mock . call . put ( new_state ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , <EOL> new_state ) <EOL> def test_subnet_update_end_delete_payload ( self ) : <EOL> prev_state = dhcp . NetModel ( dict ( id = fake_network . id , <EOL> tenant_id = fake_network . tenant_id , <EOL> admin_state_up = True , <EOL> subnets = [ fake_subnet1 , fake_subnet3 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> payload = dict ( subnet_id = fake_subnet1 . id ) <EOL> self . cache . get_network_by_subnet_id . return_value = prev_state <EOL> self . cache . get_network_by_id . return_value = prev_state <EOL> self . plugin . get_network_info . return_value = fake_network <EOL> self . dhcp . subnet_delete_end ( None , payload ) <EOL> self . cache . assert_has_calls ( [ <EOL> mock . call . get_network_by_subnet_id ( <EOL> '<STR_LIT>' ) , <EOL> mock . call . get_network_by_id ( '<STR_LIT>' ) , <EOL> mock . call . put ( fake_network ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , <EOL> fake_network ) <EOL> def test_port_update_end ( self ) : <EOL> payload = dict ( port = fake_port2 ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . cache . get_port_by_id . return_value = fake_port2 <EOL> self . dhcp . port_update_end ( None , payload ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( fake_port2 . network_id ) , <EOL> mock . call . put_port ( mock . ANY ) ] ) <EOL> self . call_driver . assert_called_once_with ( '<STR_LIT>' , <EOL> fake_network ) <EOL> def test_port_update_change_ip_on_port ( self ) : <EOL> payload = dict ( port = fake_port1 ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> updated_fake_port1 = copy . deepcopy ( fake_port1 ) <EOL> updated_fake_port1 . fixed_ips [ <NUM_LIT:0> ] . ip_address = '<STR_LIT>' <EOL> self . cache . get_port_by_id . return_value = updated_fake_port1 <EOL> self . dhcp . port_update_end ( None , payload ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_network_by_id ( fake_port1 . network_id ) , <EOL> mock . call . put_port ( mock . ANY ) ] ) <EOL> self . call_driver . assert_has_calls ( <EOL> [ mock . call . call_driver ( '<STR_LIT>' , fake_network ) ] ) <EOL> def test_port_update_change_ip_on_dhcp_agents_port ( self ) : <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . cache . get_port_by_id . return_value = fake_port1 <EOL> payload = dict ( port = copy . deepcopy ( fake_port1 ) ) <EOL> device_id = utils . get_dhcp_agent_device_id ( <EOL> payload [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] , self . dhcp . conf . host ) <EOL> payload [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> payload [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] = device_id <EOL> self . dhcp . port_update_end ( None , payload ) <EOL> self . call_driver . assert_has_calls ( <EOL> [ mock . call . call_driver ( '<STR_LIT>' , fake_network ) ] ) <EOL> def test_port_update_on_dhcp_agents_port_no_ip_change ( self ) : <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . cache . get_port_by_id . return_value = fake_port1 <EOL> payload = dict ( port = fake_port1 ) <EOL> device_id = utils . get_dhcp_agent_device_id ( <EOL> payload [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] , self . dhcp . conf . host ) <EOL> payload [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] = device_id <EOL> self . dhcp . port_update_end ( None , payload ) <EOL> self . call_driver . assert_has_calls ( <EOL> [ mock . call . call_driver ( '<STR_LIT>' , fake_network ) ] ) <EOL> def test_port_delete_end ( self ) : <EOL> payload = dict ( port_id = fake_port2 . id ) <EOL> self . cache . get_network_by_id . return_value = fake_network <EOL> self . cache . get_port_by_id . return_value = fake_port2 <EOL> self . dhcp . port_delete_end ( None , payload ) <EOL> self . cache . assert_has_calls ( <EOL> [ mock . call . get_port_by_id ( fake_port2 . id ) , <EOL> mock . call . get_network_by_id ( fake_network . id ) , <EOL> mock . call . remove_port ( fake_port2 ) ] ) <EOL> self . call_driver . assert_has_calls ( <EOL> [ mock . call . call_driver ( '<STR_LIT>' , fake_network ) ] ) <EOL> def test_port_delete_end_unknown_port ( self ) : <EOL> payload = dict ( port_id = '<STR_LIT>' ) <EOL> self . cache . get_port_by_id . return_value = None <EOL> self . dhcp . port_delete_end ( None , payload ) <EOL> self . cache . assert_has_calls ( [ mock . call . get_port_by_id ( '<STR_LIT>' ) ] ) <EOL> self . assertEqual ( self . call_driver . call_count , <NUM_LIT:0> ) <EOL> class TestDhcpPluginApiProxy ( base . BaseTestCase ) : <EOL> def _test_dhcp_api ( self , method , ** kwargs ) : <EOL> ctxt = context . get_admin_context ( ) <EOL> proxy = dhcp_agent . DhcpPluginApi ( '<STR_LIT:foo>' , ctxt , host = '<STR_LIT:foo>' ) <EOL> with mock . patch . object ( proxy . client , '<STR_LIT>' ) as rpc_mock , mock . patch . object ( proxy . client , '<STR_LIT>' ) as prepare_mock : <EOL> prepare_mock . return_value = proxy . client <EOL> rpc_mock . return_value = kwargs . pop ( '<STR_LIT>' , [ ] ) <EOL> prepare_args = { } <EOL> if '<STR_LIT:version>' in kwargs : <EOL> prepare_args [ '<STR_LIT:version>' ] = kwargs . pop ( '<STR_LIT:version>' ) <EOL> retval = getattr ( proxy , method ) ( ** kwargs ) <EOL> self . assertEqual ( retval , rpc_mock . return_value ) <EOL> prepare_mock . assert_called_once_with ( ** prepare_args ) <EOL> kwargs [ '<STR_LIT:host>' ] = proxy . host <EOL> rpc_mock . assert_called_once_with ( ctxt , method , ** kwargs ) <EOL> def test_get_active_networks_info ( self ) : <EOL> self . _test_dhcp_api ( '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test_get_network_info ( self ) : <EOL> self . _test_dhcp_api ( '<STR_LIT>' , network_id = '<STR_LIT>' , <EOL> return_value = None ) <EOL> def test_create_dhcp_port ( self ) : <EOL> self . _test_dhcp_api ( '<STR_LIT>' , port = '<STR_LIT>' , <EOL> return_value = None , version = '<STR_LIT>' ) <EOL> def test_update_dhcp_port ( self ) : <EOL> self . _test_dhcp_api ( '<STR_LIT>' , port_id = '<STR_LIT>' , <EOL> port = '<STR_LIT>' , return_value = None , version = '<STR_LIT>' ) <EOL> def test_release_dhcp_port ( self ) : <EOL> self . _test_dhcp_api ( '<STR_LIT>' , network_id = '<STR_LIT>' , <EOL> device_id = '<STR_LIT>' ) <EOL> class TestNetworkCache ( base . BaseTestCase ) : <EOL> def test_put_network ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( nc . cache , <EOL> { fake_network . id : fake_network } ) <EOL> self . assertEqual ( nc . subnet_lookup , <EOL> { fake_subnet1 . id : fake_network . id , <EOL> fake_subnet2 . id : fake_network . id } ) <EOL> self . assertEqual ( nc . port_lookup , <EOL> { fake_port1 . id : fake_network . id } ) <EOL> def test_put_network_existing ( self ) : <EOL> prev_network_info = mock . Mock ( ) <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> with mock . patch . object ( nc , '<STR_LIT>' ) as remove : <EOL> nc . cache [ fake_network . id ] = prev_network_info <EOL> nc . put ( fake_network ) <EOL> remove . assert_called_once_with ( prev_network_info ) <EOL> self . assertEqual ( nc . cache , <EOL> { fake_network . id : fake_network } ) <EOL> self . assertEqual ( nc . subnet_lookup , <EOL> { fake_subnet1 . id : fake_network . id , <EOL> fake_subnet2 . id : fake_network . id } ) <EOL> self . assertEqual ( nc . port_lookup , <EOL> { fake_port1 . id : fake_network . id } ) <EOL> def test_remove_network ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . cache = { fake_network . id : fake_network } <EOL> nc . subnet_lookup = { fake_subnet1 . id : fake_network . id , <EOL> fake_subnet2 . id : fake_network . id } <EOL> nc . port_lookup = { fake_port1 . id : fake_network . id } <EOL> nc . remove ( fake_network ) <EOL> self . assertEqual ( len ( nc . cache ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( nc . subnet_lookup ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( nc . port_lookup ) , <NUM_LIT:0> ) <EOL> def test_get_network_by_id ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( nc . get_network_by_id ( fake_network . id ) , fake_network ) <EOL> def test_get_network_ids ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( list ( nc . get_network_ids ( ) ) , [ fake_network . id ] ) <EOL> def test_get_network_by_subnet_id ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( nc . get_network_by_subnet_id ( fake_subnet1 . id ) , <EOL> fake_network ) <EOL> def test_get_network_by_port_id ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( nc . get_network_by_port_id ( fake_port1 . id ) , <EOL> fake_network ) <EOL> def test_put_port ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ fake_port1 ] ) ) <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_net ) <EOL> nc . put_port ( fake_port2 ) <EOL> self . assertEqual ( len ( nc . port_lookup ) , <NUM_LIT:2> ) <EOL> self . assertIn ( fake_port2 , fake_net . ports ) <EOL> def test_put_port_existing ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ fake_port1 , fake_port2 ] ) ) <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_net ) <EOL> nc . put_port ( fake_port2 ) <EOL> self . assertEqual ( len ( nc . port_lookup ) , <NUM_LIT:2> ) <EOL> self . assertIn ( fake_port2 , fake_net . ports ) <EOL> def test_remove_port_existing ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' , <EOL> subnets = [ fake_subnet1 ] , <EOL> ports = [ fake_port1 , fake_port2 ] ) ) <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_net ) <EOL> nc . remove_port ( fake_port2 ) <EOL> self . assertEqual ( len ( nc . port_lookup ) , <NUM_LIT:1> ) <EOL> self . assertNotIn ( fake_port2 , fake_net . ports ) <EOL> def test_get_port_by_id ( self ) : <EOL> nc = dhcp_agent . NetworkCache ( ) <EOL> nc . put ( fake_network ) <EOL> self . assertEqual ( nc . get_port_by_id ( fake_port1 . id ) , fake_port1 ) <EOL> class FakePort1 ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> class FakeV4Subnet ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> self . ip_version = <NUM_LIT:4> <EOL> self . cidr = '<STR_LIT>' <EOL> self . gateway_ip = '<STR_LIT>' <EOL> self . enable_dhcp = True <EOL> class FakeV4SubnetOutsideGateway ( FakeV4Subnet ) : <EOL> def __init__ ( self ) : <EOL> super ( FakeV4SubnetOutsideGateway , self ) . __init__ ( ) <EOL> self . gateway_ip = '<STR_LIT>' <EOL> class FakeV4SubnetNoGateway ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> self . ip_version = <NUM_LIT:4> <EOL> self . cidr = '<STR_LIT>' <EOL> self . gateway_ip = None <EOL> self . enable_dhcp = True <EOL> class FakeV4Network ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> self . subnets = [ FakeV4Subnet ( ) ] <EOL> self . ports = [ FakePort1 ( ) ] <EOL> self . namespace = '<STR_LIT>' <EOL> class FakeV4NetworkOutsideGateway ( FakeV4Network ) : <EOL> def __init__ ( self ) : <EOL> super ( FakeV4NetworkOutsideGateway , self ) . __init__ ( ) <EOL> self . subnets = [ FakeV4SubnetOutsideGateway ( ) ] <EOL> class FakeV4NetworkNoSubnet ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> self . subnets = [ ] <EOL> self . ports = [ ] <EOL> class FakeV4NetworkNoGateway ( object ) : <EOL> def __init__ ( self ) : <EOL> self . id = '<STR_LIT>' <EOL> self . subnets = [ FakeV4SubnetNoGateway ( ) ] <EOL> self . ports = [ FakePort1 ( ) ] <EOL> class TestDeviceManager ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDeviceManager , self ) . setUp ( ) <EOL> config . register_interface_driver_opts_helper ( cfg . CONF ) <EOL> cfg . CONF . register_opts ( dhcp_config . DHCP_AGENT_OPTS ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . ensure_device_is_ready_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . ensure_device_is_ready = ( self . ensure_device_is_ready_p . start ( ) ) <EOL> self . dvr_cls_p = mock . patch ( '<STR_LIT>' ) <EOL> self . iproute_cls_p = mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> driver_cls = self . dvr_cls_p . start ( ) <EOL> iproute_cls = self . iproute_cls_p . start ( ) <EOL> self . mock_driver = mock . MagicMock ( ) <EOL> self . mock_driver . DEV_NAME_LEN = ( <EOL> interface . LinuxInterfaceDriver . DEV_NAME_LEN ) <EOL> self . mock_driver . use_gateway_ips = False <EOL> self . mock_iproute = mock . MagicMock ( ) <EOL> driver_cls . return_value = self . mock_driver <EOL> iproute_cls . return_value = self . mock_iproute <EOL> iptables_cls_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> iptables_cls = iptables_cls_p . start ( ) <EOL> self . iptables_inst = mock . Mock ( ) <EOL> iptables_cls . return_value = self . iptables_inst <EOL> self . mangle_inst = mock . Mock ( ) <EOL> self . iptables_inst . ipv4 = { '<STR_LIT>' : self . mangle_inst } <EOL> self . mock_ip_wrapper_p = mock . patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . mock_ip_wrapper = self . mock_ip_wrapper_p . start ( ) <EOL> def _test_setup_helper ( self , device_is_ready , net = None , port = None ) : <EOL> net = net or fake_network <EOL> port = port or fake_port1 <EOL> plugin = mock . Mock ( ) <EOL> plugin . create_dhcp_port . return_value = port or fake_port1 <EOL> self . ensure_device_is_ready . return_value = device_is_ready <EOL> self . mock_driver . get_device_name . return_value = '<STR_LIT>' <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . _set_default_route = mock . Mock ( ) <EOL> dh . _cleanup_stale_devices = mock . Mock ( ) <EOL> interface_name = dh . setup ( net ) <EOL> self . assertEqual ( interface_name , '<STR_LIT>' ) <EOL> plugin . assert_has_calls ( [ <EOL> mock . call . create_dhcp_port ( <EOL> { '<STR_LIT:port>' : { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : net . id , '<STR_LIT>' : net . tenant_id , <EOL> '<STR_LIT>' : <EOL> [ { '<STR_LIT>' : port . fixed_ips [ <NUM_LIT:0> ] . subnet_id } ] , <EOL> '<STR_LIT>' : mock . ANY } } ) ] ) <EOL> if port == fake_ipv6_port : <EOL> expected_ips = [ '<STR_LIT>' ] <EOL> else : <EOL> expected_ips = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> expected = [ <EOL> mock . call . get_device_name ( port ) , <EOL> mock . call . init_l3 ( <EOL> '<STR_LIT>' , <EOL> expected_ips , <EOL> namespace = net . namespace ) ] <EOL> if not device_is_ready : <EOL> expected . insert ( <NUM_LIT:1> , <EOL> mock . call . plug ( net . id , <EOL> port . id , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> namespace = net . namespace , <EOL> mtu = None ) ) <EOL> self . mock_driver . assert_has_calls ( expected ) <EOL> dh . _set_default_route . assert_called_once_with ( net , '<STR_LIT>' ) <EOL> def test_setup ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , False ) <EOL> self . _test_setup_helper ( False ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> self . _test_setup_helper ( False ) <EOL> def test_setup_calls_fill_dhcp_udp_checksums ( self ) : <EOL> self . _test_setup_helper ( False ) <EOL> rule = ( '<STR_LIT>' <EOL> % const . DHCP_RESPONSE_PORT ) <EOL> expected = [ mock . call . add_rule ( '<STR_LIT>' , rule ) ] <EOL> self . mangle_inst . assert_has_calls ( expected ) <EOL> def test_setup_create_dhcp_port ( self ) : <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> plugin = mock . Mock ( ) <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = None <EOL> net = copy . deepcopy ( fake_network ) <EOL> plugin . create_dhcp_port . return_value = fake_dhcp_port <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . setup ( net ) <EOL> plugin . assert_has_calls ( [ <EOL> mock . call . create_dhcp_port ( <EOL> { '<STR_LIT:port>' : { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : net . id , <EOL> '<STR_LIT>' : net . tenant_id , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : <EOL> fake_dhcp_port . fixed_ips [ <NUM_LIT:0> ] . subnet_id } ] , <EOL> '<STR_LIT>' : mock . ANY } } ) ] ) <EOL> self . assertIn ( fake_dhcp_port , net . ports ) <EOL> def test_setup_plug_exception ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> plugin . create_dhcp_port . return_value = fake_dhcp_port <EOL> self . ensure_device_is_ready . return_value = False <EOL> self . mock_driver . get_device_name . return_value = '<STR_LIT>' <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . _set_default_route = mock . Mock ( ) <EOL> dh . _cleanup_stale_devices = mock . Mock ( ) <EOL> dh . driver = mock . Mock ( ) <EOL> dh . driver . plug . side_effect = OSError ( ) <EOL> net = copy . deepcopy ( fake_network ) <EOL> self . assertRaises ( OSError , dh . setup , net ) <EOL> plugin . release_dhcp_port . assert_called_once_with ( <EOL> net . id , mock . ANY ) <EOL> def test_setup_ipv6 ( self ) : <EOL> self . _test_setup_helper ( True , net = fake_network_ipv6 , <EOL> port = fake_ipv6_port ) <EOL> def test_setup_device_is_ready ( self ) : <EOL> self . _test_setup_helper ( True ) <EOL> def test_create_dhcp_port_raise_conflict ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> plugin . create_dhcp_port . return_value = None <EOL> self . assertRaises ( exceptions . Conflict , <EOL> dh . setup_dhcp_port , <EOL> fake_network ) <EOL> def test_create_dhcp_port_create_new ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> plugin . create_dhcp_port . return_value = fake_network . ports [ <NUM_LIT:0> ] <EOL> dh . setup_dhcp_port ( fake_network ) <EOL> plugin . assert_has_calls ( [ <EOL> mock . call . create_dhcp_port ( <EOL> { '<STR_LIT:port>' : { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : <EOL> fake_network . id , '<STR_LIT>' : fake_network . tenant_id , <EOL> '<STR_LIT>' : <EOL> [ { '<STR_LIT>' : fake_fixed_ip1 . subnet_id } ] , <EOL> '<STR_LIT>' : mock . ANY } } ) ] ) <EOL> def test_create_dhcp_port_update_add_subnet ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> fake_network_copy = copy . deepcopy ( fake_network ) <EOL> fake_network_copy . ports [ <NUM_LIT:0> ] . device_id = dh . get_device_id ( fake_network ) <EOL> fake_network_copy . subnets [ <NUM_LIT:1> ] . enable_dhcp = True <EOL> plugin . update_dhcp_port . return_value = fake_network . ports [ <NUM_LIT:0> ] <EOL> dh . setup_dhcp_port ( fake_network_copy ) <EOL> port_body = { '<STR_LIT:port>' : { <EOL> '<STR_LIT>' : fake_network . id , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : fake_fixed_ip1 . subnet_id , <EOL> '<STR_LIT>' : fake_fixed_ip1 . ip_address } , <EOL> { '<STR_LIT>' : fake_subnet2 . id } ] } } <EOL> plugin . assert_has_calls ( [ <EOL> mock . call . update_dhcp_port ( fake_network_copy . ports [ <NUM_LIT:0> ] . id , <EOL> port_body ) ] ) <EOL> def test_update_dhcp_port_raises_conflict ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> fake_network_copy = copy . deepcopy ( fake_network ) <EOL> fake_network_copy . ports [ <NUM_LIT:0> ] . device_id = dh . get_device_id ( fake_network ) <EOL> fake_network_copy . subnets [ <NUM_LIT:1> ] . enable_dhcp = True <EOL> plugin . update_dhcp_port . return_value = None <EOL> self . assertRaises ( exceptions . Conflict , <EOL> dh . setup_dhcp_port , <EOL> fake_network_copy ) <EOL> def test_create_dhcp_port_no_update_or_create ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> fake_network_copy = copy . deepcopy ( fake_network ) <EOL> fake_network_copy . ports [ <NUM_LIT:0> ] . device_id = dh . get_device_id ( fake_network ) <EOL> dh . setup_dhcp_port ( fake_network_copy ) <EOL> self . assertFalse ( plugin . setup_dhcp_port . called ) <EOL> self . assertFalse ( plugin . update_dhcp_port . called ) <EOL> def test_setup_dhcp_port_with_non_enable_dhcp_subnet ( self ) : <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> fake_network_copy = copy . deepcopy ( fake_network ) <EOL> fake_network_copy . ports [ <NUM_LIT:0> ] . device_id = dh . get_device_id ( fake_network ) <EOL> plugin . update_dhcp_port . return_value = fake_port1 <EOL> self . assertEqual ( fake_subnet1 . id , <EOL> dh . setup_dhcp_port ( fake_network_copy ) . fixed_ips [ <NUM_LIT:0> ] . subnet_id ) <EOL> def test_destroy ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = FAKE_NETWORK_UUID , <EOL> tenant_id = '<STR_LIT>' ) ) <EOL> with mock . patch ( '<STR_LIT>' ) as dvr_cls : <EOL> mock_driver = mock . MagicMock ( ) <EOL> mock_driver . get_device_name . return_value = '<STR_LIT>' <EOL> dvr_cls . return_value = mock_driver <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . destroy ( fake_net , '<STR_LIT>' ) <EOL> dvr_cls . assert_called_once_with ( cfg . CONF ) <EOL> mock_driver . assert_has_calls ( <EOL> [ mock . call . unplug ( '<STR_LIT>' , <EOL> namespace = '<STR_LIT>' + fake_net . id ) ] ) <EOL> plugin . assert_has_calls ( <EOL> [ mock . call . release_dhcp_port ( fake_net . id , mock . ANY ) ] ) <EOL> def test_destroy_with_none ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = FAKE_NETWORK_UUID , <EOL> tenant_id = '<STR_LIT>' ) ) <EOL> with mock . patch ( '<STR_LIT>' ) as dvr_cls : <EOL> mock_driver = mock . MagicMock ( ) <EOL> mock_driver . get_device_name . return_value = '<STR_LIT>' <EOL> dvr_cls . return_value = mock_driver <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . destroy ( fake_net , None ) <EOL> dvr_cls . assert_called_once_with ( cfg . CONF ) <EOL> plugin . assert_has_calls ( <EOL> [ mock . call . release_dhcp_port ( fake_net . id , mock . ANY ) ] ) <EOL> self . assertFalse ( mock_driver . called ) <EOL> def test_get_interface_name ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' ) ) <EOL> fake_port = dhcp . DictModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> mac_address = '<STR_LIT>' ) ) <EOL> with mock . patch ( '<STR_LIT>' ) as dvr_cls : <EOL> mock_driver = mock . MagicMock ( ) <EOL> mock_driver . get_device_name . return_value = '<STR_LIT>' <EOL> dvr_cls . return_value = mock_driver <EOL> plugin = mock . Mock ( ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , plugin ) <EOL> dh . get_interface_name ( fake_net , fake_port ) <EOL> dvr_cls . assert_called_once_with ( cfg . CONF ) <EOL> mock_driver . assert_has_calls ( <EOL> [ mock . call . get_device_name ( fake_port ) ] ) <EOL> self . assertEqual ( len ( plugin . mock_calls ) , <NUM_LIT:0> ) <EOL> def test_get_device_id ( self ) : <EOL> fake_net = dhcp . NetModel ( <EOL> dict ( id = '<STR_LIT>' , <EOL> tenant_id = '<STR_LIT>' ) ) <EOL> expected = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> local_hostname = cfg . CONF . host . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] <EOL> with mock . patch ( '<STR_LIT>' ) as uuid5 : <EOL> uuid5 . return_value = '<STR_LIT>' <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> self . assertEqual ( dh . get_device_id ( fake_net ) , expected ) <EOL> uuid5 . assert_called_once_with ( uuid . NAMESPACE_DNS , local_hostname ) <EOL> def test_update ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , False ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> dh . _set_default_route = mock . Mock ( ) <EOL> network = mock . Mock ( ) <EOL> dh . update ( network , '<STR_LIT>' ) <EOL> dh . _set_default_route . assert_called_once_with ( network , <EOL> '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> dh . _set_default_route = mock . Mock ( ) <EOL> dh . update ( FakeV4Network ( ) , '<STR_LIT>' ) <EOL> self . assertTrue ( dh . _set_default_route . called ) <EOL> def test_set_default_route ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = None <EOL> network = FakeV4Network ( ) <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> device . route . add_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_set_default_route_outside_subnet ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = None <EOL> network = FakeV4NetworkOutsideGateway ( ) <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> device . route . add_route . assert_called_once_with ( '<STR_LIT>' , <EOL> scope = '<STR_LIT>' ) <EOL> device . route . add_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_set_default_route_no_subnet ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = None <EOL> network = FakeV4NetworkNoSubnet ( ) <EOL> network . namespace = '<STR_LIT>' <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> self . assertFalse ( device . route . add_gateway . called ) <EOL> def test_set_default_route_no_subnet_delete_gateway ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = dict ( gateway = '<STR_LIT>' ) <EOL> network = FakeV4NetworkNoSubnet ( ) <EOL> network . namespace = '<STR_LIT>' <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> device . route . delete_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> self . assertFalse ( device . route . add_gateway . called ) <EOL> def test_set_default_route_no_gateway ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = dict ( gateway = '<STR_LIT>' ) <EOL> network = FakeV4NetworkNoGateway ( ) <EOL> network . namespace = '<STR_LIT>' <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> device . route . delete_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> self . assertFalse ( device . route . add_gateway . called ) <EOL> def test_set_default_route_do_nothing ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = dict ( gateway = '<STR_LIT>' ) <EOL> network = FakeV4Network ( ) <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> self . assertFalse ( device . route . add_gateway . called ) <EOL> def test_set_default_route_change_gateway ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = dict ( gateway = '<STR_LIT>' ) <EOL> network = FakeV4Network ( ) <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> device . route . add_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_set_default_route_change_gateway_outside_subnet ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . list_onlink_routes . return_value = ( <EOL> [ { '<STR_LIT>' : '<STR_LIT>' } ] ) <EOL> device . route . get_gateway . return_value = dict ( gateway = '<STR_LIT>' ) <EOL> network = FakeV4NetworkOutsideGateway ( ) <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertEqual ( device . route . list_onlink_routes . call_count , <NUM_LIT:2> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> device . route . delete_route . assert_called_once_with ( '<STR_LIT>' , <EOL> scope = '<STR_LIT>' ) <EOL> device . route . add_route . assert_called_once_with ( '<STR_LIT>' , <EOL> scope = '<STR_LIT>' ) <EOL> device . route . add_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_set_default_route_two_subnets ( self ) : <EOL> dh = dhcp . DeviceManager ( cfg . CONF , None ) <EOL> with mock . patch . object ( dhcp . ip_lib , '<STR_LIT>' ) as mock_IPDevice : <EOL> device = mock . Mock ( ) <EOL> mock_IPDevice . return_value = device <EOL> device . route . get_gateway . return_value = None <EOL> network = FakeV4Network ( ) <EOL> subnet2 = FakeV4Subnet ( ) <EOL> subnet2 . gateway_ip = '<STR_LIT>' <EOL> network . subnets = [ subnet2 , FakeV4Subnet ( ) ] <EOL> dh . _set_default_route ( network , '<STR_LIT>' ) <EOL> self . assertEqual ( device . route . get_gateway . call_count , <NUM_LIT:1> ) <EOL> self . assertFalse ( device . route . delete_gateway . called ) <EOL> device . route . add_gateway . assert_called_once_with ( '<STR_LIT>' ) <EOL> class TestDictModel ( base . BaseTestCase ) : <EOL> def test_basic_dict ( self ) : <EOL> d = dict ( a = <NUM_LIT:1> , b = <NUM_LIT:2> ) <EOL> m = dhcp . DictModel ( d ) <EOL> self . assertEqual ( m . a , <NUM_LIT:1> ) <EOL> self . assertEqual ( m . b , <NUM_LIT:2> ) <EOL> def test_dict_has_sub_dict ( self ) : <EOL> d = dict ( a = dict ( b = <NUM_LIT:2> ) ) <EOL> m = dhcp . DictModel ( d ) <EOL> self . assertEqual ( m . a . b , <NUM_LIT:2> ) <EOL> def test_dict_contains_list ( self ) : <EOL> d = dict ( a = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> m = dhcp . DictModel ( d ) <EOL> self . assertEqual ( m . a , [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> def test_dict_contains_list_of_dicts ( self ) : <EOL> d = dict ( a = [ dict ( b = <NUM_LIT:2> ) , dict ( c = <NUM_LIT:3> ) ] ) <EOL> m = dhcp . DictModel ( d ) <EOL> self . assertEqual ( m . a [ <NUM_LIT:0> ] . b , <NUM_LIT:2> ) <EOL> self . assertEqual ( m . a [ <NUM_LIT:1> ] . c , <NUM_LIT:3> ) </s>
