<s> import mock <EOL> from neutron_lib import constants as n_const <EOL> from neutron . tests . unit . plugins . ml2 . drivers . l2pop . rpc_manager import l2population_rpc_base <EOL> class TestL2populationRpcCallBackTunnelMixin ( <EOL> l2population_rpc_base . TestL2populationRpcCallBackTunnelMixinBase ) : <EOL> def test_get_agent_ports_no_data ( self ) : <EOL> self . assertFalse ( <EOL> list ( self . fakeagent . get_agent_ports ( self . fdb_entries1 , { } ) ) ) <EOL> def test_get_agent_ports_non_existence_key_in_lvm ( self ) : <EOL> results = { } <EOL> del self . local_vlan_map1 [ self . lvms [ <NUM_LIT:1> ] . net ] <EOL> for lvm , agent_ports in self . fakeagent . get_agent_ports ( <EOL> self . fdb_entries1 , self . local_vlan_map1 ) : <EOL> results [ lvm ] = agent_ports <EOL> expected = { <EOL> self . lvm1 : { <EOL> self . ports [ <NUM_LIT:0> ] . ip : [ ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) ] , <EOL> self . local_ip : [ ] } , <EOL> self . lvm3 : { <EOL> self . ports [ <NUM_LIT:2> ] . ip : [ ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) ] , <EOL> self . local_ip : [ ] } , <EOL> } <EOL> self . assertEqual ( expected , results ) <EOL> def test_get_agent_ports_no_agent_ports ( self ) : <EOL> results = { } <EOL> self . fdb_entries1 [ self . lvms [ <NUM_LIT:1> ] . net ] [ '<STR_LIT>' ] = { } <EOL> for lvm , agent_ports in self . fakeagent . get_agent_ports ( <EOL> self . fdb_entries1 , self . local_vlan_map1 ) : <EOL> results [ lvm ] = agent_ports <EOL> expected = { <EOL> self . lvm1 : { <EOL> self . ports [ <NUM_LIT:0> ] . ip : [ ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) ] , <EOL> self . local_ip : [ ] } , <EOL> self . lvm2 : { } , <EOL> self . lvm3 : { <EOL> self . ports [ <NUM_LIT:2> ] . ip : [ ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) ] , <EOL> self . local_ip : [ ] } , <EOL> } <EOL> self . assertEqual ( expected , results ) <EOL> def test_fdb_add_tun ( self ) : <EOL> with mock . patch . object ( self . fakeagent , '<STR_LIT>' ) , mock . patch . object ( self . fakeagent , '<STR_LIT>' <EOL> ) as mock_add_fdb_flow : <EOL> self . fakeagent . fdb_add_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:1> ] . mac , self . lvms [ <NUM_LIT:1> ] . ip ) , <EOL> self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:1> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_add_fdb_flow . call_args_list ) ) <EOL> def test_fdb_add_tun_non_existence_key_in_ofports ( self ) : <EOL> ofport = self . lvm1 . network_type + '<STR_LIT>' <EOL> del self . ofports [ self . type_gre ] [ self . ports [ <NUM_LIT:1> ] . ip ] <EOL> with mock . patch . object ( self . fakeagent , '<STR_LIT>' , <EOL> return_value = ofport <EOL> ) as mock_setup_tunnel_port , mock . patch . object ( self . fakeagent , '<STR_LIT>' <EOL> ) as mock_add_fdb_flow : <EOL> self . fakeagent . fdb_add_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> mock_setup_tunnel_port . assert_called_once_with ( <EOL> self . fakebr , self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 . network_type ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:1> ] . mac , self . lvms [ <NUM_LIT:1> ] . ip ) , <EOL> self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 , ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_add_fdb_flow . call_args_list ) ) <EOL> def test_fdb_add_tun_unavailable_ofport ( self ) : <EOL> del self . ofports [ self . type_gre ] [ self . ports [ <NUM_LIT:1> ] . ip ] <EOL> with mock . patch . object ( self . fakeagent , '<STR_LIT>' , <EOL> return_value = <NUM_LIT:0> <EOL> ) as mock_setup_tunnel_port , mock . patch . object ( self . fakeagent , '<STR_LIT>' <EOL> ) as mock_add_fdb_flow : <EOL> self . fakeagent . fdb_add_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> mock_setup_tunnel_port . assert_called_once_with ( <EOL> self . fakebr , self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 . network_type ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_add_fdb_flow . call_args_list ) ) <EOL> def test_fdb_remove_tun ( self ) : <EOL> with mock . patch . object ( <EOL> self . fakeagent , '<STR_LIT>' ) as mock_del_fdb_flow : <EOL> self . fakeagent . fdb_remove_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:1> ] . mac , self . lvms [ <NUM_LIT:1> ] . ip ) , <EOL> self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:1> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_del_fdb_flow . call_args_list ) ) <EOL> def test_fdb_remove_tun_flooding_entry ( self ) : <EOL> self . agent_ports [ self . ports [ <NUM_LIT:1> ] . ip ] = [ n_const . FLOODING_ENTRY ] <EOL> with mock . patch . object ( self . fakeagent , '<STR_LIT>' <EOL> ) as mock_del_fdb_flow , mock . patch . object ( self . fakeagent , '<STR_LIT>' <EOL> ) as mock_cleanup_tunnel_port : <EOL> self . fakeagent . fdb_remove_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , <EOL> ( n_const . FLOODING_ENTRY [ <NUM_LIT:0> ] , n_const . FLOODING_ENTRY [ <NUM_LIT:1> ] ) , <EOL> self . ports [ <NUM_LIT:1> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:1> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_del_fdb_flow . call_args_list ) ) <EOL> mock_cleanup_tunnel_port . assert_called_once_with ( <EOL> self . fakebr , self . ports [ <NUM_LIT:1> ] . ofport , self . lvm1 . network_type ) <EOL> def test_fdb_remove_tun_non_existence_key_in_ofports ( self ) : <EOL> del self . ofports [ self . type_gre ] [ self . ports [ <NUM_LIT:1> ] . ip ] <EOL> with mock . patch . object ( <EOL> self . fakeagent , '<STR_LIT>' ) as mock_del_fdb_flow : <EOL> self . fakeagent . fdb_remove_tun ( '<STR_LIT>' , self . fakebr , self . lvm1 , <EOL> self . agent_ports , <EOL> self . _tunnel_port_lookup ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> self . ports [ <NUM_LIT:0> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:0> ] . ofport ) , <EOL> mock . call ( self . fakebr , ( self . lvms [ <NUM_LIT:2> ] . mac , self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> self . ports [ <NUM_LIT:2> ] . ip , self . lvm1 , self . ports [ <NUM_LIT:2> ] . ofport ) , <EOL> ] <EOL> self . assertEqual ( sorted ( expected ) , <EOL> sorted ( mock_del_fdb_flow . call_args_list ) ) <EOL> def test_fdb_update ( self ) : <EOL> fake__fdb_chg_ip = mock . Mock ( ) <EOL> self . fakeagent . _fdb_chg_ip = fake__fdb_chg_ip <EOL> self . fakeagent . fdb_update ( '<STR_LIT>' , self . upd_fdb_entry1 ) <EOL> fake__fdb_chg_ip . assert_called_once_with ( <EOL> '<STR_LIT>' , self . upd_fdb_entry1_val ) <EOL> def test_fdb_update_non_existence_method ( self ) : <EOL> self . assertRaises ( NotImplementedError , <EOL> self . fakeagent . fdb_update , <EOL> '<STR_LIT>' , self . upd_fdb_entry1 ) <EOL> def test__fdb_chg_ip ( self ) : <EOL> m_setup_entry_for_arp_reply = mock . Mock ( ) <EOL> self . fakeagent . setup_entry_for_arp_reply = m_setup_entry_for_arp_reply <EOL> self . fakeagent . fdb_chg_ip_tun ( '<STR_LIT>' , self . fakebr , <EOL> self . upd_fdb_entry1_val , self . local_ip , <EOL> self . local_vlan_map1 ) <EOL> expected = [ <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm1 . vlan , self . lvms [ <NUM_LIT:0> ] . mac , <EOL> self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm1 . vlan , self . lvms [ <NUM_LIT:1> ] . mac , <EOL> self . lvms [ <NUM_LIT:1> ] . ip ) , <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm1 . vlan , self . lvms [ <NUM_LIT:0> ] . mac , <EOL> self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm1 . vlan , self . lvms [ <NUM_LIT:1> ] . mac , <EOL> self . lvms [ <NUM_LIT:1> ] . ip ) , <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm2 . vlan , self . lvms [ <NUM_LIT:0> ] . mac , <EOL> self . lvms [ <NUM_LIT:0> ] . ip ) , <EOL> mock . call ( self . fakebr , '<STR_LIT>' , self . lvm2 . vlan , self . lvms [ <NUM_LIT:2> ] . mac , <EOL> self . lvms [ <NUM_LIT:2> ] . ip ) , <EOL> ] <EOL> m_setup_entry_for_arp_reply . assert_has_calls ( expected , any_order = True ) <EOL> def test__fdb_chg_ip_no_lvm ( self ) : <EOL> m_setup_entry_for_arp_reply = mock . Mock ( ) <EOL> self . fakeagent . setup_entry_for_arp_reply = m_setup_entry_for_arp_reply <EOL> self . fakeagent . fdb_chg_ip_tun ( <EOL> '<STR_LIT>' , self . fakebr , self . upd_fdb_entry1 , self . local_ip , { } ) <EOL> self . assertFalse ( m_setup_entry_for_arp_reply . call_count ) <EOL> def test__fdb_chg_ip_ip_is_local_ip ( self ) : <EOL> upd_fdb_entry_val = { <EOL> self . lvms [ <NUM_LIT:0> ] . net : { <EOL> self . local_ip : { <EOL> '<STR_LIT>' : [ ( self . lvms [ <NUM_LIT:0> ] . mac , self . lvms [ <NUM_LIT:0> ] . ip ) ] , <EOL> '<STR_LIT>' : [ ( self . lvms [ <NUM_LIT:1> ] . mac , self . lvms [ <NUM_LIT:1> ] . ip ) ] , <EOL> } , <EOL> } , <EOL> } <EOL> m_setup_entry_for_arp_reply = mock . Mock ( ) <EOL> self . fakeagent . setup_entry_for_arp_reply = m_setup_entry_for_arp_reply <EOL> self . fakeagent . fdb_chg_ip_tun ( '<STR_LIT>' , self . fakebr , <EOL> upd_fdb_entry_val , self . local_ip , <EOL> self . local_vlan_map1 ) <EOL> self . assertFalse ( m_setup_entry_for_arp_reply . call_count ) <EOL> def test_fdb_chg_ip_tun_empty_before_after ( self ) : <EOL> upd_fdb_entry_val = { <EOL> self . lvms [ <NUM_LIT:0> ] . net : { <EOL> self . local_ip : { } , <EOL> } , <EOL> } <EOL> m_setup_entry_for_arp_reply = mock . Mock ( ) <EOL> self . fakeagent . setup_entry_for_arp_reply = m_setup_entry_for_arp_reply <EOL> self . fakeagent . fdb_chg_ip_tun ( '<STR_LIT>' , self . fakebr , <EOL> upd_fdb_entry_val , "<STR_LIT>" , <EOL> self . local_vlan_map1 ) <EOL> self . assertFalse ( m_setup_entry_for_arp_reply . call_count ) </s>
