<s> from neutron_lib . api import validators <EOL> from oslo_config import cfg <EOL> from webob import exc as web_exc <EOL> from neutron . db import allowedaddresspairs_db as addr_pair_db <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import portsecurity_db <EOL> from neutron . extensions import allowedaddresspairs as addr_pair <EOL> from neutron . extensions import portsecurity as psec <EOL> from neutron . extensions import securitygroup as secgroup <EOL> from neutron import manager <EOL> from neutron . tests . unit . db import test_db_base_plugin_v2 <EOL> DB_PLUGIN_KLASS = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> class AllowedAddressPairTestCase ( <EOL> test_db_base_plugin_v2 . NeutronDbPluginV2TestCase ) : <EOL> def setUp ( self , plugin = None , ext_mgr = None ) : <EOL> super ( AllowedAddressPairTestCase , self ) . setUp ( plugin ) <EOL> plugin_obj = manager . NeutronManager . get_plugin ( ) <EOL> self . _skip_port_security = ( '<STR_LIT>' not in <EOL> plugin_obj . supported_extension_aliases ) <EOL> class AllowedAddressPairTestPlugin ( portsecurity_db . PortSecurityDbMixin , <EOL> db_base_plugin_v2 . NeutronDbPluginV2 , <EOL> addr_pair_db . AllowedAddressPairsMixin ) : <EOL> """<STR_LIT>""" <EOL> supported_extension_aliases = [ "<STR_LIT>" ] <EOL> def create_port ( self , context , port ) : <EOL> p = port [ '<STR_LIT:port>' ] <EOL> with context . session . begin ( subtransactions = True ) : <EOL> neutron_db = super ( AllowedAddressPairTestPlugin , self ) . create_port ( <EOL> context , port ) <EOL> p . update ( neutron_db ) <EOL> if validators . is_attr_set ( p . get ( addr_pair . ADDRESS_PAIRS ) ) : <EOL> self . _process_create_allowed_address_pairs ( <EOL> context , p , <EOL> p [ addr_pair . ADDRESS_PAIRS ] ) <EOL> else : <EOL> p [ addr_pair . ADDRESS_PAIRS ] = None <EOL> return port [ '<STR_LIT:port>' ] <EOL> def update_port ( self , context , id , port ) : <EOL> delete_addr_pairs = self . _check_update_deletes_allowed_address_pairs ( <EOL> port ) <EOL> has_addr_pairs = self . _check_update_has_allowed_address_pairs ( port ) <EOL> with context . session . begin ( subtransactions = True ) : <EOL> ret_port = super ( AllowedAddressPairTestPlugin , self ) . update_port ( <EOL> context , id , port ) <EOL> port [ '<STR_LIT:port>' ] . pop ( '<STR_LIT>' , None ) <EOL> ret_port . update ( port [ '<STR_LIT:port>' ] ) <EOL> if ( delete_addr_pairs or has_addr_pairs ) : <EOL> self . _delete_allowed_address_pairs ( context , id ) <EOL> self . _process_create_allowed_address_pairs ( <EOL> context , ret_port , <EOL> ret_port [ addr_pair . ADDRESS_PAIRS ] ) <EOL> return ret_port <EOL> class AllowedAddressPairDBTestCase ( AllowedAddressPairTestCase ) : <EOL> def setUp ( self , plugin = None , ext_mgr = None ) : <EOL> plugin = plugin or DB_PLUGIN_KLASS <EOL> super ( AllowedAddressPairDBTestCase , <EOL> self ) . setUp ( plugin = plugin , ext_mgr = ext_mgr ) <EOL> class TestAllowedAddressPairs ( AllowedAddressPairDBTestCase ) : <EOL> def test_create_port_allowed_address_pairs_bad_format ( self ) : <EOL> with self . network ( ) as net : <EOL> bad_values = [ False , True , <NUM_LIT> , <NUM_LIT:1> ] <EOL> for value in bad_values : <EOL> self . _create_port ( <EOL> self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> expected_res_status = web_exc . HTTPBadRequest . code , <EOL> arg_list = ( addr_pair . ADDRESS_PAIRS , ) , <EOL> allowed_address_pairs = value ) <EOL> def test_create_port_allowed_address_pairs ( self ) : <EOL> with self . network ( ) as net : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( addr_pair . ADDRESS_PAIRS , ) , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> self . assertEqual ( port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] , <EOL> address_pairs ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_port_security_true_allowed_address_pairs ( self ) : <EOL> if self . _skip_port_security : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> with self . network ( ) as net : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( '<STR_LIT>' , <EOL> addr_pair . ADDRESS_PAIRS , ) , <EOL> port_security_enabled = True , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> self . assertTrue ( port [ '<STR_LIT:port>' ] [ psec . PORTSECURITY ] ) <EOL> self . assertEqual ( port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] , <EOL> address_pairs ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_port_security_false_allowed_address_pairs ( self ) : <EOL> if self . _skip_port_security : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> with self . network ( ) as net : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( '<STR_LIT>' , <EOL> addr_pair . ADDRESS_PAIRS , ) , <EOL> port_security_enabled = False , <EOL> allowed_address_pairs = address_pairs ) <EOL> self . deserialize ( self . fmt , res ) <EOL> self . assertEqual ( <NUM_LIT> , res . status_int ) <EOL> address_pairs = [ ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( '<STR_LIT>' , <EOL> addr_pair . ADDRESS_PAIRS , ) , <EOL> port_security_enabled = False , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> self . assertFalse ( port [ '<STR_LIT:port>' ] [ psec . PORTSECURITY ] ) <EOL> self . assertEqual ( port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] , <EOL> address_pairs ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_port_bad_mac ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_create_port_bad_ip ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_create_missing_ip_field ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_create_duplicate_mac_ip ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_more_than_max_allowed_address_pair ( self ) : <EOL> cfg . CONF . set_default ( '<STR_LIT>' , <NUM_LIT:3> ) <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_equal_to_max_allowed_address_pair ( self ) : <EOL> cfg . CONF . set_default ( '<STR_LIT>' , <NUM_LIT:3> ) <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def test_create_overlap_with_fixed_ip ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> with self . network ( ) as network : <EOL> with self . subnet ( network = network , cidr = '<STR_LIT>' ) as subnet : <EOL> fixed_ips = [ { '<STR_LIT>' : subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , network [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( addr_pair . ADDRESS_PAIRS , <EOL> '<STR_LIT>' ) , <EOL> allowed_address_pairs = address_pairs , <EOL> fixed_ips = fixed_ips ) <EOL> self . assertEqual ( res . status_int , <NUM_LIT> ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_port_extra_args ( self ) : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _create_port_with_address_pairs ( address_pairs , <NUM_LIT> ) <EOL> def _create_port_with_address_pairs ( self , address_pairs , ret_code ) : <EOL> with self . network ( ) as net : <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( addr_pair . ADDRESS_PAIRS , ) , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> self . assertEqual ( res . status_int , ret_code ) <EOL> if ret_code == <NUM_LIT> : <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_update_add_address_pairs ( self ) : <EOL> with self . network ( ) as net : <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> update_port = { '<STR_LIT:port>' : { addr_pair . ADDRESS_PAIRS : <EOL> address_pairs } } <EOL> req = self . new_update_request ( '<STR_LIT>' , update_port , <EOL> port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> port = self . deserialize ( self . fmt , req . get_response ( self . api ) ) <EOL> self . assertEqual ( port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] , <EOL> address_pairs ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_create_address_gets_port_mac ( self ) : <EOL> with self . network ( ) as net : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( '<STR_LIT>' , <EOL> addr_pair . ADDRESS_PAIRS , ) , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) [ '<STR_LIT:port>' ] <EOL> port_addr_mac = port [ addr_pair . ADDRESS_PAIRS ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( port_addr_mac , <EOL> port [ '<STR_LIT>' ] ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:id>' ] ) <EOL> def test_update_port_security_off_address_pairs ( self ) : <EOL> if self . _skip_port_security : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> with self . network ( ) as net : <EOL> with self . subnet ( network = net ) as subnet : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> with self . port ( subnet = subnet , <EOL> arg_list = ( psec . PORTSECURITY , <EOL> addr_pair . ADDRESS_PAIRS , <EOL> secgroup . SECURITYGROUPS ) , <EOL> port_security_enabled = True , <EOL> allowed_address_pairs = address_pairs , <EOL> security_groups = [ ] ) as port : <EOL> update_port = { '<STR_LIT:port>' : { psec . PORTSECURITY : False } } <EOL> req = self . new_update_request ( '<STR_LIT>' , update_port , <EOL> port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> res = req . get_response ( self . api ) <EOL> self . assertEqual ( <NUM_LIT> , res . status_int ) <EOL> def test_update_with_none_and_own_mac_for_duplicate_ip ( self ) : <EOL> with self . network ( ) as net : <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> mac_address = port [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : mac_address , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> update_port = { '<STR_LIT:port>' : { addr_pair . ADDRESS_PAIRS : <EOL> address_pairs } } <EOL> req = self . new_update_request ( '<STR_LIT>' , update_port , <EOL> port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> res = req . get_response ( self . api ) <EOL> self . assertEqual ( <NUM_LIT> , res . status_int ) <EOL> def test_create_port_remove_allowed_address_pairs_with_list ( self ) : <EOL> self . _test_create_port_remove_allowed_address_pairs ( [ ] ) <EOL> def test_create_port_remove_allowed_address_pairs_with_none ( self ) : <EOL> self . _test_create_port_remove_allowed_address_pairs ( None ) <EOL> def _test_create_port_remove_allowed_address_pairs ( self , update_value ) : <EOL> with self . network ( ) as net : <EOL> address_pairs = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> res = self . _create_port ( self . fmt , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , <EOL> arg_list = ( addr_pair . ADDRESS_PAIRS , ) , <EOL> allowed_address_pairs = address_pairs ) <EOL> port = self . deserialize ( self . fmt , res ) <EOL> update_port = { '<STR_LIT:port>' : { addr_pair . ADDRESS_PAIRS : update_value } } <EOL> req = self . new_update_request ( '<STR_LIT>' , update_port , <EOL> port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) <EOL> port = self . deserialize ( self . fmt , req . get_response ( self . api ) ) <EOL> self . assertEqual ( [ ] , port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:port>' ] [ '<STR_LIT:id>' ] ) </s>
