¡@
OpenStack Study: allowedaddresspairs.py**** CubicPower OpenStack Study ****
# Copyright 2013 VMware, Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import webob.exc from neutron.api.v2 import attributes as attr from neutron.common import exceptions as nexception
**** CubicPower OpenStack Study **** class AllowedAddressPairsMissingIP(nexception.InvalidInput): message = _("AllowedAddressPair must contain ip_address")
**** CubicPower OpenStack Study **** class AddressPairAndPortSecurityRequired(nexception.Conflict): message = _("Port Security must be enabled in order to have allowed " "address pairs on a port.")
**** CubicPower OpenStack Study **** class DuplicateAddressPairInRequest(nexception.InvalidInput): message = _("Request contains duplicate address pair: " "mac_address %(mac_address)s ip_address %(ip_address)s.")
**** CubicPower OpenStack Study **** class AddressPairMatchesPortFixedIPAndMac(nexception.InvalidInput): message = _("Port's Fixed IP and Mac Address match an address pair entry.")
**** CubicPower OpenStack Study **** def _validate_allowed_address_pairs(address_pairs, valid_values=None): unique_check = {} for address_pair in address_pairs: # mac_address is optional, if not set we use the mac on the port if 'mac_address' in address_pair: msg = attr._validate_mac_address(address_pair['mac_address']) if msg: raise webob.exc.HTTPBadRequest(msg) if 'ip_address' not in address_pair: raise AllowedAddressPairsMissingIP() mac = address_pair.get('mac_address') ip_address = address_pair['ip_address'] if (mac, ip_address) not in unique_check: unique_check[(mac, ip_address)] = None else: raise DuplicateAddressPairInRequest(mac_address=mac, ip_address=ip_address) invalid_attrs = set(address_pair.keys()) - set(['mac_address', 'ip_address']) if invalid_attrs: msg = (_("Unrecognized attribute(s) '%s'") % ', '.join(set(address_pair.keys()) - set(['mac_address', 'ip_address']))) raise webob.exc.HTTPBadRequest(msg) if '/' in ip_address: msg = attr._validate_subnet(ip_address) else: msg = attr._validate_ip_address(ip_address) if msg: raise webob.exc.HTTPBadRequest(msg) attr.validators['type:validate_allowed_address_pairs'] = ( _validate_allowed_address_pairs) ADDRESS_PAIRS = 'allowed_address_pairs' EXTENDED_ATTRIBUTES_2_0 = { 'ports': { ADDRESS_PAIRS: {'allow_post': True, 'allow_put': True, 'convert_list_to': attr.convert_kvp_list_to_dict, 'validate': {'type:validate_allowed_address_pairs': None}, 'enforce_policy': True, 'default': attr.ATTR_NOT_SPECIFIED, 'is_visible': True}, } }
**** CubicPower OpenStack Study **** class Allowedaddresspairs(object): """Extension class supporting allowed address pairs.""" @classmethod
**** CubicPower OpenStack Study **** def get_name(cls): return "Allowed Address Pairs" @classmethod
**** CubicPower OpenStack Study **** def get_alias(cls): return "allowed-address-pairs" @classmethod
**** CubicPower OpenStack Study **** def get_description(cls): return "Provides allowed address pairs" @classmethod
**** CubicPower OpenStack Study **** def get_namespace(cls): return "http://docs.openstack.org/ext/allowedaddresspairs/api/v2.0" @classmethod
**** CubicPower OpenStack Study **** def get_updated(cls): return "2013-07-23T10:00:00-00:00"
**** CubicPower OpenStack Study **** def get_extended_resources(self, version): if version == "2.0": attr.PLURALS.update({'allowed_address_pairs': 'allowed_address_pair'}) return EXTENDED_ATTRIBUTES_2_0 else: return {}
|