<s> import mock <EOL> import netaddr <EOL> import neutron . plugins . ml2 . drivers . openvswitch . agent . common . constants as ovs_const <EOL> from neutron . tests . unit . plugins . ml2 . drivers . openvswitch . agent . openflow . ovs_ofctl import ovs_bridge_test_base <EOL> call = mock . call <EOL> class OVSTunnelBridgeTest ( ovs_bridge_test_base . OVSBridgeTestBase , <EOL> ovs_bridge_test_base . OVSDVRProcessTestMixin ) : <EOL> dvr_process_table_id = ovs_const . DVR_PROCESS <EOL> dvr_process_next_table_id = ovs_const . PATCH_LV_TO_TUN <EOL> def setUp ( self ) : <EOL> super ( OVSTunnelBridgeTest , self ) . setUp ( ) <EOL> self . setup_bridge_mock ( '<STR_LIT>' , self . br_tun_cls ) <EOL> self . stamp = self . br . default_cookie <EOL> def test_setup_default_table ( self ) : <EOL> patch_int_ofport = <NUM_LIT> <EOL> mock_do_action_flows = mock . patch . object ( self . br , <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . mock . attach_mock ( mock_do_action_flows , '<STR_LIT>' ) <EOL> self . br . setup_default_table ( patch_int_ofport = patch_int_ofport , <EOL> arp_responder_enabled = False ) <EOL> flow_args = [ { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : patch_int_ofport , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:4> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:6> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : '<STR_LIT>' + str ( self . stamp ) + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % patch_int_ofport } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:20> , '<STR_LIT>' : '<STR_LIT>' } <EOL> ] <EOL> expected = [ call . do_action_flows ( '<STR_LIT>' , flow_args ) , <EOL> call . add_flow ( priority = <NUM_LIT:0> , table = <NUM_LIT> , actions = '<STR_LIT>' ) ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_setup_default_table_arp_responder_enabled ( self ) : <EOL> patch_int_ofport = <NUM_LIT> <EOL> mock_do_action_flows = mock . patch . object ( self . br , <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . mock . attach_mock ( mock_do_action_flows , '<STR_LIT>' ) <EOL> self . br . setup_default_table ( patch_int_ofport = patch_int_ofport , <EOL> arp_responder_enabled = True ) <EOL> flow_args = [ { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : patch_int_ofport , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:4> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:6> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : '<STR_LIT>' + str ( self . stamp ) + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % patch_int_ofport } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:20> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : '<STR_LIT>' } <EOL> ] <EOL> expected = [ call . do_action_flows ( '<STR_LIT>' , flow_args ) , <EOL> call . add_flow ( priority = <NUM_LIT:0> , table = <NUM_LIT> , actions = '<STR_LIT>' ) ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_provision_local_vlan ( self ) : <EOL> network_type = '<STR_LIT>' <EOL> lvid = <NUM_LIT> <EOL> segmentation_id = <NUM_LIT> <EOL> distributed = False <EOL> self . br . provision_local_vlan ( network_type = network_type , lvid = lvid , <EOL> segmentation_id = segmentation_id , <EOL> distributed = distributed ) <EOL> expected = [ <EOL> call . add_flow ( priority = <NUM_LIT:1> , tun_id = segmentation_id , <EOL> actions = '<STR_LIT>' % lvid , <EOL> table = <NUM_LIT:4> ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_reclaim_local_vlan ( self ) : <EOL> network_type = '<STR_LIT>' <EOL> segmentation_id = <NUM_LIT> <EOL> self . br . reclaim_local_vlan ( network_type = network_type , <EOL> segmentation_id = segmentation_id ) <EOL> expected = [ <EOL> call . delete_flows ( tun_id = segmentation_id , table = <NUM_LIT:4> ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_install_flood_to_tun ( self ) : <EOL> vlan = <NUM_LIT> <EOL> tun_id = <NUM_LIT> <EOL> ports = [ <NUM_LIT:11> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> self . br . install_flood_to_tun ( vlan = vlan , <EOL> tun_id = tun_id , <EOL> ports = ports ) <EOL> expected = [ <EOL> call . mod_flow ( table = <NUM_LIT> , dl_vlan = vlan , <EOL> actions = '<STR_LIT>' <EOL> '<STR_LIT>' % { <EOL> '<STR_LIT>' : tun_id , <EOL> '<STR_LIT>' : '<STR_LIT:U+002C>' . join ( map ( str , ports ) ) , <EOL> } ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_delete_flood_to_tun ( self ) : <EOL> vlan = <NUM_LIT> <EOL> self . br . delete_flood_to_tun ( vlan = vlan ) <EOL> expected = [ <EOL> call . delete_flows ( table = <NUM_LIT> , dl_vlan = vlan ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_install_unicast_to_tun ( self ) : <EOL> vlan = <NUM_LIT> <EOL> port = <NUM_LIT> <EOL> mac = '<STR_LIT>' <EOL> tun_id = <NUM_LIT> <EOL> self . br . install_unicast_to_tun ( vlan = vlan , <EOL> tun_id = tun_id , <EOL> port = port , <EOL> mac = mac ) <EOL> expected = [ <EOL> call . add_flow ( priority = <NUM_LIT:2> , table = <NUM_LIT:20> , dl_dst = mac , dl_vlan = vlan , <EOL> actions = '<STR_LIT>' <EOL> '<STR_LIT>' % { <EOL> '<STR_LIT>' : tun_id , <EOL> '<STR_LIT:port>' : port , <EOL> } ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_delete_unicast_to_tun ( self ) : <EOL> vlan = <NUM_LIT> <EOL> mac = '<STR_LIT>' <EOL> self . br . delete_unicast_to_tun ( vlan = vlan , mac = mac ) <EOL> expected = [ <EOL> call . delete_flows ( table = <NUM_LIT:20> , dl_dst = mac , dl_vlan = vlan ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_delete_unicast_to_tun_without_mac ( self ) : <EOL> vlan = <NUM_LIT> <EOL> mac = None <EOL> self . br . delete_unicast_to_tun ( vlan = vlan , mac = mac ) <EOL> expected = [ <EOL> call . delete_flows ( table = <NUM_LIT:20> , dl_vlan = vlan ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_install_arp_responder ( self ) : <EOL> vlan = <NUM_LIT> <EOL> ip = '<STR_LIT>' <EOL> mac = '<STR_LIT>' <EOL> self . br . install_arp_responder ( vlan = vlan , ip = ip , mac = mac ) <EOL> expected = [ <EOL> call . add_flow ( proto = '<STR_LIT>' , nw_dst = ip , <EOL> actions = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % { <EOL> '<STR_LIT>' : netaddr . EUI ( mac , <EOL> dialect = netaddr . mac_unix ) , <EOL> '<STR_LIT>' : netaddr . IPAddress ( ip ) , <EOL> } , <EOL> priority = <NUM_LIT:1> , table = <NUM_LIT> , dl_vlan = vlan ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_delete_arp_responder ( self ) : <EOL> vlan = <NUM_LIT> <EOL> ip = '<STR_LIT>' <EOL> self . br . delete_arp_responder ( vlan = vlan , ip = ip ) <EOL> expected = [ <EOL> call . delete_flows ( table = <NUM_LIT> , dl_vlan = vlan , proto = '<STR_LIT>' , nw_dst = ip ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_delete_arp_responder_without_ip ( self ) : <EOL> vlan = <NUM_LIT> <EOL> ip = None <EOL> self . br . delete_arp_responder ( vlan = vlan , ip = ip ) <EOL> expected = [ <EOL> call . delete_flows ( table = <NUM_LIT> , dl_vlan = vlan , proto = '<STR_LIT>' ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_setup_tunnel_port ( self ) : <EOL> network_type = '<STR_LIT>' <EOL> port = <NUM_LIT> <EOL> self . br . setup_tunnel_port ( network_type = network_type , port = port ) <EOL> expected = [ <EOL> call . add_flow ( priority = <NUM_LIT:1> , in_port = port , actions = '<STR_LIT>' ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_cleanup_tunnel_port ( self ) : <EOL> port = <NUM_LIT> <EOL> self . br . cleanup_tunnel_port ( port = port ) <EOL> expected = [ <EOL> call . delete_flows ( in_port = port ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_add_dvr_mac_tun ( self ) : <EOL> mac = '<STR_LIT>' <EOL> port = <NUM_LIT> <EOL> self . br . add_dvr_mac_tun ( mac = mac , port = port ) <EOL> expected = [ <EOL> call . add_flow ( priority = <NUM_LIT:1> , table = <NUM_LIT:9> , dl_src = mac , <EOL> actions = '<STR_LIT>' % port ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def test_remove_dvr_mac_tun ( self ) : <EOL> mac = '<STR_LIT>' <EOL> self . br . remove_dvr_mac_tun ( mac = mac ) <EOL> expected = [ <EOL> call . delete_flows ( eth_src = mac , table_id = <NUM_LIT:9> ) , <EOL> ] <EOL> self . assertEqual ( expected , self . mock . mock_calls ) <EOL> def _mock_add_tunnel_port ( self , deferred_br = False ) : <EOL> port_name = '<STR_LIT>' <EOL> remote_ip = '<STR_LIT>' <EOL> local_ip = '<STR_LIT>' <EOL> tunnel_type = '<STR_LIT>' <EOL> vxlan_udp_port = '<STR_LIT>' <EOL> dont_fragment = True <EOL> if deferred_br : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = <NUM_LIT> ) as add_port , self . br . deferred ( ) as deferred_br : <EOL> ofport = deferred_br . add_tunnel_port ( port_name , remote_ip , <EOL> local_ip , tunnel_type , <EOL> vxlan_udp_port , <EOL> dont_fragment ) <EOL> else : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> return_value = <NUM_LIT> ) as add_port : <EOL> ofport = self . br . add_tunnel_port ( port_name , remote_ip , <EOL> local_ip , tunnel_type , <EOL> vxlan_udp_port , <EOL> dont_fragment ) <EOL> self . assertEqual ( <NUM_LIT> , ofport ) <EOL> self . assertEqual ( <NUM_LIT:1> , add_port . call_count ) <EOL> self . assertEqual ( port_name , add_port . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> def _mock_delete_port ( self , deferred_br = False ) : <EOL> port_name = '<STR_LIT>' <EOL> if deferred_br : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as delete_port , self . br . deferred ( ) as deferred_br : <EOL> deferred_br . delete_port ( port_name ) <EOL> else : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as delete_port : <EOL> self . br . delete_port ( port_name ) <EOL> self . assertEqual ( [ call ( port_name ) ] , delete_port . mock_calls ) <EOL> def test_add_tunnel_port ( self ) : <EOL> self . _mock_add_tunnel_port ( ) <EOL> def test_delete_port ( self ) : <EOL> self . _mock_delete_port ( ) <EOL> def test_deferred_br_add_tunnel_port ( self ) : <EOL> self . _mock_add_tunnel_port ( True ) <EOL> def test_deferred_br_delete_port ( self ) : <EOL> self . _mock_delete_port ( True ) </s>
