openstack/openstacksdk
Code Issues Proposed changes

resource: Rework fields

Browse Source 
The Body, Header, URI and Computed field types are not values in their
own right: rather, they are indicators of where the real values will
come from. At some point, we should probably make better use of type
annotations but for now, modify things so that 'openstack.resource.Body'
et al are functions that return 'Any' and don't conflict with explicit
type annotations. This allows us to change some of the attributes of the
base resource - namely the 'name' and 'location' fields - to reflect
their "real" types.

Change-Id: Ia771d18cf6d8d72dbf995418166e996515f5d52c
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2024-07-20 20:02:53 +01:00
parent 6395008300
commit a11df932f6
14 changed files with 623 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
openstack/baremetal/v1/_common.py
View File

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from openstack import fields
from openstack import resource
@ -144,7 +145,7 @@ def fields_type(value, resource_type):
resource_mapping = {
key: value.name
for key, value in resource_type.__dict__.items()
if isinstance(value, resource.Body)
if isinstance(value, fields.Body)
}
return comma_separated_list(resource_mapping.get(x, x) for x in value)

2
openstack/common/metadata.py
View File

@ -16,7 +16,7 @@ from openstack import utils
class MetadataMixin:
id: resource.Body
id: str
base_path: str
_body: resource._ComponentManager

2
openstack/common/tag.py
View File

@ -16,7 +16,7 @@ from openstack import utils
class TagMixin:
id: resource.Body
id: str
base_path: str
_body: resource._ComponentManager

235
openstack/fields.py Normal file
View File

@ -0,0 +1,235 @@
# 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 abc
import typing as ty
import warnings
from requests import structures
from openstack import format
from openstack import warnings as os_warnings
_SEEN_FORMAT = '{name}_seen'
def _convert_type(value, data_type, list_type=None):
# This should allow handling list of dicts that have their own
# Component type directly. See openstack/compute/v2/limits.py
# and the RateLimit type for an example.
if not data_type:
return value
if issubclass(data_type, list):
if isinstance(value, (list, tuple, set)):
if not list_type:
return value
ret = []
for raw in value:
ret.append(_convert_type(raw, list_type))
return ret
elif list_type:
return [_convert_type(value, list_type)]
# "if-match" in Object is a good example of the need here
return [value]
elif isinstance(value, data_type):
return value
if not isinstance(value, data_type):
if issubclass(data_type, format.Formatter):
return data_type.deserialize(value)
# This should allow handling sub-dicts that have their own
# Component type directly. See openstack/compute/v2/limits.py
# and the AbsoluteLimits type for an example.
if isinstance(value, dict):
return data_type(**value)
try:
return data_type(value)
except ValueError:
# If we can not convert data to the expected type return empty
# instance of the expected type.
# This is necessary to handle issues like with flavor.swap where
# empty string means "0".
return data_type()
class _BaseComponent(abc.ABC):
# The name this component is being tracked as in the Resource
key: str
# The class to be used for mappings
_map_cls: type[ty.Mapping] = dict
#: Marks the property as deprecated.
deprecated = False
#: Deprecation reason message used to warn users when deprecated == True
deprecation_reason = None
#: Control field used to manage the deprecation warning. We want to warn
#: only once when the attribute is retrieved in the code.
already_warned_deprecation = False
def __init__(
self,
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
**kwargs,
):
"""A typed descriptor for a component that makes up a Resource
:param name: The name this component exists as on the server
:param type:
The type this component is expected to be by the server.
By default this is None, meaning any value you specify
will work. If you specify type=dict and then set a
component to a string, __set__ will fail, for example.
:param default: Typically None, but any other default can be set.
:param alias: If set, alternative attribute on object to return.
:param aka: If set, additional name attribute would be available under.
:param alternate_id:
When `True`, this property is known internally as a value that
can be sent with requests that require an ID but when `id` is
not a name the Resource has. This is a relatively uncommon case,
and this setting should only be used once per Resource.
:param list_type:
If type is `list`, list_type designates what the type of the
elements of the list should be.
:param coerce_to_default:
If the Component is None or not present, force the given default
to be used. If a default is not given but a type is given,
construct an empty version of the type in question.
:param deprecated:
Indicates if the option is deprecated. If it is, we display a
warning message to the user.
:param deprecation_reason:
Custom deprecation message.
"""
self.name = name
self.type = type
if type is not None and coerce_to_default and not default:
self.default = type()
else:
self.default = default
self.alias = alias
self.aka = aka
self.alternate_id = alternate_id
self.list_type = list_type
self.coerce_to_default = coerce_to_default
self.deprecated = deprecated
self.deprecation_reason = deprecation_reason
def __get__(self, instance, owner):
if instance is None:
return self
attributes = getattr(instance, self.key)
try:
value = attributes[self.name]
except KeyError:
value = self.default
if self.alias:
# Resource attributes can be aliased to each other. If neither
# of them exist, then simply doing a
# getattr(instance, self.alias) here sends things into
# infinite recursion (this _get method is what gets called
# when getattr(instance) is called.
# To combat that, we set a flag on the instance saying that
# we have seen the current name, and we check before trying
# to resolve the alias if there is already a flag set for that
# alias name. We then remove the seen flag for ourselves after
# we exit the alias getattr to clean up after ourselves for
# the next time.
alias_flag = _SEEN_FORMAT.format(name=self.alias)
if not getattr(instance, alias_flag, False):
seen_flag = _SEEN_FORMAT.format(name=self.name)
# Prevent infinite recursion
setattr(instance, seen_flag, True)
value = getattr(instance, self.alias)
delattr(instance, seen_flag)
self.warn_if_deprecated_property(value)
return value
# self.type() should not be called on None objects.
if value is None:
return None
# This warning are pretty intruisive. Every time attribute is accessed
# a warning is being thrown. In neutron clients we have way too many
# places that still refer to tenant_id even though they may also
# properly support project_id. For now we silence tenant_id warnings.
if self.name != "tenant_id":
self.warn_if_deprecated_property(value)
return _convert_type(value, self.type, self.list_type)
def warn_if_deprecated_property(self, value):
deprecated = object.__getattribute__(self, 'deprecated')
deprecation_reason = object.__getattribute__(
self,
'deprecation_reason',
)
if value and deprecated:
warnings.warn(
"The field {!r} has been deprecated. {}".format(
self.name, deprecation_reason or "Avoid usage."
),
os_warnings.RemovedFieldWarning,
)
return value
def __set__(self, instance, value):
if self.coerce_to_default and value is None:
value = self.default
if value != self.default:
value = _convert_type(value, self.type, self.list_type)
attributes = getattr(instance, self.key)
attributes[self.name] = value
def __delete__(self, instance):
try:
attributes = getattr(instance, self.key)
del attributes[self.name]
except KeyError:
pass
class Body(_BaseComponent):
"""Body attributes"""
key = "_body"
class Header(_BaseComponent):
"""Header attributes"""
key = "_header"
_map_cls = structures.CaseInsensitiveDict
class URI(_BaseComponent):
"""URI attributes"""
key = "_uri"
class Computed(_BaseComponent):
"""Computed attributes"""
key = "_computed"

