<s> import collections <EOL> import uuid <EOL> import mock <EOL> from neutron . agent . common import ovs_lib <EOL> from neutron . agent . linux import ip_lib <EOL> from neutron . tests import base as tests_base <EOL> from neutron . tests . common import net_helpers <EOL> from neutron . tests . functional . agent . linux import base <EOL> class OVSBridgeTestBase ( base . BaseOVSLinuxTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( OVSBridgeTestBase , self ) . setUp ( ) <EOL> self . ovs = ovs_lib . BaseOVS ( ) <EOL> self . br = self . useFixture ( net_helpers . OVSBridgeFixture ( ) ) . bridge <EOL> def create_ovs_port ( self , * interface_attrs ) : <EOL> attrs = collections . OrderedDict ( interface_attrs ) <EOL> attrs . setdefault ( '<STR_LIT:type>' , '<STR_LIT>' ) <EOL> port_name = tests_base . get_rand_device_name ( net_helpers . PORT_PREFIX ) <EOL> return ( port_name , self . br . add_port ( port_name , * attrs . items ( ) ) ) <EOL> def create_ovs_vif_port ( self , iface_id = None , mac = None , <EOL> iface_field = '<STR_LIT>' ) : <EOL> if iface_id is None : <EOL> iface_id = base . get_rand_name ( ) <EOL> if mac is None : <EOL> mac = base . get_rand_name ( ) <EOL> attrs = ( '<STR_LIT>' , { iface_field : iface_id , '<STR_LIT>' : mac } ) <EOL> port_name , ofport = self . create_ovs_port ( attrs ) <EOL> return ovs_lib . VifPort ( port_name , ofport , iface_id , mac , self . br ) <EOL> class OVSBridgeTestCase ( OVSBridgeTestBase ) : <EOL> def test_port_lifecycle ( self ) : <EOL> ( port_name , ofport ) = self . create_ovs_port ( ( '<STR_LIT:type>' , '<STR_LIT>' ) ) <EOL> self . assertTrue ( int ( ofport ) ) <EOL> self . assertTrue ( int ( self . br . get_port_ofport ( port_name ) ) ) <EOL> self . assertTrue ( self . br . port_exists ( port_name ) ) <EOL> self . assertEqual ( self . br . br_name , <EOL> self . br . get_bridge_for_iface ( port_name ) ) <EOL> self . br . delete_port ( port_name ) <EOL> self . assertFalse ( self . br . port_exists ( port_name ) ) <EOL> def test_duplicate_port_may_exist_false ( self ) : <EOL> port_name , ofport = self . create_ovs_port ( ( '<STR_LIT:type>' , '<STR_LIT>' ) ) <EOL> cmd = self . br . ovsdb . add_port ( self . br . br_name , <EOL> port_name , may_exist = False ) <EOL> self . assertRaises ( RuntimeError , cmd . execute , check_error = True ) <EOL> def test_delete_port_if_exists_false ( self ) : <EOL> cmd = self . br . ovsdb . del_port ( '<STR_LIT>' , if_exists = False ) <EOL> self . assertRaises ( RuntimeError , cmd . execute , check_error = True ) <EOL> def test_replace_port ( self ) : <EOL> port_name = tests_base . get_rand_device_name ( net_helpers . PORT_PREFIX ) <EOL> self . br . replace_port ( port_name , ( '<STR_LIT:type>' , '<STR_LIT>' ) ) <EOL> self . assertTrue ( self . br . port_exists ( port_name ) ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . br . db_get_val ( '<STR_LIT>' , port_name , '<STR_LIT:type>' ) ) <EOL> self . br . replace_port ( port_name , ( '<STR_LIT:type>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:test>' : '<STR_LIT:test>' } ) ) <EOL> self . assertTrue ( self . br . port_exists ( port_name ) ) <EOL> self . assertEqual ( '<STR_LIT:test>' , self . br . db_get_val ( '<STR_LIT>' , port_name , <EOL> '<STR_LIT>' ) [ '<STR_LIT:test>' ] ) <EOL> def test_attribute_lifecycle ( self ) : <EOL> ( port_name , ofport ) = self . create_ovs_port ( ) <EOL> tag = <NUM_LIT> <EOL> self . ovs . set_db_attribute ( '<STR_LIT>' , port_name , '<STR_LIT>' , tag ) <EOL> self . assertEqual ( tag , self . ovs . db_get_val ( '<STR_LIT>' , port_name , '<STR_LIT>' ) ) <EOL> self . assertEqual ( tag , self . br . get_port_tag_dict ( ) [ port_name ] ) <EOL> self . ovs . clear_db_attribute ( '<STR_LIT>' , port_name , '<STR_LIT>' ) <EOL> self . assertEqual ( [ ] , self . ovs . db_get_val ( '<STR_LIT>' , port_name , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ ] , self . br . get_port_tag_dict ( ) [ port_name ] ) <EOL> def test_get_bridge_external_bridge_id ( self ) : <EOL> self . ovs . set_db_attribute ( '<STR_LIT>' , self . br . br_name , <EOL> '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . br . br_name } ) <EOL> self . assertEqual ( <EOL> self . br . br_name , <EOL> self . ovs . get_bridge_external_bridge_id ( self . br . br_name ) ) <EOL> def test_controller_lifecycle ( self ) : <EOL> controllers = { '<STR_LIT>' , '<STR_LIT>' } <EOL> self . br . set_controller ( controllers ) <EOL> self . assertSetEqual ( controllers , set ( self . br . get_controller ( ) ) ) <EOL> self . br . del_controller ( ) <EOL> self . assertEqual ( [ ] , self . br . get_controller ( ) ) <EOL> def test_non_index_queries ( self ) : <EOL> controllers = [ '<STR_LIT>' ] <EOL> self . br . set_controller ( controllers ) <EOL> cmd = self . br . ovsdb . db_set ( '<STR_LIT>' , self . br . br_name , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> cmd . execute ( check_error = True ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . br . db_get_val ( '<STR_LIT>' , self . br . br_name , <EOL> '<STR_LIT>' ) ) <EOL> def test_set_fail_mode_secure ( self ) : <EOL> self . br . set_secure_mode ( ) <EOL> self . _assert_br_fail_mode ( ovs_lib . FAILMODE_SECURE ) <EOL> def test_set_fail_mode_standalone ( self ) : <EOL> self . br . set_standalone_mode ( ) <EOL> self . _assert_br_fail_mode ( ovs_lib . FAILMODE_STANDALONE ) <EOL> def _assert_br_fail_mode ( self , fail_mode ) : <EOL> self . assertEqual ( <EOL> self . br . db_get_val ( '<STR_LIT>' , self . br . br_name , '<STR_LIT>' ) , <EOL> fail_mode ) <EOL> def test_set_protocols ( self ) : <EOL> self . br . set_protocols ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . br . db_get_val ( '<STR_LIT>' , self . br . br_name , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> def test_get_datapath_id ( self ) : <EOL> brdev = ip_lib . IPDevice ( self . br . br_name ) <EOL> dpid = brdev . link . attributes [ '<STR_LIT>' ] . replace ( '<STR_LIT::>' , '<STR_LIT>' ) <EOL> self . br . set_db_attribute ( '<STR_LIT>' , <EOL> self . br . br_name , '<STR_LIT>' , dpid ) <EOL> self . assertIn ( dpid , self . br . get_datapath_id ( ) ) <EOL> def test_add_tunnel_port ( self ) : <EOL> attrs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> port_name = tests_base . get_rand_device_name ( net_helpers . PORT_PREFIX ) <EOL> self . br . add_tunnel_port ( port_name , attrs [ '<STR_LIT>' ] , <EOL> attrs [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . ovs . db_get_val ( '<STR_LIT>' , port_name , '<STR_LIT:type>' ) , <EOL> '<STR_LIT>' ) <EOL> options = self . ovs . db_get_val ( '<STR_LIT>' , port_name , '<STR_LIT>' ) <EOL> for attr , val in attrs . items ( ) : <EOL> self . assertEqual ( val , options [ attr ] ) <EOL> def test_add_patch_port ( self ) : <EOL> local = tests_base . get_rand_device_name ( net_helpers . PORT_PREFIX ) <EOL> peer = '<STR_LIT>' <EOL> self . br . add_patch_port ( local , peer ) <EOL> self . assertEqual ( self . ovs . db_get_val ( '<STR_LIT>' , local , '<STR_LIT:type>' ) , <EOL> '<STR_LIT>' ) <EOL> options = self . ovs . db_get_val ( '<STR_LIT>' , local , '<STR_LIT>' ) <EOL> self . assertEqual ( peer , options [ '<STR_LIT>' ] ) <EOL> def test_get_port_name_list ( self ) : <EOL> ports = { self . create_ovs_port ( ) [ <NUM_LIT:0> ] for i in range ( <NUM_LIT:5> ) } <EOL> self . assertSetEqual ( ports , set ( self . br . get_port_name_list ( ) ) ) <EOL> def test_get_iface_name_list ( self ) : <EOL> ifaces = { self . create_ovs_port ( ) [ <NUM_LIT:0> ] for i in range ( <NUM_LIT:5> ) } <EOL> self . assertSetEqual ( ifaces , set ( self . br . get_iface_name_list ( ) ) ) <EOL> def test_get_port_stats ( self ) : <EOL> ( port_name , ofport ) = self . create_ovs_port ( ) <EOL> stats = set ( self . br . get_port_stats ( port_name ) . keys ( ) ) <EOL> self . assertTrue ( set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) . issubset ( stats ) ) <EOL> def test_get_vif_ports ( self ) : <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> ports = self . br . get_vif_ports ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( ports ) ) <EOL> self . assertTrue ( all ( [ isinstance ( x , ovs_lib . VifPort ) for x in ports ] ) ) <EOL> self . assertEqual ( sorted ( [ x . port_name for x in vif_ports ] ) , <EOL> sorted ( [ x . port_name for x in ports ] ) ) <EOL> def test_get_vif_ports_with_bond ( self ) : <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> orig = self . br . get_port_name_list <EOL> new_port_name_list = lambda : orig ( ) + [ '<STR_LIT>' ] <EOL> mock . patch . object ( self . br , '<STR_LIT>' , <EOL> new = new_port_name_list ) . start ( ) <EOL> ports = self . br . get_vif_ports ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( ports ) ) <EOL> self . assertTrue ( all ( [ isinstance ( x , ovs_lib . VifPort ) for x in ports ] ) ) <EOL> self . assertEqual ( sorted ( [ x . port_name for x in vif_ports ] ) , <EOL> sorted ( [ x . port_name for x in ports ] ) ) <EOL> def test_get_vif_port_set ( self ) : <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) for i in range ( <NUM_LIT:2> ) ] <EOL> ports = self . br . get_vif_port_set ( ) <EOL> expected = set ( [ x . vif_id for x in vif_ports ] ) <EOL> self . assertEqual ( expected , ports ) <EOL> def test_get_vif_port_set_with_missing_port ( self ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) ] <EOL> orig = self . br . get_port_name_list <EOL> new_port_name_list = lambda : orig ( ) + [ '<STR_LIT>' ] <EOL> mock . patch . object ( self . br , '<STR_LIT>' , <EOL> new = new_port_name_list ) . start ( ) <EOL> ports = self . br . get_vif_port_set ( ) <EOL> expected = set ( [ vif_ports [ <NUM_LIT:0> ] . vif_id ] ) <EOL> self . assertEqual ( expected , ports ) <EOL> def test_get_vif_port_set_on_empty_bridge_returns_empty_set ( self ) : <EOL> self . create_ovs_vif_port ( ) <EOL> br_2 = self . useFixture ( net_helpers . OVSBridgeFixture ( ) ) . bridge <EOL> self . assertEqual ( set ( ) , br_2 . get_vif_port_set ( ) ) <EOL> def test_get_ports_attributes ( self ) : <EOL> port_names = [ self . create_ovs_port ( ) [ <NUM_LIT:0> ] , self . create_ovs_port ( ) [ <NUM_LIT:0> ] ] <EOL> db_ports = self . br . get_ports_attributes ( '<STR_LIT>' , columns = [ '<STR_LIT:name>' ] ) <EOL> db_ports_names = [ p [ '<STR_LIT:name>' ] for p in db_ports ] <EOL> self . assertEqual ( sorted ( port_names ) , sorted ( db_ports_names ) ) <EOL> def test_get_port_tag_dict ( self ) : <EOL> pass <EOL> def test_get_vif_port_by_id ( self ) : <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> for vif in vif_ports : <EOL> self . assertEqual ( self . br . get_vif_port_by_id ( vif . vif_id ) . vif_id , <EOL> vif . vif_id ) <EOL> def test_get_vifs_by_ids ( self ) : <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> self . create_ovs_port ( ) <EOL> vif_ports = [ self . create_ovs_vif_port ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> by_id = self . br . get_vifs_by_ids ( [ v . vif_id for v in vif_ports ] ) <EOL> by_id = { vid : str ( vport ) for vid , vport in by_id . items ( ) } <EOL> self . assertEqual ( { v . vif_id : str ( v ) for v in vif_ports } , by_id ) <EOL> def test_delete_ports ( self ) : <EOL> nonvifs = { self . create_ovs_port ( ) [ <NUM_LIT:0> ] for i in range ( <NUM_LIT:2> ) } <EOL> vifs = { self . create_ovs_vif_port ( ) . port_name for i in range ( <NUM_LIT:2> ) } <EOL> self . assertSetEqual ( nonvifs . union ( vifs ) , <EOL> set ( self . br . get_port_name_list ( ) ) ) <EOL> self . br . delete_ports ( ) <EOL> self . assertSetEqual ( nonvifs , set ( self . br . get_port_name_list ( ) ) ) <EOL> self . br . delete_ports ( all_ports = True ) <EOL> self . assertEqual ( len ( self . br . get_port_name_list ( ) ) , <NUM_LIT:0> ) <EOL> def test_set_controller_connection_mode ( self ) : <EOL> controllers = [ '<STR_LIT>' ] <EOL> self . _set_controllers_connection_mode ( controllers ) <EOL> def test_set_multi_controllers_connection_mode ( self ) : <EOL> controllers = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . _set_controllers_connection_mode ( controllers ) <EOL> def _set_controllers_connection_mode ( self , controllers ) : <EOL> self . br . set_controller ( controllers ) <EOL> self . assertEqual ( sorted ( controllers ) , sorted ( self . br . get_controller ( ) ) ) <EOL> self . br . set_controllers_connection_mode ( '<STR_LIT>' ) <EOL> self . _assert_controllers_connection_mode ( '<STR_LIT>' ) <EOL> self . br . del_controller ( ) <EOL> self . assertEqual ( [ ] , self . br . get_controller ( ) ) <EOL> def _assert_controllers_connection_mode ( self , connection_mode ) : <EOL> controllers = self . br . db_get_val ( '<STR_LIT>' , self . br . br_name , <EOL> '<STR_LIT>' ) <EOL> controllers = [ controllers ] if isinstance ( <EOL> controllers , uuid . UUID ) else controllers <EOL> for controller in controllers : <EOL> self . assertEqual ( connection_mode , <EOL> self . br . db_get_val ( '<STR_LIT>' , <EOL> controller , <EOL> '<STR_LIT>' ) ) <EOL> def test_egress_bw_limit ( self ) : <EOL> port_name , _ = self . create_ovs_port ( ) <EOL> self . br . create_egress_bw_limit_for_port ( port_name , <NUM_LIT> , <NUM_LIT> ) <EOL> max_rate , burst = self . br . get_egress_bw_limit_for_port ( port_name ) <EOL> self . assertEqual ( <NUM_LIT> , max_rate ) <EOL> self . assertEqual ( <NUM_LIT> , burst ) <EOL> self . br . delete_egress_bw_limit_for_port ( port_name ) <EOL> max_rate , burst = self . br . get_egress_bw_limit_for_port ( port_name ) <EOL> self . assertIsNone ( max_rate ) <EOL> self . assertIsNone ( burst ) <EOL> class OVSLibTestCase ( base . BaseOVSLinuxTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( OVSLibTestCase , self ) . setUp ( ) <EOL> self . ovs = ovs_lib . BaseOVS ( ) <EOL> def test_bridge_lifecycle_baseovs ( self ) : <EOL> name = base . get_rand_name ( prefix = net_helpers . BR_PREFIX ) <EOL> self . addCleanup ( self . ovs . delete_bridge , name ) <EOL> br = self . ovs . add_bridge ( name ) <EOL> self . assertEqual ( br . br_name , name ) <EOL> self . assertTrue ( self . ovs . bridge_exists ( name ) ) <EOL> self . ovs . delete_bridge ( name ) <EOL> self . assertFalse ( self . ovs . bridge_exists ( name ) ) <EOL> def test_get_bridges ( self ) : <EOL> bridges = { <EOL> self . useFixture ( net_helpers . OVSBridgeFixture ( ) ) . bridge . br_name <EOL> for i in range ( <NUM_LIT:5> ) } <EOL> self . assertTrue ( set ( self . ovs . get_bridges ( ) ) . issuperset ( bridges ) ) <EOL> def test_bridge_lifecycle_ovsbridge ( self ) : <EOL> name = base . get_rand_name ( prefix = net_helpers . BR_PREFIX ) <EOL> br = ovs_lib . OVSBridge ( name ) <EOL> self . assertEqual ( br . br_name , name ) <EOL> self . assertFalse ( self . ovs . bridge_exists ( name ) ) <EOL> self . addCleanup ( self . ovs . delete_bridge , name ) <EOL> br . create ( ) <EOL> self . assertTrue ( self . ovs . bridge_exists ( name ) ) <EOL> br . destroy ( ) <EOL> self . assertFalse ( self . ovs . bridge_exists ( name ) ) </s>
