# Copyright 2014 Rackspace
#
# 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 ipaddress
from wsme import exc as wsme_exc
from wsme import types as wtypes
from octavia.common import constants
from octavia.common import exceptions
from octavia.common import validate
[docs]
class IPAddressType(wtypes.UserType):
basetype = str
name = 'ipaddress'
[docs]
@staticmethod
def validate(value):
"""Validates whether value is an IPv4 or IPv6 address."""
try:
wtypes.IPv4AddressType.validate(value)
return value
except ValueError:
try:
wtypes.IPv6AddressType.validate(value)
return value
except ValueError as e:
error = 'Value should be IPv4 or IPv6 format'
raise ValueError(error) from e
[docs]
class CidrType(wtypes.UserType):
basetype = str
name = 'cidr'
[docs]
@staticmethod
def validate(value):
"""Validates whether value is an IPv4 or IPv6 CIDR."""
try:
return ipaddress.ip_network(value, strict=False).with_prefixlen
except Exception as e:
error = 'Value should be IPv4 or IPv6 CIDR format'
raise ValueError(error) from e
[docs]
class AlpnProtocolType(wtypes.UserType):
basetype = str
name = 'alpn_protocol'
[docs]
@staticmethod
def validate(value):
"""Validates whether value is a valid ALPN protocol ID."""
if value in constants.SUPPORTED_ALPN_PROTOCOLS:
return value
error = 'Value should be a valid ALPN protocol ID'
raise ValueError(error)
[docs]
class URLType(wtypes.UserType):
basetype = str
name = 'url'
def __init__(self, require_scheme=True):
super().__init__()
self.require_scheme = require_scheme
[docs]
def validate(self, value):
try:
validate.url(value, require_scheme=self.require_scheme)
except exceptions.InvalidURL as e:
error = 'Value must be a valid URL string'
raise ValueError(error) from e
return value
[docs]
class URLPathType(wtypes.UserType):
basetype = str
name = 'url_path'
[docs]
@staticmethod
def validate(value):
try:
validate.url_path(value)
except exceptions.InvalidURLPath as e:
error = 'Value must be a valid URL Path string'
raise ValueError(error) from e
return value
[docs]
class BaseType(wtypes.Base, metaclass=BaseMeta):
@classmethod
def _full_response(cls):
return False
[docs]
@classmethod
def from_db_obj(cls, db_obj):
"""Map DB object to WSME attributes.
The behaviour can be individually extended by implementing from_db_obj
closer to the type.
:param db_obj: the DB object to process
"""
result = cls()
def try_set_attr(result, attr_name, attr):
# This is not as messy as it may seem. Trying to setattr and
# discarding InvalidInput exceptions allows from_db_obj to stay as
# generic as possible and effectively ignore attributes that need
# a developer-defined processing path. This should only affect
# development in the event that a new property is added to an API
# response type but needs an explicit procesing path (for example
# for attributes that are IdOnlyType normally but XFullResponse
# when _full_response).
try:
setattr(result, attr_name, attr)
except wsme_exc.InvalidInput:
pass
attr_names = [attr.name for attr in cls._wsme_attributes]
for attr_name in attr_names:
if hasattr(db_obj, attr_name):
attr = getattr(db_obj, attr_name)
if attr_name in ['name', 'description'] and attr is None:
setattr(result, attr_name, '')
else:
try_set_attr(result, attr_name, attr)
elif hasattr(cls, '_type_to_db_map'):
if (attr_name not in cls._type_to_db_map or
'.' in cls._type_to_db_map[attr_name]):
continue
attr = getattr(db_obj, cls._type_to_db_map[attr_name])
try_set_attr(result, attr_name, attr)
return result
[docs]
@classmethod
def translate_dict_keys_to_db_obj(cls, wsme_dict):
"""Translate the keys from wsme class type, to db_obj."""
if not hasattr(cls, '_type_to_db_map'):
return wsme_dict
res = {}
for (k, v) in wsme_dict.items():
if k in cls._type_to_db_map:
k = cls._type_to_db_map[k]
if '.' in k:
parent, child = k.split('.')
if parent not in res:
res[parent] = {}
res[parent][child] = v
continue
res[k] = v
return res
[docs]
@classmethod
def translate_key_to_db_obj(cls, key):
"""Translate the keys from wsme class type, to db_obj."""
if not hasattr(cls, '_type_to_db_map') or (
key not in cls._type_to_db_map):
return key
return cls._type_to_db_map[key]
[docs]
def to_dict(self, render_unsets=False):
"""Converts Octavia WSME type to dictionary.
:param render_unsets: If True, will convert items that are WSME Unset
types to None. If False, does not add the item
"""
# Set project_id equal tenant_id if project_id is unset and tenant_id
# is
if hasattr(self, 'project_id') and hasattr(self, 'tenant_id'):
# pylint: disable=access-member-before-definition
if (isinstance(self.project_id, wtypes.UnsetType) and
not isinstance(self.tenant_id, wtypes.UnsetType)):
self.project_id = self.tenant_id
if hasattr(self, 'admin_state_up') and getattr(
self, 'admin_state_up') is None:
# This situation will be hit during request with
# admin_state_up is null. If users specify this field to null,
# then we treat it as False
self.admin_state_up = bool(self.admin_state_up)
wsme_dict = {}
for attr in dir(self):
if attr.startswith('_'):
continue
value = getattr(self, attr, None)
# TODO(blogan): Investigate wsme types handling the duality of
# tenant_id and project_id in a clean way. One way could be
# wsme.rest.json.fromjson and using the @fromjson.when_object
# decorator.
if attr == 'tenant_id':
continue
if value and callable(value):
continue
if value and isinstance(value, BaseType):
value = value.to_dict(render_unsets=render_unsets)
if value and isinstance(value, list):
value = [val.to_dict(render_unsets=render_unsets)
if isinstance(val, BaseType) else val
for val in value]
if isinstance(value, wtypes.UnsetType):
if render_unsets:
value = None
else:
continue
wsme_dict[attr] = value
return self.translate_dict_keys_to_db_obj(wsme_dict)
[docs]
class IdOnlyType(BaseType):
id = wtypes.wsattr(wtypes.UuidType(), mandatory=True)
[docs]
class NameOnlyType(BaseType):
name = wtypes.wsattr(wtypes.StringType(max_length=255), mandatory=True)
[docs]
class PageType(BaseType):
href = wtypes.StringType()
rel = wtypes.StringType()