3
openstack/image/_download.py
View File

@ -14,7 +14,6 @@ import hashlib
import io
from openstack import exceptions
from openstack import resource
from openstack import utils
@ -28,7 +27,7 @@ def _verify_checksum(md5, checksum):
class DownloadMixin:
id: resource.Body
id: str
base_path: str
def fetch(

7
openstack/image/v2/metadef_property.py
View File

@ -11,6 +11,7 @@
# under the License.
from openstack import exceptions
from openstack import fields
from openstack import resource
@ -57,7 +58,7 @@ class MetadefProperty(resource.Resource):
# FIXME(stephenfin): This is causing conflicts due to the 'dict.items'
# method. Perhaps we need to rename it?
#: Schema for the items in an array.
items = resource.Body('items', type=dict) # type: ignore
items = resource.Body('items', type=dict)
#: Indicates whether all values in the array must be distinct.
require_unique_items = resource.Body(
'uniqueItems', type=bool, default=False
@ -114,7 +115,7 @@ class MetadefProperty(resource.Resource):
# Known attr
hasattr(cls, k)
# Is real attr property
and isinstance(getattr(cls, k), resource.Body)
and isinstance(getattr(cls, k), fields.Body)
# not included in the query_params
and k not in cls._query_mapping._mapping.keys()
):
@ -125,7 +126,7 @@ class MetadefProperty(resource.Resource):
for k, v in params.items():
# We need to gather URI parts to set them on the resource later
if hasattr(cls, k) and isinstance(getattr(cls, k), resource.URI):
if hasattr(cls, k) and isinstance(getattr(cls, k), fields.URI):
uri_params[k] = v
def _dict_filter(f, d):

4
openstack/orchestration/v1/stack_environment.py
View File

@ -29,9 +29,9 @@ class StackEnvironment(resource.Resource):
# Backwards compat
stack_name = name
#: ID of the stack where the template is referenced.
id = resource.URI('stack_id') # type: ignore
id = resource.URI('stack_id')
# Backwards compat
stack_id = id # type: ignore
stack_id = id
#: A list of parameter names whose values are encrypted
encrypted_param_names = resource.Body('encrypted_param_names')
#: A list of event sinks

4
openstack/orchestration/v1/stack_files.py
View File

@ -29,9 +29,9 @@ class StackFiles(resource.Resource):
# Backwards compat
stack_name = name
#: ID of the stack where the template is referenced.
id = resource.URI('stack_id') # type: ignore
id = resource.URI('stack_id')
# Backwards compat
stack_id = id # type: ignore
stack_id = id
def fetch(
self, session, requires_id=False, base_path=None, *args, **kwargs

5
openstack/placement/v1/resource_provider_inventory.py
View File

@ -11,6 +11,7 @@
# under the License.
from openstack import exceptions
from openstack import fields
from openstack import resource
@ -125,7 +126,7 @@ class ResourceProviderInventory(resource.Resource):
# Known attr
hasattr(cls, k)
# Is real attr property
and isinstance(getattr(cls, k), resource.Body)
and isinstance(getattr(cls, k), fields.Body)
# not included in the query_params
and k not in cls._query_mapping._mapping.keys()
):
@ -136,7 +137,7 @@ class ResourceProviderInventory(resource.Resource):
for k, v in params.items():
# We need to gather URI parts to set them on the resource later
if hasattr(cls, k) and isinstance(getattr(cls, k), resource.URI):
if hasattr(cls, k) and isinstance(getattr(cls, k), fields.URI):
uri_params[k] = v
def _dict_filter(f, d):

5
openstack/placement/v1/trait.py
View File

@ -11,6 +11,7 @@
# under the License.
from openstack import exceptions
from openstack import fields
from openstack import resource
@ -76,7 +77,7 @@ class Trait(resource.Resource):
# Known attr
hasattr(cls, k)
# Is real attr property
and isinstance(getattr(cls, k), resource.Body)
and isinstance(getattr(cls, k), fields.Body)
# not included in the query_params
and k not in cls._query_mapping._mapping.keys()
):
@ -87,7 +88,7 @@ class Trait(resource.Resource):
for k, v in params.items():
# We need to gather URI parts to set them on the resource later
if hasattr(cls, k) and isinstance(getattr(cls, k), resource.URI):
if hasattr(cls, k) and isinstance(getattr(cls, k), fields.URI):
uri_params[k] = v
def _dict_filter(f, d):

