<s> import mock <EOL> import six <EOL> from neutron . tests import base <EOL> from networking_cisco . apps . saf . common import config <EOL> from networking_cisco . apps . saf . common import constants <EOL> from networking_cisco . apps . saf . db import dfa_db_models as dbm <EOL> from networking_cisco . apps . saf . server import cisco_dfa_rest as cdr <EOL> from networking_cisco . apps . saf . server import dfa_events_handler as deh <EOL> from networking_cisco . apps . saf . server import dfa_fail_recovery as dfr <EOL> from networking_cisco . apps . saf . server import dfa_instance_api as dia <EOL> FAKE_NETWORK_NAME = '<STR_LIT>' <EOL> FAKE_NETWORK_ID = '<STR_LIT>' <EOL> FAKE_CFG_PROF_ID = '<STR_LIT>' <EOL> FAKE_SEG_ID = <NUM_LIT> <EOL> FAKE_PROJECT_NAME = '<STR_LIT>' <EOL> FAKE_PROJECT_ID = '<STR_LIT>' <EOL> FAKE_CFG_PROFILE_NAME = '<STR_LIT>' <EOL> FAKE_INSTANCE_NAME = '<STR_LIT>' <EOL> FAKE_SUBNET_ID = '<STR_LIT>' <EOL> FAKE_PORT_ID = '<STR_LIT>' <EOL> FAKE_DEVICE_ID = '<STR_LIT>' <EOL> FAKE_SECURITY_GRP_ID = '<STR_LIT>' <EOL> FAKE_MAC_ADDR = '<STR_LIT>' <EOL> FAKE_IP_ADDR = '<STR_LIT>' <EOL> FAKE_GW_ADDR = '<STR_LIT>' <EOL> FAKE_DHCP_IP_START = '<STR_LIT>' <EOL> FAKE_DHCP_IP_END = '<STR_LIT>' <EOL> FAKE_HOST_ID = '<STR_LIT>' <EOL> FAKE_FWD_MODE = '<STR_LIT>' <EOL> FAKE_DCNM_USERNAME = '<STR_LIT>' <EOL> FAKE_DCNM_PASSWD = '<STR_LIT:password>' <EOL> FAKE_DCNM_IP = '<STR_LIT>' <EOL> class FakeClass ( object ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def imitate ( cls , * others ) : <EOL> for other in others : <EOL> for name in other . __dict__ : <EOL> try : <EOL> setattr ( cls , name , mock . Mock ( ) ) <EOL> except ( TypeError , AttributeError ) : <EOL> pass <EOL> return cls <EOL> class FakeProject ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , proj_id , name , dci_id , desc ) : <EOL> self . id = proj_id <EOL> self . name = name <EOL> self . dci_id = dci_id <EOL> self . description = desc <EOL> class TestDFAServer ( base . BaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestDFAServer , self ) . setUp ( ) <EOL> self . dcnmpatcher = mock . patch ( cdr . __name__ + '<STR_LIT>' ) <EOL> self . mdcnm = self . dcnmpatcher . start ( ) <EOL> self . keys_patcher = mock . patch ( deh . __name__ + '<STR_LIT>' ) <EOL> self . mkeys = self . keys_patcher . start ( ) <EOL> self . inst_api_patcher = mock . patch ( dia . __name__ + '<STR_LIT>' ) <EOL> self . m_inst_api = self . inst_api_patcher . start ( ) <EOL> self . module_patcher = mock . patch . dict ( '<STR_LIT>' , <EOL> { '<STR_LIT>' : mock . Mock ( ) } ) <EOL> self . module_patcher . start ( ) <EOL> from networking_cisco . apps . saf . server import dfa_listen_dcnm as dld <EOL> from networking_cisco . apps . saf . server import dfa_server as ds <EOL> self . dld_patcher = mock . patch ( dld . __name__ + '<STR_LIT>' ) <EOL> self . dld = self . dld_patcher . start ( ) <EOL> ds . DfaServer . __bases__ = ( FakeClass . imitate ( <EOL> dfr . DfaFailureRecovery , dbm . DfaDBMixin ) , ) <EOL> ds . DfaServer . get_all_projects . return_value = [ ] <EOL> ds . DfaServer . get_all_networks . return_value = [ ] <EOL> ds . DfaServer . _setup_rpc = mock . Mock ( ) <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = FAKE_DCNM_IP <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = FAKE_DCNM_USERNAME <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = FAKE_DCNM_PASSWD <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = <NUM_LIT> <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = <NUM_LIT> <EOL> config . default_dcnm_opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = <NUM_LIT> <EOL> self . cfg = config . CiscoDFAConfig ( ) . cfg <EOL> self . segid = int ( self . cfg . dcnm . segmentation_id_min ) + <NUM_LIT:10> <EOL> self . dfa_server = ds . DfaServer ( self . cfg ) <EOL> self . dciid = str ( <NUM_LIT> ) <EOL> self . proj_desc = '<STR_LIT>' <EOL> projs = [ <EOL> FakeProject ( FAKE_PROJECT_ID , FAKE_PROJECT_NAME , <EOL> self . dciid , self . proj_desc ) ] <EOL> self . dfa_server . get_all_projects . return_value = projs <EOL> self . dfa_server . _load_project_info_cache ( ) <EOL> self . part_name = self . cfg . dcnm . default_partition_name <EOL> def _get_port_info ( self ) : <EOL> port_info = { '<STR_LIT:port>' : { <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : FAKE_HOST_ID , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : FAKE_SUBNET_ID , <EOL> '<STR_LIT>' : FAKE_IP_ADDR } ] , <EOL> '<STR_LIT:id>' : FAKE_PORT_ID , <EOL> '<STR_LIT>' : [ FAKE_SECURITY_GRP_ID ] , <EOL> '<STR_LIT>' : FAKE_DEVICE_ID , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : FAKE_NETWORK_ID , <EOL> '<STR_LIT>' : FAKE_PROJECT_ID , <EOL> '<STR_LIT>' : { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : FAKE_MAC_ADDR } } <EOL> return port_info <EOL> def _load_network_info ( self ) : <EOL> dnet = mock . Mock ( ) <EOL> dnet . network_id = FAKE_NETWORK_ID <EOL> dnet . segmentation_id = self . segid <EOL> dnet . config_profile = FAKE_CFG_PROFILE_NAME <EOL> dnet . fwd_mod = FAKE_FWD_MODE <EOL> dnet . tenant_id = FAKE_PROJECT_ID <EOL> dnet . name = FAKE_NETWORK_NAME <EOL> self . dfa_server . get_all_networks . return_value = [ dnet ] <EOL> self . dfa_server . _load_network_info ( ) <EOL> def test_update_project_info_cache ( self ) : <EOL> """<STR_LIT>""" <EOL> pid = FAKE_PROJECT_ID <EOL> name = FAKE_PROJECT_NAME <EOL> dciid = <NUM_LIT:1000> <EOL> result = constants . RESULT_SUCCESS <EOL> self . dfa_server . update_project_info_cache ( pid , dci_id = dciid , <EOL> name = name , opcode = '<STR_LIT>' ) <EOL> self . assertTrue ( self . dfa_server . add_project_db . called ) <EOL> self . assertFalse ( self . dfa_server . update_project_entry . called ) <EOL> self . assertFalse ( self . dfa_server . del_project_db . called ) <EOL> self . dfa_server . add_project_db . assert_called_with ( pid , name , <EOL> dciid , result ) <EOL> self . dfa_server . update_project_info_cache ( pid , dci_id = dciid , <EOL> name = name , <EOL> opcode = '<STR_LIT>' ) <EOL> self . assertTrue ( self . dfa_server . update_project_entry . called ) <EOL> self . assertFalse ( self . dfa_server . del_project_db . called ) <EOL> self . dfa_server . update_project_entry . assert_called_with ( pid , dciid , <EOL> result ) <EOL> def test_project_create_event ( self ) : <EOL> """<STR_LIT>""" <EOL> dciid = str ( <NUM_LIT> ) <EOL> proj_desc = '<STR_LIT>' <EOL> proj_info = { '<STR_LIT>' : FAKE_PROJECT_ID } <EOL> proj = mock . Mock ( ) <EOL> proj . name = FAKE_PROJECT_NAME <EOL> proj . description = proj_desc <EOL> part_name = self . cfg . dcnm . default_partition_name <EOL> self . dfa_server . keystone_event . _service . projects . get . return_value = ( <EOL> proj ) <EOL> self . dfa_server . project_create_event ( proj_info ) <EOL> proj . name = FAKE_PROJECT_NAME + '<STR_LIT>' + dciid <EOL> self . dfa_server . project_create_event ( proj_info ) <EOL> expected_calls = [ <EOL> mock . call ( FAKE_PROJECT_NAME , part_name , None , proj . description ) , <EOL> mock . call ( FAKE_PROJECT_NAME , part_name , dciid , proj . description ) ] <EOL> self . assertEqual ( <EOL> expected_calls , <EOL> self . dfa_server . dcnm_client . create_project . call_args_list ) <EOL> def test_project_update_event ( self ) : <EOL> """<STR_LIT>""" <EOL> proj_info = { '<STR_LIT>' : FAKE_PROJECT_ID } <EOL> proj = mock . Mock ( ) <EOL> proj . name = FAKE_PROJECT_NAME + '<STR_LIT>' + self . dciid <EOL> proj . description = self . proj_desc <EOL> self . dfa_server . keystone_event . _service . projects . get . return_value = ( <EOL> proj ) <EOL> self . dfa_server . project_update_event ( proj_info ) <EOL> self . assertFalse ( <EOL> self . dfa_server . dcnm_client . update_project . called ) <EOL> proj . name = FAKE_PROJECT_NAME + '<STR_LIT>' + '<STR_LIT>' + self . dciid <EOL> self . dfa_server . project_update_event ( proj_info ) <EOL> self . assertFalse ( <EOL> self . dfa_server . dcnm_client . update_project . called ) <EOL> proj . name = FAKE_PROJECT_NAME + '<STR_LIT>' + str ( <NUM_LIT> ) <EOL> self . dfa_server . project_update_event ( proj_info ) <EOL> self . assertTrue ( <EOL> self . dfa_server . dcnm_client . update_project . called ) <EOL> expected_calls = [ mock . call ( FAKE_PROJECT_NAME , <EOL> self . cfg . dcnm . default_partition_name , <EOL> str ( <NUM_LIT> ) ) ] <EOL> self . assertEqual ( <EOL> expected_calls , <EOL> self . dfa_server . dcnm_client . update_project . call_args_list ) <EOL> def test_project_delete_event ( self ) : <EOL> """<STR_LIT>""" <EOL> proj_name = FAKE_PROJECT_NAME <EOL> proj_info = { '<STR_LIT>' : FAKE_PROJECT_ID } <EOL> part_name = self . cfg . dcnm . default_partition_name <EOL> self . dfa_server . project_delete_event ( proj_info ) <EOL> self . dfa_server . dcnm_client . delete_project . assert_called_with ( <EOL> proj_name , part_name ) <EOL> self . dfa_server . del_project_db . assert_called_with ( FAKE_PROJECT_ID ) <EOL> def test_network_create_event ( self ) : <EOL> """<STR_LIT>""" <EOL> network_info = { '<STR_LIT>' : <EOL> { '<STR_LIT:name>' : FAKE_NETWORK_NAME , <EOL> '<STR_LIT>' : FAKE_PROJECT_ID , <EOL> '<STR_LIT:id>' : FAKE_NETWORK_ID } } <EOL> dcnmclnt = self . dfa_server . dcnm_client <EOL> dcnmclnt . get_config_profile_for_network . return_value = ( <EOL> FAKE_CFG_PROFILE_NAME , FAKE_FWD_MODE ) <EOL> self . dfa_server . network_create_event ( network_info ) <EOL> dfa_net = self . dfa_server . network [ FAKE_NETWORK_ID ] <EOL> expected_calls = [ mock . call ( FAKE_NETWORK_ID , dfa_net , '<STR_LIT>' , <EOL> constants . RESULT_SUCCESS ) ] <EOL> self . assertEqual ( expected_calls , <EOL> self . dfa_server . add_network_db . call_args_list ) <EOL> def test_subnet_create_event ( self ) : <EOL> """<STR_LIT>""" <EOL> network_info = { '<STR_LIT>' : <EOL> { '<STR_LIT:name>' : FAKE_NETWORK_NAME , <EOL> '<STR_LIT>' : FAKE_PROJECT_ID , <EOL> '<STR_LIT:id>' : FAKE_NETWORK_ID } } <EOL> subnet_info = { '<STR_LIT>' : { <EOL> '<STR_LIT>' : FAKE_NETWORK_ID , <EOL> '<STR_LIT>' : FAKE_PROJECT_ID , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:start>' : FAKE_DHCP_IP_START , '<STR_LIT:end>' : FAKE_DHCP_IP_END } ] , <EOL> '<STR_LIT>' : FAKE_GW_ADDR , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : FAKE_IP_ADDR + '<STR_LIT>' , <EOL> '<STR_LIT:id>' : FAKE_SUBNET_ID } } <EOL> dcnmclnt = self . dfa_server . dcnm_client <EOL> dcnmclnt . get_config_profile_for_network . return_value = ( <EOL> FAKE_CFG_PROFILE_NAME , FAKE_FWD_MODE ) <EOL> self . dfa_server . network_create_event ( network_info ) <EOL> fake_network = mock . Mock ( ) <EOL> fake_network . source = '<STR_LIT>' <EOL> fake_network . name = FAKE_NETWORK_NAME <EOL> self . dfa_server . get_network . return_value = fake_network <EOL> self . dfa_server . subnet_create_event ( subnet_info ) <EOL> self . assertFalse ( self . dfa_server . dcnm_client . create_network . called ) <EOL> fake_network . source = '<STR_LIT>' <EOL> self . dfa_server . subnet_create_event ( subnet_info ) <EOL> self . assertTrue ( self . dfa_server . dcnm_client . create_network . called ) <EOL> create_call = self . dfa_server . dcnm_client . create_network . call_args <EOL> arg1 , arg2 = create_call <EOL> self . assertTrue ( arg1 [ <NUM_LIT:0> ] == FAKE_PROJECT_NAME ) <EOL> self . assertTrue ( <EOL> arg1 [ <NUM_LIT:1> ] . __dict__ == self . dfa_server . network [ FAKE_NETWORK_ID ] ) <EOL> self . assertTrue ( <EOL> arg1 [ <NUM_LIT:2> ] . __dict__ == self . dfa_server . subnet [ FAKE_SUBNET_ID ] ) <EOL> def test_network_delete_event ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _load_network_info ( ) <EOL> self . assertFalse ( self . segid in self . dfa_server . segmentation_pool ) <EOL> network_info = { '<STR_LIT>' : FAKE_NETWORK_ID } <EOL> self . dfa_server . get_vms . return_value = [ ] <EOL> self . dfa_server . network_delete_event ( network_info ) <EOL> self . assertTrue ( self . dfa_server . dcnm_client . delete_network . called ) <EOL> dcall = self . dfa_server . dcnm_client . delete_network . call_args <EOL> arg1 , arg2 = dcall <EOL> self . assertTrue ( arg1 [ <NUM_LIT:0> ] == FAKE_PROJECT_NAME ) <EOL> self . assertTrue ( arg1 [ <NUM_LIT:1> ] . name == FAKE_NETWORK_NAME ) <EOL> self . assertTrue ( arg1 [ <NUM_LIT:1> ] . segmentation_id == self . segid ) <EOL> self . assertTrue ( self . segid in self . dfa_server . segmentation_pool ) <EOL> self . assertTrue ( self . dfa_server . delete_network_db . called ) <EOL> def test_dcnm_network_create_event ( self ) : <EOL> """<STR_LIT>""" <EOL> network_info = { '<STR_LIT>' : FAKE_SEG_ID , <EOL> '<STR_LIT>' : FAKE_PROJECT_NAME , <EOL> '<STR_LIT>' : self . part_name } <EOL> self . dfa_server . get_network_by_segid . return_value = None <EOL> self . dfa_server . get_project_id . return_value = FAKE_PROJECT_ID <EOL> dcnm_network = { '<STR_LIT>' : FAKE_SEG_ID , <EOL> '<STR_LIT>' : FAKE_CFG_PROFILE_NAME , <EOL> '<STR_LIT>' : FAKE_NETWORK_NAME , <EOL> '<STR_LIT>' : FAKE_PROJECT_NAME , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : FAKE_GW_ADDR } <EOL> self . dfa_server . dcnm_client . get_network . return_value = dcnm_network <EOL> dcnmclnt = self . dfa_server . dcnm_client <EOL> dcnmclnt . config_profile_fwding_mode_get . return_value = FAKE_FWD_MODE <EOL> self . dfa_server . dcnm_network_create_event ( network_info ) <EOL> self . dfa_server . dcnm_client . get_network . assert_called_with ( <EOL> FAKE_PROJECT_NAME , FAKE_SEG_ID ) <EOL> for netid , dcnmnet in six . iteritems ( self . dfa_server . network ) : <EOL> self . dfa_server . add_network_db . assert_called_with ( <EOL> netid , dcnmnet , '<STR_LIT>' , constants . RESULT_SUCCESS ) <EOL> self . assertTrue ( self . dfa_server . neutronclient . create_network . called ) <EOL> net_ext_name = self . cfg . dcnm . dcnm_net_ext <EOL> call_args = self . dfa_server . neutronclient . create_network . call_args <EOL> cargs , ckwargs = call_args <EOL> net_name = ckwargs . get ( '<STR_LIT:body>' ) . get ( '<STR_LIT>' ) . get ( '<STR_LIT:name>' ) <EOL> self . assertTrue ( net_name == ( <EOL> FAKE_NETWORK_NAME + net_ext_name + str ( FAKE_SEG_ID ) ) ) <EOL> self . assertTrue ( self . dfa_server . neutronclient . create_subnet . called ) <EOL> def test_dcnm_network_delete_event ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _load_network_info ( ) <EOL> network_info = { '<STR_LIT>' : ( <EOL> self . dfa_server . network [ FAKE_NETWORK_ID ] [ '<STR_LIT>' ] ) } <EOL> dcnmnet = mock . Mock ( ) <EOL> dcnmnet . network_id = FAKE_NETWORK_ID <EOL> self . dfa_server . get_network_by_segid . return_value = dcnmnet <EOL> self . dfa_server . dcnm_network_delete_event ( network_info ) <EOL> self . assertTrue ( self . dfa_server . network == { } ) <EOL> self . dfa_server . neutronclient . delete_network . assert_called_with ( <EOL> FAKE_NETWORK_ID ) <EOL> self . dfa_server . delete_network_db . assert_called_with ( FAKE_NETWORK_ID ) <EOL> def test_port_create_event ( self ) : <EOL> """<STR_LIT>""" <EOL> port_info = self . _get_port_info ( ) <EOL> self . _load_network_info ( ) <EOL> self . dfa_server . _inst_api . get_instance_for_uuid . return_value = ( <EOL> FAKE_INSTANCE_NAME ) <EOL> self . dfa_server . dcnm_dhcp = True <EOL> self . dfa_server . port_create_event ( port_info ) <EOL> self . assertTrue ( self . dfa_server . neutron_event . send_vm_info . called ) <EOL> call_args = self . dfa_server . neutron_event . send_vm_info . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertTrue ( cargs [ <NUM_LIT:0> ] == FAKE_HOST_ID ) <EOL> self . assertTrue ( str ( self . dfa_server . port [ FAKE_PORT_ID ] ) == cargs [ <NUM_LIT:1> ] ) <EOL> self . assertTrue ( self . dfa_server . add_vms_db . called ) <EOL> call_args = self . dfa_server . add_vms_db . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertTrue ( self . dfa_server . port [ FAKE_PORT_ID ] == cargs [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( constants . RESULT_SUCCESS == cargs [ <NUM_LIT:1> ] ) <EOL> def test_port_update_event ( self ) : <EOL> """<STR_LIT>""" <EOL> port_info = self . _get_port_info ( ) <EOL> mvm = mock . Mock ( ) <EOL> mvm . host = None <EOL> mvm . port_id = FAKE_PORT_ID <EOL> self . dfa_server . get_vm . return_value = mvm <EOL> self . dfa_server . _inst_api . get_instance_for_uuid . return_value = ( <EOL> FAKE_INSTANCE_NAME ) <EOL> self . dfa_server . port_update_event ( port_info ) <EOL> self . dfa_server . neutron_event . send_vm_info . assert_called_with ( <EOL> port_info [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] , <EOL> str ( self . dfa_server . port [ port_info [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ] ) ) <EOL> params = dict ( columns = dict ( <EOL> instance_id = FAKE_DEVICE_ID . replace ( '<STR_LIT:->' , '<STR_LIT>' ) , <EOL> host = port_info [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] , <EOL> result = constants . RESULT_SUCCESS , <EOL> name = FAKE_INSTANCE_NAME ) ) <EOL> self . dfa_server . update_vm_db . assert_called_with ( <EOL> port_info [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] , ** params ) <EOL> def test_port_delete_event ( self ) : <EOL> """<STR_LIT>""" <EOL> vm = mock . Mock ( ) <EOL> vm . mac = FAKE_MAC_ADDR <EOL> vm . port_id = FAKE_PORT_ID <EOL> vm . segmentation_id = self . segid <EOL> vm . network_id = FAKE_NETWORK_ID , <EOL> vm . port_id = FAKE_PORT_ID <EOL> vm . ip = FAKE_IP_ADDR <EOL> vm . gw_mac = FAKE_GW_ADDR <EOL> vm . instance_id = FAKE_DEVICE_ID <EOL> vm . fwd_mod = FAKE_FWD_MODE <EOL> vm . host = FAKE_HOST_ID <EOL> vm . name = FAKE_INSTANCE_NAME <EOL> self . dfa_server . get_vm . return_value = vm <EOL> vm_info = dict ( status = '<STR_LIT>' , vm_mac = vm . mac , <EOL> segmentation_id = vm . segmentation_id , <EOL> host = vm . host , port_uuid = vm . port_id , <EOL> net_uuid = vm . network_id , <EOL> oui = dict ( ip_addr = vm . ip , vm_name = vm . name , <EOL> vm_uuid = vm . instance_id , gw_mac = vm . gw_mac , <EOL> fwd_mod = vm . fwd_mod , oui_id = '<STR_LIT>' ) ) <EOL> port_info = { '<STR_LIT>' : FAKE_PORT_ID } <EOL> self . dfa_server . port_delete_event ( port_info ) <EOL> self . assertTrue ( self . dfa_server . neutron_event . send_vm_info . called ) <EOL> call_args = self . dfa_server . neutron_event . send_vm_info . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertTrue ( cargs [ <NUM_LIT:0> ] == FAKE_HOST_ID ) <EOL> self . assertTrue ( str ( vm_info ) == cargs [ <NUM_LIT:1> ] ) <EOL> self . dfa_server . delete_vm_db . assert_called_with ( vm . instance_id ) <EOL> def test_add_dhcp_port ( self ) : <EOL> """<STR_LIT>""" <EOL> self . dfa_server . get_vm . return_value = None <EOL> port_info = self . _get_port_info ( ) . get ( "<STR_LIT:port>" ) <EOL> self . _load_network_info ( ) <EOL> self . dfa_server . _inst_api . get_instance_for_uuid . return_value = ( <EOL> FAKE_INSTANCE_NAME ) <EOL> self . dfa_server . dcnm_dhcp = False <EOL> self . dfa_server . add_dhcp_port ( port_info ) <EOL> self . assertTrue ( self . dfa_server . neutron_event . send_vm_info . called ) <EOL> call_args = self . dfa_server . neutron_event . send_vm_info . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertEqual ( FAKE_HOST_ID , cargs [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( str ( self . dfa_server . port [ FAKE_PORT_ID ] ) , cargs [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( self . dfa_server . port [ FAKE_PORT_ID ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( self . dfa_server . add_vms_db . called ) <EOL> call_args = self . dfa_server . add_vms_db . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertEqual ( self . dfa_server . port [ FAKE_PORT_ID ] , cargs [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( constants . RESULT_SUCCESS , cargs [ <NUM_LIT:1> ] ) <EOL> def test_correct_dhcp_ports ( self ) : <EOL> """<STR_LIT>""" <EOL> port_info = self . _get_port_info ( ) . get ( "<STR_LIT:port>" ) <EOL> port_info [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> ports_list_data = { "<STR_LIT>" : [ port_info ] } <EOL> self . dfa_server . neutronclient . list_ports . return_value = ports_list_data <EOL> self . dfa_server . neutron_event . _clients . get . return_value = True <EOL> self . dfa_server . add_dhcp_port = mock . Mock ( ) <EOL> self . dfa_server . correct_dhcp_ports ( FAKE_NETWORK_ID ) <EOL> self . assertTrue ( self . dfa_server . add_dhcp_port . called ) <EOL> call_args = self . dfa_server . add_dhcp_port . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertEqual ( FAKE_PORT_ID , cargs [ <NUM_LIT:0> ] . get ( "<STR_LIT:id>" ) ) <EOL> def test_strip_wait_dhcp ( self ) : <EOL> vm = mock . Mock <EOL> vm . mac = FAKE_MAC_ADDR <EOL> vm . port_id = FAKE_PORT_ID <EOL> vm . segmentation_id = self . segid <EOL> vm . network_id = FAKE_NETWORK_ID , <EOL> vm . port_id = FAKE_PORT_ID <EOL> vm . ip = "<STR_LIT>" <EOL> vm . gw_mac = FAKE_GW_ADDR <EOL> vm . instance_id = FAKE_DEVICE_ID <EOL> vm . fwd_mod = FAKE_FWD_MODE <EOL> vm . host = FAKE_HOST_ID <EOL> vm . name = FAKE_INSTANCE_NAME <EOL> self . dfa_server . strip_wait_dhcp ( vm ) <EOL> self . assertTrue ( self . dfa_server . update_vm_db . called ) <EOL> call_args = self . dfa_server . update_vm_db . call_args <EOL> cargs , ckwargs = call_args <EOL> self . assertEqual ( FAKE_PORT_ID , cargs [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( "<STR_LIT>" , ckwargs . get ( "<STR_LIT>" ) . get ( "<STR_LIT>" ) ) </s>
