<s> from neutron_lib . api import validators <EOL> from oslo_db import exception as db_exc <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import orm <EOL> from neutron . api . v2 import attributes as attr <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import model_base <EOL> from neutron . db import models_v2 <EOL> from neutron . extensions import allowedaddresspairs as addr_pair <EOL> class AllowedAddressPair ( model_base . BASEV2 ) : <EOL> port_id = sa . Column ( sa . String ( <NUM_LIT> ) , <EOL> sa . ForeignKey ( '<STR_LIT>' , ondelete = "<STR_LIT>" ) , <EOL> primary_key = True ) <EOL> mac_address = sa . Column ( sa . String ( <NUM_LIT:32> ) , nullable = False , primary_key = True ) <EOL> ip_address = sa . Column ( sa . String ( <NUM_LIT:64> ) , nullable = False , primary_key = True ) <EOL> port = orm . relationship ( <EOL> models_v2 . Port , <EOL> backref = orm . backref ( "<STR_LIT>" , <EOL> lazy = "<STR_LIT>" , cascade = "<STR_LIT>" ) ) <EOL> class AllowedAddressPairsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def _process_create_allowed_address_pairs ( self , context , port , <EOL> allowed_address_pairs ) : <EOL> if not validators . is_attr_set ( allowed_address_pairs ) : <EOL> return [ ] <EOL> try : <EOL> with context . session . begin ( subtransactions = True ) : <EOL> for address_pair in allowed_address_pairs : <EOL> if '<STR_LIT>' not in address_pair : <EOL> address_pair [ '<STR_LIT>' ] = port [ '<STR_LIT>' ] <EOL> db_pair = AllowedAddressPair ( <EOL> port_id = port [ '<STR_LIT:id>' ] , <EOL> mac_address = address_pair [ '<STR_LIT>' ] , <EOL> ip_address = address_pair [ '<STR_LIT>' ] ) <EOL> context . session . add ( db_pair ) <EOL> except db_exc . DBDuplicateEntry : <EOL> raise addr_pair . DuplicateAddressPairInRequest ( <EOL> mac_address = address_pair [ '<STR_LIT>' ] , <EOL> ip_address = address_pair [ '<STR_LIT>' ] ) <EOL> return allowed_address_pairs <EOL> def get_allowed_address_pairs ( self , context , port_id ) : <EOL> pairs = ( context . session . query ( AllowedAddressPair ) . <EOL> filter_by ( port_id = port_id ) ) <EOL> return [ self . _make_allowed_address_pairs_dict ( pair ) <EOL> for pair in pairs ] <EOL> def _extend_port_dict_allowed_address_pairs ( self , port_res , port_db ) : <EOL> allowed_address_pairs = [ <EOL> self . _make_allowed_address_pairs_dict ( address_pair ) for <EOL> address_pair in port_db . allowed_address_pairs ] <EOL> port_res [ addr_pair . ADDRESS_PAIRS ] = allowed_address_pairs <EOL> return port_res <EOL> db_base_plugin_v2 . NeutronDbPluginV2 . register_dict_extend_funcs ( <EOL> attr . PORTS , [ '<STR_LIT>' ] ) <EOL> def _delete_allowed_address_pairs ( self , context , id ) : <EOL> query = self . _model_query ( context , AllowedAddressPair ) <EOL> with context . session . begin ( subtransactions = True ) : <EOL> query . filter ( AllowedAddressPair . port_id == id ) . delete ( ) <EOL> def _make_allowed_address_pairs_dict ( self , allowed_address_pairs , <EOL> fields = None ) : <EOL> res = { '<STR_LIT>' : allowed_address_pairs [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : allowed_address_pairs [ '<STR_LIT>' ] } <EOL> return self . _fields ( res , fields ) <EOL> def _has_address_pairs ( self , port ) : <EOL> return ( validators . is_attr_set ( port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] ) <EOL> and port [ '<STR_LIT:port>' ] [ addr_pair . ADDRESS_PAIRS ] != [ ] ) <EOL> def _check_update_has_allowed_address_pairs ( self , port ) : <EOL> """<STR_LIT>""" <EOL> return ( addr_pair . ADDRESS_PAIRS in port [ '<STR_LIT:port>' ] and <EOL> self . _has_address_pairs ( port ) ) <EOL> def _check_update_deletes_allowed_address_pairs ( self , port ) : <EOL> """<STR_LIT>""" <EOL> return ( addr_pair . ADDRESS_PAIRS in port [ '<STR_LIT:port>' ] and <EOL> not self . _has_address_pairs ( port ) ) <EOL> def is_address_pairs_attribute_updated ( self , port , update_attrs ) : <EOL> """<STR_LIT>""" <EOL> new_pairs = update_attrs . get ( addr_pair . ADDRESS_PAIRS ) <EOL> if new_pairs is None : <EOL> return False <EOL> old_pairs = port . get ( addr_pair . ADDRESS_PAIRS ) <EOL> return new_pairs != old_pairs <EOL> def update_address_pairs_on_port ( self , context , port_id , port , <EOL> original_port , updated_port ) : <EOL> """<STR_LIT>""" <EOL> new_pairs = port [ '<STR_LIT:port>' ] . get ( addr_pair . ADDRESS_PAIRS ) <EOL> if self . is_address_pairs_attribute_updated ( original_port , <EOL> port [ '<STR_LIT:port>' ] ) : <EOL> updated_port [ addr_pair . ADDRESS_PAIRS ] = new_pairs <EOL> self . _delete_allowed_address_pairs ( context , port_id ) <EOL> self . _process_create_allowed_address_pairs ( <EOL> context , updated_port , new_pairs ) <EOL> return True <EOL> return False </s>