372
openstack/resource.py
View File

@ -16,8 +16,8 @@ The :class:`~openstack.resource.Resource` class is a base
class that represent a remote resource. The attributes that
comprise a request or response for this resource are specified
as class members on the Resource subclass where their values
are of a component type, including :class:`~openstack.resource.Body`,
:class:`~openstack.resource.Header`, and :class:`~openstack.resource.URI`.
are of a component type, including :class:`~openstack.fields.Body`,
:class:`~openstack.fields.Header`, and :class:`~openstack.fields.URI`.
For update management, :class:`~openstack.resource.Resource` employs
a series of :class:`~openstack.resource._ComponentManager` instances
@ -32,7 +32,6 @@ converted into this Resource class' appropriate components and types
and then returned to the caller.
"""
import abc
import collections
import inspect
import itertools
@ -45,230 +44,126 @@ import warnings
import jsonpatch
from keystoneauth1 import adapter
from keystoneauth1 import discover
from requests import structures
from openstack import _log
from openstack import exceptions
from openstack import format
from openstack import fields
from openstack import utils
from openstack import warnings as os_warnings
_SEEN_FORMAT = '{name}_seen'
LOG = _log.setup_logging(__name__)
def _convert_type(value, data_type, list_type=None):
# This should allow handling list of dicts that have their own
# Component type directly. See openstack/compute/v2/limits.py
# and the RateLimit type for an example.
if not data_type:
return value
if issubclass(data_type, list):
if isinstance(value, (list, tuple, set)):
if not list_type:
return value
ret = []
for raw in value:
ret.append(_convert_type(raw, list_type))
return ret
elif list_type:
return [_convert_type(value, list_type)]
# "if-match" in Object is a good example of the need here
return [value]
elif isinstance(value, data_type):
return value
if not isinstance(value, data_type):
if issubclass(data_type, format.Formatter):
return data_type.deserialize(value)
# This should allow handling sub-dicts that have their own
# Component type directly. See openstack/compute/v2/limits.py
# and the AbsoluteLimits type for an example.
if isinstance(value, dict):
return data_type(**value)
try:
return data_type(value)
except ValueError:
# If we can not convert data to the expected type return empty
# instance of the expected type.
# This is necessary to handle issues like with flavor.swap where
# empty string means "0".
return data_type()
class _BaseComponent(abc.ABC):
# The name this component is being tracked as in the Resource
key: str
# The class to be used for mappings
_map_cls: type[ty.Mapping] = dict
#: Marks the property as deprecated.
deprecated = False
#: Deprecation reason message used to warn users when deprecated == True
deprecation_reason = None
#: Control field used to manage the deprecation warning. We want to warn
#: only once when the attribute is retrieved in the code.
already_warned_deprecation = False
def __init__(
self,
def Body(
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
**kwargs,
):
return fields.Body(
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
type=type,
default=default,
alias=alias,
aka=aka,
alternate_id=alternate_id,
list_type=list_type,
coerce_to_default=coerce_to_default,
deprecated=deprecated,
deprecation_reason=deprecation_reason,
**kwargs,
):
"""A typed descriptor for a component that makes up a Resource
:param name: The name this component exists as on the server
:param type:
The type this component is expected to be by the server.
By default this is None, meaning any value you specify
will work. If you specify type=dict and then set a
component to a string, __set__ will fail, for example.
:param default: Typically None, but any other default can be set.
:param alias: If set, alternative attribute on object to return.
:param aka: If set, additional name attribute would be available under.
:param alternate_id:
When `True`, this property is known internally as a value that
can be sent with requests that require an ID but when `id` is
not a name the Resource has. This is a relatively uncommon case,
and this setting should only be used once per Resource.
:param list_type:
If type is `list`, list_type designates what the type of the
elements of the list should be.
:param coerce_to_default:
If the Component is None or not present, force the given default
to be used. If a default is not given but a type is given,
construct an empty version of the type in question.
:param deprecated:
Indicates if the option is deprecated. If it is, we display a
warning message to the user.
:param deprecation_reason:
Custom deprecation message.
"""
self.name = name
self.type = type
if type is not None and coerce_to_default and not default:
self.default = type()
else:
self.default = default
self.alias = alias
self.aka = aka
self.alternate_id = alternate_id
self.list_type = list_type
self.coerce_to_default = coerce_to_default
self.deprecated = deprecated
self.deprecation_reason = deprecation_reason
def __get__(self, instance, owner):
if instance is None:
return self
attributes = getattr(instance, self.key)
try:
value = attributes[self.name]
except KeyError:
value = self.default
if self.alias:
# Resource attributes can be aliased to each other. If neither
# of them exist, then simply doing a
# getattr(instance, self.alias) here sends things into
# infinite recursion (this _get method is what gets called
# when getattr(instance) is called.
# To combat that, we set a flag on the instance saying that
# we have seen the current name, and we check before trying
# to resolve the alias if there is already a flag set for that
# alias name. We then remove the seen flag for ourselves after
# we exit the alias getattr to clean up after ourselves for
# the next time.
alias_flag = _SEEN_FORMAT.format(name=self.alias)
if not getattr(instance, alias_flag, False):
seen_flag = _SEEN_FORMAT.format(name=self.name)
# Prevent infinite recursion
setattr(instance, seen_flag, True)
value = getattr(instance, self.alias)
delattr(instance, seen_flag)
self.warn_if_deprecated_property(value)
return value
# self.type() should not be called on None objects.
if value is None:
return None
# This warning are pretty intruisive. Every time attribute is accessed
# a warning is being thrown. In neutron clients we have way too many
# places that still refer to tenant_id even though they may also
# properly support project_id. For now we silence tenant_id warnings.
if self.name != "tenant_id":
self.warn_if_deprecated_property(value)
return _convert_type(value, self.type, self.list_type)
def warn_if_deprecated_property(self, value):
deprecated = object.__getattribute__(self, 'deprecated')
deprecation_reason = object.__getattribute__(
self,
'deprecation_reason',
)
if value and deprecated:
warnings.warn(
"The field {!r} has been deprecated. {}".format(
self.name, deprecation_reason or "Avoid usage."
),
os_warnings.RemovedFieldWarning,
)
return value
def __set__(self, instance, value):
if self.coerce_to_default and value is None:
value = self.default
if value != self.default:
value = _convert_type(value, self.type, self.list_type)
attributes = getattr(instance, self.key)
attributes[self.name] = value
def __delete__(self, instance):
try:
attributes = getattr(instance, self.key)
del attributes[self.name]
except KeyError:
pass
)
class Body(_BaseComponent):
"""Body attributes"""
key = "_body"
def Header(
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
**kwargs,
):
return fields.Header(
name,
type=type,
default=default,
alias=alias,
aka=aka,
alternate_id=alternate_id,
list_type=list_type,
coerce_to_default=coerce_to_default,
deprecated=deprecated,
deprecation_reason=deprecation_reason,
**kwargs,
)
class Header(_BaseComponent):
"""Header attributes"""
key = "_header"
_map_cls = structures.CaseInsensitiveDict
def URI(
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
**kwargs,
):
return fields.URI(
name,
type=type,
default=default,
alias=alias,
aka=aka,
alternate_id=alternate_id,
list_type=list_type,
coerce_to_default=coerce_to_default,
deprecated=deprecated,
deprecation_reason=deprecation_reason,
**kwargs,
)
class URI(_BaseComponent):
"""URI attributes"""
key = "_uri"
class Computed(_BaseComponent):
"""Computed attributes"""
key = "_computed"
def Computed(
name,
type=None,
default=None,
alias=None,
aka=None,
alternate_id=False,
list_type=None,
coerce_to_default=False,
deprecated=False,
deprecation_reason=None,
**kwargs,
):
return fields.Computed(
name,
type=type,
default=default,
alias=alias,
aka=aka,
alternate_id=alternate_id,
list_type=list_type,
coerce_to_default=coerce_to_default,
deprecated=deprecated,
deprecation_reason=deprecation_reason,
**kwargs,
)
class _ComponentManager(collections.abc.MutableMapping):
@ -465,9 +360,9 @@ class Resource(dict):
id = Body("id")
#: The name of this resource.
name: ty.Union[Body, URI] = Body("name")
name: str = Body("name")
#: The OpenStack location of this resource.
location: ty.Union[Computed, Body, Header] = Computed('location')
location: dict[str, ty.Any] = Computed('location')
#: Mapping of accepted query parameter names.
_query_mapping = QueryParameters()
@ -599,7 +494,9 @@ class Resource(dict):
dict.update(self, self.to_dict())
@classmethod
def _attributes_iterator(cls, components=tuple([Body, Header])):
def _attributes_iterator(
cls, components=tuple([fields.Body, fields.Header])
):
"""Iterator over all Resource attributes"""
# isinstance stricly requires this to be a tuple
# Since we're looking at class definitions we need to include
@ -678,14 +575,14 @@ class Resource(dict):
# Not found? But we know an alias exists.
name = self._attr_aliases[name]
real_item = getattr(self.__class__, name, None)
if isinstance(real_item, _BaseComponent):
if isinstance(real_item, fields._BaseComponent):
return getattr(self, name)
if not real_item:
# In order to maintain backwards compatibility where we were
# returning Munch (and server side names) and Resource object with
# normalized attributes we can offer dict access via server side
# names.
for attr, component in self._attributes_iterator(tuple([Body])):
for attr, component in self._attributes_iterator((fields.Body,)):
if component.name == name:
warnings.warn(
f"Access to '{self.__class__}[{name}]' is deprecated. "
@ -703,7 +600,7 @@ class Resource(dict):
def __setitem__(self, name, value):
real_item = getattr(self.__class__, name, None)
if isinstance(real_item, _BaseComponent):
if isinstance(real_item, fields._BaseComponent):
self.__setattr__(name, value)
else:
if self._allow_unknown_attrs_in_body:
@ -722,7 +619,12 @@ class Resource(dict):
attributes = []
if not components:
components = tuple([Body, Header, Computed, URI])
components = (
fields.Body,
fields.Header,
fields.Computed,
fields.URI,
)
for attr, component in self._attributes_iterator(components):
key = attr if not remote_names else component.name
@ -841,13 +743,13 @@ class Resource(dict):
return {}
def _consume_body_attrs(self, attrs):
return self._consume_mapped_attrs(Body, attrs)
return self._consume_mapped_attrs(fields.Body, attrs)
def _consume_header_attrs(self, attrs):
return self._consume_mapped_attrs(Header, attrs)
return self._consume_mapped_attrs(fields.Header, attrs)
def _consume_uri_attrs(self, attrs):
return self._consume_mapped_attrs(URI, attrs)
return self._consume_mapped_attrs(fields.URI, attrs)
def _update_from_body_attrs(self, attrs):
body = self._consume_body_attrs(attrs)
@ -925,22 +827,22 @@ class Resource(dict):
@classmethod
def _body_mapping(cls):
"""Return all Body members of this class"""
return cls._get_mapping(Body)
return cls._get_mapping(fields.Body)
@classmethod
def _header_mapping(cls):
"""Return all Header members of this class"""
return cls._get_mapping(Header)
return cls._get_mapping(fields.Header)
@classmethod
def _uri_mapping(cls):
"""Return all URI members of this class"""
return cls._get_mapping(URI)
return cls._get_mapping(fields.URI)
@classmethod
def _computed_mapping(cls):
"""Return all URI members of this class"""
return cls._get_mapping(Computed)
"""Return all Computed members of this class"""
return cls._get_mapping(fields.Computed)
@classmethod
def _alternate_id(cls):
@ -953,7 +855,7 @@ class Resource(dict):
consumed by _get_id and passed to getattr.
"""
for value in cls.__dict__.values():
if isinstance(value, Body):
if isinstance(value, fields.Body):
if value.alternate_id:
return value.name
return ""
@ -1054,11 +956,11 @@ class Resource(dict):
):
"""Return a dictionary of this resource's contents
:param bool body: Include the :class:`~openstack.resource.Body`
:param bool body: Include the :class:`~openstack.fields.Body`
attributes in the returned dictionary.
:param bool headers: Include the :class:`~openstack.resource.Header`
:param bool headers: Include the :class:`~openstack.fields.Header`
attributes in the returned dictionary.
:param bool computed: Include the :class:`~openstack.resource.Computed`
:param bool computed: Include the :class:`~openstack.fields.Computed`
attributes in the returned dictionary.
:param bool ignore_none: When True, exclude key/value pairs where
the value is None. This will exclude attributes that the server
@ -1077,13 +979,13 @@ class Resource(dict):
else:
mapping = {}
components: list[type[_BaseComponent]] = []
components: list[type[fields._BaseComponent]] = []
if body:
components.append(Body)
components.append(fields.Body)
if headers:
components.append(Header)
components.append(fields.Header)
if computed:
components.append(Computed)
components.append(fields.Computed)
if not components:
raise ValueError(
"At least one of `body`, `headers` or `computed` must be True"
@ -2050,7 +1952,7 @@ class Resource(dict):
# Known attr
hasattr(cls, k)
# Is real attr property
and isinstance(getattr(cls, k), Body)
and isinstance(getattr(cls, k), fields.Body)
# not included in the query_params
and k not in cls._query_mapping._mapping.keys()
):
@ -2063,7 +1965,7 @@ class Resource(dict):
for k, v in params.items():
# We need to gather URI parts to set them on the resource later
if hasattr(cls, k) and isinstance(getattr(cls, k), URI):
if hasattr(cls, k) and isinstance(getattr(cls, k), fields.URI):
uri_params[k] = v
def _dict_filter(f, d):

7
openstack/test/fakes.py
View File

@ -31,6 +31,7 @@ from collections.abc import Generator
from unittest import mock
import uuid
from openstack import fields
from openstack import format as _format
from openstack import proxy
from openstack import resource
@ -68,9 +69,9 @@ def generate_fake_resource(
base_attrs: dict[str, Any] = {}
for name, value in inspect.getmembers(
resource_type,
predicate=lambda x: isinstance(x, (resource.Body, resource.URI)),
predicate=lambda x: isinstance(x, (fields.Body, fields.URI)),
):
if isinstance(value, resource.Body):
if isinstance(value, fields.Body):
target_type = value.type
if target_type is None:
if (
@ -128,7 +129,7 @@ def generate_fake_resource(
msg = f"Fake value for {resource_type.__name__}.{name} can not be generated"
raise NotImplementedError(msg)
if isinstance(value, resource.URI):
if isinstance(value, fields.URI):
# For URI we just generate something
base_attrs[name] = uuid.uuid4().hex

224
openstack/tests/unit/test_fields.py Normal file
View File

@ -0,0 +1,224 @@
# 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.
from openstack import fields
from openstack import format
from openstack.tests.unit import base
class TestComponent(base.TestCase):
class ExampleComponent(fields._BaseComponent):
key = "_example"
# Since we're testing ExampleComponent, which is as isolated as we
# can test _BaseComponent due to it's needing to be a data member
# of a class that has an attribute on the parent class named `key`,
# each test has to implement a class with a name that is the same
# as ExampleComponent.key, which should be a dict containing the
# keys and values to test against.
def test_implementations(self):
self.assertEqual("_body", fields.Body.key)
self.assertEqual("_header", fields.Header.key)
self.assertEqual("_uri", fields.URI.key)
def test_creation(self):
sot = fields._BaseComponent(
"name", type=int, default=1, alternate_id=True, aka="alias"
)
self.assertEqual("name", sot.name)
self.assertEqual(int, sot.type)
self.assertEqual(1, sot.default)
self.assertEqual("alias", sot.aka)
self.assertTrue(sot.alternate_id)
def test_get_no_instance(self):
sot = fields._BaseComponent("test")
# Test that we short-circuit everything when given no instance.
result = sot.__get__(None, None)
self.assertIs(sot, result)
# NOTE: Some tests will use a default=1 setting when testing result
# values that should be None because the default-for-default is also None.
def test_get_name_None(self):
name = "name"
class Parent:
_example = {name: None}
instance = Parent()
sot = TestComponent.ExampleComponent(name, default=1)
# Test that we short-circuit any typing of a None value.
result = sot.__get__(instance, None)
self.assertIsNone(result)
def test_get_default(self):
expected_result = 123
class Parent:
_example = {}
instance = Parent()
# NOTE: type=dict but the default value is an int. If we didn't
# short-circuit the typing part of __get__ it would fail.
sot = TestComponent.ExampleComponent(
"name", type=dict, default=expected_result
)
# Test that we directly return any default value.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
def test_get_name_untyped(self):
name = "name"
expected_result = 123
class Parent:
_example = {name: expected_result}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
# Test that we return any the value as it is set.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
# The code path for typing after a raw value has been found is the same.
def test_get_name_typed(self):
name = "name"
value = "123"
class Parent:
_example = {name: value}
instance = Parent()
sot = TestComponent.ExampleComponent("name", type=int)
# Test that we run the underlying value through type conversion.
result = sot.__get__(instance, None)
self.assertEqual(int(value), result)
def test_get_name_formatter(self):
name = "name"
value = "123"
expected_result = "one hundred twenty three"
class Parent:
_example = {name: value}
class FakeFormatter(format.Formatter):
@classmethod
def deserialize(cls, value):
return expected_result
instance = Parent()
sot = TestComponent.ExampleComponent("name", type=FakeFormatter)
# Mock out issubclass rather than having an actual format.Formatter
# This can't be mocked via decorator, isolate it to wrapping the call.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
def test_set_name_untyped(self):
name = "name"
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
# Test that we don't run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual(expected_value, instance._example[name])
def test_set_name_typed(self):
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
# The type we give to ExampleComponent has to be an actual type,
# not an instance, so we can't get the niceties of a mock.Mock
# instance that would allow us to call `assert_called_once_with` to
# ensure that we're sending the value through the type.
# Instead, we use this tiny version of a similar thing.
class FakeType:
calls = []
def __init__(self, arg):
FakeType.calls.append(arg)
sot = TestComponent.ExampleComponent("name", type=FakeType)
# Test that we run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual([expected_value], FakeType.calls)
def test_set_name_formatter(self):
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
# As with test_set_name_typed, create a pseudo-Mock to track what
# gets called on the type.
class FakeFormatter(format.Formatter):
calls = []
@classmethod
def deserialize(cls, arg):
FakeFormatter.calls.append(arg)
sot = TestComponent.ExampleComponent("name", type=FakeFormatter)
# Test that we run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual([expected_value], FakeFormatter.calls)
def test_delete_name(self):
name = "name"
expected_value = "123"
class Parent:
_example = {name: expected_value}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
sot.__delete__(instance)
self.assertNotIn(name, instance._example)
def test_delete_name_doesnt_exist(self):
name = "name"
expected_value = "123"
class Parent:
_example = {"what": expected_value}
instance = Parent()
sot = TestComponent.ExampleComponent(name)
sot.__delete__(instance)
self.assertNotIn(name, instance._example)

221
openstack/tests/unit/test_resource.py
View File

@ -20,7 +20,7 @@ import requests
from openstack import dns
from openstack import exceptions
from openstack import format
from openstack import fields
from openstack import resource
from openstack.tests.unit import base
from openstack import utils
@ -37,215 +37,6 @@ class FakeResponse:
return self.body
class TestComponent(base.TestCase):
class ExampleComponent(resource._BaseComponent):
key = "_example"
# Since we're testing ExampleComponent, which is as isolated as we
# can test _BaseComponent due to it's needing to be a data member
# of a class that has an attribute on the parent class named `key`,
# each test has to implement a class with a name that is the same
# as ExampleComponent.key, which should be a dict containing the
# keys and values to test against.
def test_implementations(self):
self.assertEqual("_body", resource.Body.key)
self.assertEqual("_header", resource.Header.key)
self.assertEqual("_uri", resource.URI.key)
def test_creation(self):
sot = resource._BaseComponent(
"name", type=int, default=1, alternate_id=True, aka="alias"
)
self.assertEqual("name", sot.name)
self.assertEqual(int, sot.type)
self.assertEqual(1, sot.default)
self.assertEqual("alias", sot.aka)
self.assertTrue(sot.alternate_id)
def test_get_no_instance(self):
sot = resource._BaseComponent("test")
# Test that we short-circuit everything when given no instance.
result = sot.__get__(None, None)
self.assertIs(sot, result)
# NOTE: Some tests will use a default=1 setting when testing result
# values that should be None because the default-for-default is also None.
def test_get_name_None(self):
name = "name"
class Parent:
_example = {name: None}
instance = Parent()
sot = TestComponent.ExampleComponent(name, default=1)
# Test that we short-circuit any typing of a None value.
result = sot.__get__(instance, None)
self.assertIsNone(result)
def test_get_default(self):
expected_result = 123
class Parent:
_example = {}
instance = Parent()
# NOTE: type=dict but the default value is an int. If we didn't
# short-circuit the typing part of __get__ it would fail.
sot = TestComponent.ExampleComponent(
"name", type=dict, default=expected_result
)
# Test that we directly return any default value.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
def test_get_name_untyped(self):
name = "name"
expected_result = 123
class Parent:
_example = {name: expected_result}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
# Test that we return any the value as it is set.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
# The code path for typing after a raw value has been found is the same.
def test_get_name_typed(self):
name = "name"
value = "123"
class Parent:
_example = {name: value}
instance = Parent()
sot = TestComponent.ExampleComponent("name", type=int)
# Test that we run the underlying value through type conversion.
result = sot.__get__(instance, None)
self.assertEqual(int(value), result)
def test_get_name_formatter(self):
name = "name"
value = "123"
expected_result = "one hundred twenty three"
class Parent:
_example = {name: value}
class FakeFormatter(format.Formatter):
@classmethod
def deserialize(cls, value):
return expected_result
instance = Parent()
sot = TestComponent.ExampleComponent("name", type=FakeFormatter)
# Mock out issubclass rather than having an actual format.Formatter
# This can't be mocked via decorator, isolate it to wrapping the call.
result = sot.__get__(instance, None)
self.assertEqual(expected_result, result)
def test_set_name_untyped(self):
name = "name"
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
# Test that we don't run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual(expected_value, instance._example[name])
def test_set_name_typed(self):
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
# The type we give to ExampleComponent has to be an actual type,
# not an instance, so we can't get the niceties of a mock.Mock
# instance that would allow us to call `assert_called_once_with` to
# ensure that we're sending the value through the type.
# Instead, we use this tiny version of a similar thing.
class FakeType:
calls = []
def __init__(self, arg):
FakeType.calls.append(arg)
sot = TestComponent.ExampleComponent("name", type=FakeType)
# Test that we run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual([expected_value], FakeType.calls)
def test_set_name_formatter(self):
expected_value = "123"
class Parent:
_example = {}
instance = Parent()
# As with test_set_name_typed, create a pseudo-Mock to track what
# gets called on the type.
class FakeFormatter(format.Formatter):
calls = []
@classmethod
def deserialize(cls, arg):
FakeFormatter.calls.append(arg)
sot = TestComponent.ExampleComponent("name", type=FakeFormatter)
# Test that we run the value through type conversion.
sot.__set__(instance, expected_value)
self.assertEqual([expected_value], FakeFormatter.calls)
def test_delete_name(self):
name = "name"
expected_value = "123"
class Parent:
_example = {name: expected_value}
instance = Parent()
sot = TestComponent.ExampleComponent("name")
sot.__delete__(instance)
self.assertNotIn(name, instance._example)
def test_delete_name_doesnt_exist(self):
name = "name"
expected_value = "123"
class Parent:
_example = {"what": expected_value}
instance = Parent()
sot = TestComponent.ExampleComponent(name)
sot.__delete__(instance)
self.assertNotIn(name, instance._example)
class TestComponentManager(base.TestCase):
def test_create_basic(self):
sot = resource._ComponentManager()
@ -747,16 +538,12 @@ class TestResource(base.TestCase):
self.assertEqual(
sorted(['bar', '_bar', 'bar_local', 'id', 'name', 'location']),
sorted(
sot._attributes(
components=tuple([resource.Body, resource.Computed])
)
),
sorted(sot._attributes(components=(fields.Body, fields.Computed))),
)
self.assertEqual(
('foo',),
tuple(sot._attributes(components=tuple([resource.Header]))),
tuple(sot._attributes(components=(fields.Header,))),
)
def test__attributes_iterator(self):
@ -780,7 +567,7 @@ class TestResource(base.TestCase):
# Check we iterate only over headers
for attr, component in sot._attributes_iterator(
components=tuple([resource.Header])
components=(fields.Header,)
):
if attr in expected:
expected.remove(attr)