#NEWLINE# def test_dvr_handle_new_service_port( self ) : #NEWLINE# #INDENT# port = { #NEWLINE# #INDENT# 'id' : 'port1' , #NEWLINE# 'device_id' : 'abcd' , #NEWLINE# 'device_owner' : DEVICE_OWNER_COMPUTE_NOVA , #NEWLINE# portbindings . HOST_ID : 'host1' , #NEWLINE# 'fixed_ips' : [ { #NEWLINE# #INDENT# 'subnet_id' : '80947d4a-fbc8-484b-9f92-623a6bfcf3e0' , #NEWLINE# 'ip_address' : '10.10.10.3' , #NEWLINE# #UNINDENT# } ] , #NEWLINE# #UNINDENT# } #NEWLINE# dvr_ports = [ { #NEWLINE# #INDENT# 'id' : 'dvr_port1' , #NEWLINE# 'device_id' : 'r1' , #NEWLINE# 'device_owner' : constants . DEVICE_OWNER_DVR_INTERFACE , #NEWLINE# 'fixed_ips' : [ { #NEWLINE# #INDENT# 'subnet_id' : '80947d4a-fbc8-484b-9f92-623a6bfcf3e0' , #NEWLINE# 'ip_address' : '10.10.10.1' , #NEWLINE# #UNINDENT# } ] , #NEWLINE# #UNINDENT# } , { #NEWLINE# #INDENT# 'id' : 'dvr_port2' , #NEWLINE# 'device_id' : 'r2' , #NEWLINE# 'device_owner' : constants . DEVICE_OWNER_DVR_INTERFACE , #NEWLINE# 'fixed_ips' : [ { #NEWLINE# #INDENT# 'subnet_id' : '80947d4a-fbc8-484b-9f92-623a6bfcf3e0' , #NEWLINE# 'ip_address' : '10.10.10.123' , #NEWLINE# #UNINDENT# } ] , #NEWLINE# #UNINDENT# } ] #NEWLINE# agent_on_host = { #NEWLINE# #INDENT# 'id' : 'agent1' , #NEWLINE# #UNINDENT# } #NEWLINE# with mock . patch ( 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_ports' , return_value = dvr_ports ) , mock . patch ( 'neutron.api.rpc.agentnotifiers.l3_rpc_agent_api.L3AgentNotifyAPI' ) , mock . patch . object ( self . dut , 'get_l3_agents' , return_value = [ get_l3_agents ] ) as get_l3_agents : #NEWLINE# #INDENT# self . dut . dvr_handle_new_service_port ( self . adminContext , port ) #NEWLINE# get_l3_agents . assert_called_once_with ( self . adminContext , filters = { #NEWLINE# #INDENT# 'host' : [ port [ portbindings . HOST_ID ] ] , #NEWLINE# #UNINDENT# } ) #NEWLINE# self . dut . l3_rpc_notifier . routers_updated_on_host . assert_called_once_with ( self . adminContext , { 'r1' , 'r2' } , 'host1' ) #NEWLINE# self . assertFalse ( self . dut . l3_rpc_notifier . routers_updated . called )