openstack/openstacksdk
Code Issues Proposed changes

Move tag mixin into the common

Browse Source 
As a continuation of a further generalization of methods add a common
module and move tag there.

Change-Id: Id23145e427221e7bd80696b9faec2d4c8d76b477
This commit is contained in:
Artem Goncharov 2021-05-17 17:36:08 +02:00 committed by Artem Goncharov
parent 8a1d255ca8
commit c6494db0ac
27 changed files with 371 additions and 337 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
openstack/common/__init__.py Normal file
View File

127
openstack/common/tag.py Normal file
View File

@ -0,0 +1,127 @@
# 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 exceptions
from openstack import resource
from openstack import utils
class TagMixin:
_tag_query_parameters = {
'tags': 'tags',
'any_tags': 'tags-any',
'not_tags': 'not-tags',
'not_any_tags': 'not-tags-any',
}
#: A list of associated tags
#: *Type: list of tag strings*
tags = resource.Body('tags', type=list, default=[])
def fetch_tags(self, session):
"""Lists tags set on the entity.
:param session: The session to use for making this request.
:return: The list with tags attached to the entity
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(response)
# NOTE(gtema): since this is a common method
# we can't rely on the resource_key, because tags are returned
# without resource_key. Do parse response here
json = response.json()
if 'tags' in json:
self._body.attributes.update({'tags': json['tags']})
return self
def set_tags(self, session, tags=[]):
"""Sets/Replaces all tags on the resource.
:param session: The session to use for making this request.
:param list tags: List with tags to be set on the resource
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.put(url, json={'tags': tags})
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': tags})
return self
def remove_all_tags(self, session):
"""Removes all tags on the entity.
:param session: The session to use for making this request.
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': []})
return self
def check_tag(self, session, tag):
"""Checks if tag exists on the entity.
If the tag does not exist a 404 will be returned
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(response,
error_message='Tag does not exist')
return self
def add_tag(self, session, tag):
"""Adds a single tag to the resource.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.put(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
tags.append(tag)
self._body.attributes.update({
'tags': tags
})
return self
def remove_tag(self, session, tag):
"""Removes a single tag from the specified server.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
try:
# NOTE(gtema): if tags were not fetched, but request suceeded
# it is ok. Just ensure tag does not exist locally
tags.remove(tag)
except ValueError:
pass # do nothing!
self._body.attributes.update({
'tags': tags
})
return self

6
openstack/compute/v2/server.py
View File

@ -9,7 +9,7 @@
# 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.common import tag
from openstack.compute.v2 import metadata
from openstack import exceptions
from openstack.image.v2 import image
@ -26,7 +26,7 @@ CONSOLE_TYPE_ACTION_MAPPING = {
}
class Server(resource.Resource, metadata.MetadataMixin, resource.TagMixin):
class Server(resource.Resource, metadata.MetadataMixin, tag.TagMixin):
resource_key = 'server'
resources_key = 'servers'
base_path = '/servers'
@ -60,7 +60,7 @@ class Server(resource.Resource, metadata.MetadataMixin, resource.TagMixin):
changes_before="changes-before",
id="uuid",
all_projects="all_tenants",
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
_max_microversion = '2.72'

6
openstack/identity/v3/project.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack import resource
from openstack import utils
class Project(resource.Resource, resource.TagMixin):
class Project(resource.Resource, tag.TagMixin):
resource_key = 'project'
resources_key = 'projects'
base_path = '/projects'
@ -33,7 +33,7 @@ class Project(resource.Resource, resource.TagMixin):
'name',
'parent_id',
is_enabled='enabled',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

3
openstack/image/v2/image.py
View File

@ -9,13 +9,14 @@
# 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.common import tag
from openstack import exceptions
from openstack.image import _download
from openstack import resource
from openstack import utils
class Image(resource.Resource, resource.TagMixin, _download.DownloadMixin):
class Image(resource.Resource, tag.TagMixin, _download.DownloadMixin):
resources_key = 'images'
base_path = '/images'

6
openstack/load_balancer/v2/health_monitor.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class HealthMonitor(resource.Resource, resource.TagMixin):
class HealthMonitor(resource.Resource, tag.TagMixin):
resource_key = 'healthmonitor'
resources_key = 'healthmonitors'
base_path = '/lbaas/healthmonitors'
@ -30,7 +30,7 @@ class HealthMonitor(resource.Resource, resource.TagMixin):
'http_method', 'max_retries', 'max_retries_down', 'pool_id',
'provisioning_status', 'operating_status', 'timeout',
'project_id', 'type', 'url_path', is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
#: Properties

6
openstack/load_balancer/v2/l7_policy.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class L7Policy(resource.Resource, resource.TagMixin):
class L7Policy(resource.Resource, tag.TagMixin):
resource_key = 'l7policy'
resources_key = 'l7policies'
base_path = '/lbaas/l7policies'
@ -30,7 +30,7 @@ class L7Policy(resource.Resource, resource.TagMixin):
'redirect_pool_id', 'redirect_url', 'provisioning_status',
'operating_status', 'redirect_prefix', 'project_id',
is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
#: Properties

6
openstack/load_balancer/v2/l7_rule.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class L7Rule(resource.Resource, resource.TagMixin):
class L7Rule(resource.Resource, tag.TagMixin):
resource_key = 'rule'
resources_key = 'rules'
base_path = '/lbaas/l7policies/%(l7policy_id)s/rules'
@ -29,7 +29,7 @@ class L7Rule(resource.Resource, resource.TagMixin):
'compare_type', 'created_at', 'invert', 'key', 'project_id',
'provisioning_status', 'type', 'updated_at', 'rule_value',
'operating_status', is_admin_state_up='admin_state_up',
l7_policy_id='l7policy_id', **resource.TagMixin._tag_query_parameters
l7_policy_id='l7policy_id', **tag.TagMixin._tag_query_parameters
)
#: Properties

6
openstack/load_balancer/v2/listener.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class Listener(resource.Resource, resource.TagMixin):
class Listener(resource.Resource, tag.TagMixin):
resource_key = 'listener'
resources_key = 'listeners'
base_path = '/lbaas/listeners'
@ -34,7 +34,7 @@ class Listener(resource.Resource, resource.TagMixin):
'timeout_member_data', 'timeout_tcp_inspect', 'allowed_cidrs',
'tls_ciphers', 'tls_versions', 'alpn_protocols',
is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/load_balancer/v2/load_balancer.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class LoadBalancer(resource.Resource, resource.TagMixin):
class LoadBalancer(resource.Resource, tag.TagMixin):
resource_key = 'loadbalancer'
resources_key = 'loadbalancers'
base_path = '/lbaas/loadbalancers'
@ -30,7 +30,7 @@ class LoadBalancer(resource.Resource, resource.TagMixin):
'vip_address', 'vip_network_id', 'vip_port_id', 'vip_subnet_id',
'vip_qos_policy_id', 'provisioning_status', 'operating_status',
'availability_zone', is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/load_balancer/v2/member.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class Member(resource.Resource, resource.TagMixin):
class Member(resource.Resource, tag.TagMixin):
resource_key = 'member'
resources_key = 'members'
base_path = '/lbaas/pools/%(pool_id)s/members'
@ -30,7 +30,7 @@ class Member(resource.Resource, resource.TagMixin):
'created_at', 'updated_at', 'provisioning_status', 'operating_status',
'project_id', 'monitor_address', 'monitor_port', 'backup',
is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/load_balancer/v2/pool.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class Pool(resource.Resource, resource.TagMixin):
class Pool(resource.Resource, tag.TagMixin):
resource_key = 'pool'
resources_key = 'pools'
base_path = '/lbaas/pools'
@ -31,7 +31,7 @@ class Pool(resource.Resource, resource.TagMixin):
'created_at', 'updated_at', 'provisioning_status', 'operating_status',
'tls_enabled', 'tls_ciphers', 'tls_versions', 'alpn_protocols',
is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
#: Properties

6
openstack/network/v2/floating_ip.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class FloatingIP(_base.NetworkResource, resource.TagMixin):
class FloatingIP(_base.NetworkResource, tag.TagMixin):
name_attribute = "floating_ip_address"
resource_name = "floating ip"
resource_key = 'floatingip'
@ -35,7 +35,7 @@ class FloatingIP(_base.NetworkResource, resource.TagMixin):
'port_id', 'router_id', 'status', 'subnet_id',
'project_id', 'tenant_id',
tenant_id='project_id',
**resource.TagMixin._tag_query_parameters)
**tag.TagMixin._tag_query_parameters)
# Properties
#: Timestamp at which the floating IP was created.

6
openstack/network/v2/network.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class Network(_base.NetworkResource, resource.TagMixin):
class Network(_base.NetworkResource, tag.TagMixin):
resource_key = 'network'
resources_key = 'networks'
base_path = '/networks'
@ -39,7 +39,7 @@ class Network(_base.NetworkResource, resource.TagMixin):
provider_network_type='provider:network_type',
provider_physical_network='provider:physical_network',
provider_segmentation_id='provider:segmentation_id',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/port.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class Port(_base.NetworkResource, resource.TagMixin):
class Port(_base.NetworkResource, tag.TagMixin):
resource_key = 'port'
resources_key = 'ports'
base_path = '/ports'
@ -35,7 +35,7 @@ class Port(_base.NetworkResource, resource.TagMixin):
'subnet_id', 'project_id',
is_admin_state_up='admin_state_up',
is_port_security_enabled='port_security_enabled',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/qos_policy.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack import resource
from openstack import utils
class QoSPolicy(resource.Resource, resource.TagMixin):
class QoSPolicy(resource.Resource, tag.TagMixin):
resource_key = 'policy'
resources_key = 'policies'
base_path = '/qos/policies'
@ -32,7 +32,7 @@ class QoSPolicy(resource.Resource, resource.TagMixin):
'name', 'description', 'is_default',
'project_id',
is_shared='shared',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/router.py
View File

@ -9,14 +9,14 @@
# 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.common import tag
from openstack import exceptions
from openstack.network.v2 import _base
from openstack import resource
from openstack import utils
class Router(_base.NetworkResource, resource.TagMixin):
class Router(_base.NetworkResource, tag.TagMixin):
resource_key = 'router'
resources_key = 'routers'
base_path = '/routers'
@ -34,7 +34,7 @@ class Router(_base.NetworkResource, resource.TagMixin):
is_admin_state_up='admin_state_up',
is_distributed='distributed',
is_ha='ha',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/security_group.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class SecurityGroup(_base.NetworkResource, resource.TagMixin):
class SecurityGroup(_base.NetworkResource, tag.TagMixin):
resource_key = 'security_group'
resources_key = 'security_groups'
base_path = '/security-groups'
@ -29,7 +29,7 @@ class SecurityGroup(_base.NetworkResource, resource.TagMixin):
_query_mapping = resource.QueryParameters(
'description', 'fields', 'id', 'name', 'stateful', 'project_id',
'tenant_id', 'revision_number', 'sort_dir', 'sort_key',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

7
openstack/network/v2/security_group_rule.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class SecurityGroupRule(_base.NetworkResource, resource.TagMixin):
class SecurityGroupRule(_base.NetworkResource, tag.TagMixin):
resource_key = 'security_group_rule'
resources_key = 'security_group_rules'
base_path = '/security-group-rules'
@ -35,8 +35,7 @@ class SecurityGroupRule(_base.NetworkResource, resource.TagMixin):
'project_id', 'tenant_id',
'sort_dir', 'sort_key',
ether_type='ethertype',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/subnet.py
View File

@ -9,12 +9,12 @@
# 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.common import tag
from openstack.network.v2 import _base
from openstack import resource
class Subnet(_base.NetworkResource, resource.TagMixin):
class Subnet(_base.NetworkResource, tag.TagMixin):
resource_key = 'subnet'
resources_key = 'subnets'
base_path = '/subnets'
@ -34,7 +34,7 @@ class Subnet(_base.NetworkResource, resource.TagMixin):
is_dhcp_enabled='enable_dhcp',
subnet_pool_id='subnetpool_id',
use_default_subnet_pool='use_default_subnetpool',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/subnet_pool.py
View File

@ -9,11 +9,11 @@
# 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.common import tag
from openstack import resource
class SubnetPool(resource.Resource, resource.TagMixin):
class SubnetPool(resource.Resource, tag.TagMixin):
resource_key = 'subnetpool'
resources_key = 'subnetpools'
base_path = '/subnetpools'
@ -31,7 +31,7 @@ class SubnetPool(resource.Resource, resource.TagMixin):
'address_scope_id', 'description', 'ip_version', 'is_default',
'name', 'project_id',
is_shared='shared',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

6
openstack/network/v2/trunk.py
View File

@ -9,13 +9,13 @@
# 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.common import tag
from openstack import exceptions
from openstack import resource
from openstack import utils
class Trunk(resource.Resource, resource.TagMixin):
class Trunk(resource.Resource, tag.TagMixin):
resource_key = 'trunk'
resources_key = 'trunks'
base_path = '/trunks'
@ -33,7 +33,7 @@ class Trunk(resource.Resource, resource.TagMixin):
'name', 'description', 'port_id', 'status', 'sub_ports',
'project_id',
is_admin_state_up='admin_state_up',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

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

@ -9,7 +9,7 @@
# 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.common import tag
from openstack import exceptions
from openstack import resource
from openstack import utils
@ -32,7 +32,7 @@ class Stack(resource.Resource):
'action', 'name', 'status',
'project_id', 'owner_id', 'username',
project_id='tenant_id',
**resource.TagMixin._tag_query_parameters
**tag.TagMixin._tag_query_parameters
)
# Properties

113
openstack/resource.py
View File

@ -1980,119 +1980,6 @@ class Resource(dict):
"No %s found for %s" % (cls.__name__, name_or_id))
class TagMixin:
_tag_query_parameters = {
'tags': 'tags',
'any_tags': 'tags-any',
'not_tags': 'not-tags',
'not_any_tags': 'not-tags-any',
}
#: A list of associated tags
#: *Type: list of tag strings*
tags = Body('tags', type=list, default=[])
def fetch_tags(self, session):
"""Lists tags set on the entity.
:param session: The session to use for making this request.
:return: The list with tags attached to the entity
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(response)
# NOTE(gtema): since this is a common method
# we can't rely on the resource_key, because tags are returned
# without resource_key. Do parse response here
json = response.json()
if 'tags' in json:
self._body.attributes.update({'tags': json['tags']})
return self
def set_tags(self, session, tags=[]):
"""Sets/Replaces all tags on the resource.
:param session: The session to use for making this request.
:param list tags: List with tags to be set on the resource
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.put(url, json={'tags': tags})
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': tags})
return self
def remove_all_tags(self, session):
"""Removes all tags on the entity.
:param session: The session to use for making this request.
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': []})
return self
def check_tag(self, session, tag):
"""Checks if tag exists on the entity.
If the tag does not exist a 404 will be returned
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(response,
error_message='Tag does not exist')
return self
def add_tag(self, session, tag):
"""Adds a single tag to the resource.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.put(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
tags.append(tag)
self._body.attributes.update({
'tags': tags
})
return self
def remove_tag(self, session, tag):
"""Removes a single tag from the specified server.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
try:
# NOTE(gtema): if tags were not fetched, but request suceeded
# it is ok. Just ensure tag does not exist locally
tags.remove(tag)
except ValueError:
pass # do nothing!
self._body.attributes.update({
'tags': tags
})
return self
def _normalize_status(status):
if status is not None:
status = status.lower()

0
openstack/tests/unit/common/__init__.py Normal file
View File

183
openstack/tests/unit/common/test_tag.py Normal file
View File

@ -0,0 +1,183 @@
# 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 unittest import mock
from keystoneauth1 import adapter
from openstack.common import tag
from openstack import exceptions
from openstack import resource
from openstack.tests.unit import base
from openstack.tests.unit.test_resource import FakeResponse
class TestTagMixin(base.TestCase):
def setUp(self):
super(TestTagMixin, self).setUp()
self.service_name = "service"
self.base_path = "base_path"
class Test(resource.Resource, tag.TagMixin):
service = self.service_name
base_path = self.base_path
resources_key = 'resources'
allow_create = True
allow_fetch = True
allow_head = True
allow_commit = True
allow_delete = True
allow_list = True
self.test_class = Test
self.request = mock.Mock(spec=resource._Request)
self.request.url = "uri"
self.request.body = "body"
self.request.headers = "headers"
self.response = FakeResponse({})
self.sot = Test.new(id="id", tags=[])
self.sot._prepare_request = mock.Mock(return_value=self.request)
self.sot._translate_response = mock.Mock()
self.session = mock.Mock(spec=adapter.Adapter)
self.session.get = mock.Mock(return_value=self.response)
self.session.put = mock.Mock(return_value=self.response)
self.session.delete = mock.Mock(return_value=self.response)
def test_tags_attribute(self):
res = self.sot
self.assertTrue(hasattr(res, 'tags'))
self.assertIsInstance(res.tags, list)
def test_fetch_tags(self):
res = self.sot
sess = self.session
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.links = {}
mock_response.json.return_value = {'tags': ['blue1', 'green1']}
sess.get.side_effect = [mock_response]
result = res.fetch_tags(sess)
# Check tags attribute is updated
self.assertEqual(['blue1', 'green1'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.get.assert_called_once_with(url)
def test_set_tags(self):
res = self.sot
sess = self.session
# Set some initial value to check rewrite
res.tags.extend(['blue_old', 'green_old'])
result = res.set_tags(sess, ['blue', 'green'])
# Check tags attribute is updated
self.assertEqual(['blue', 'green'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.put.assert_called_once_with(
url,
json={'tags': ['blue', 'green']}
)
def test_remove_all_tags(self):
res = self.sot
sess = self.session
# Set some initial value to check removal
res.tags.extend(['blue_old', 'green_old'])
result = res.remove_all_tags(sess)
# Check tags attribute is updated
self.assertEqual([], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.delete.assert_called_once_with(url)
def test_remove_single_tag(self):
res = self.sot
sess = self.session
res.tags.extend(['blue', 'dummy'])
result = res.remove_tag(sess, 'dummy')
# Check tags attribute is updated
self.assertEqual(['blue'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/dummy'
sess.delete.assert_called_once_with(url)
def test_check_tag_exists(self):
res = self.sot
sess = self.session
sess.get.side_effect = [FakeResponse(None, 202)]
result = res.check_tag(sess, 'blue')
# Check tags attribute is updated
self.assertEqual([], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/blue'
sess.get.assert_called_once_with(url)
def test_check_tag_not_exists(self):
res = self.sot
sess = self.session
mock_response = mock.Mock()
mock_response.status_code = 404
mock_response.links = {}
mock_response.content = None
sess.get.side_effect = [mock_response]
# ensure we get 404
self.assertRaises(
exceptions.NotFoundException,
res.check_tag,
sess,
'dummy',
)
def test_add_tag(self):
res = self.sot
sess = self.session
# Set some initial value to check add
res.tags.extend(['blue', 'green'])
result = res.add_tag(sess, 'lila')
# Check tags attribute is updated
self.assertEqual(['blue', 'green', 'lila'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/lila'
sess.put.assert_called_once_with(url)
def test_tagged_resource_always_created_with_empty_tag_list(self):
res = self.sot
self.assertIsNotNone(res.tags)
self.assertEqual(res.tags, list())

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

@ -3121,166 +3121,3 @@ class TestAssertMicroversionFor(base.TestCase):
self.res._assert_microversion_for,
self.session, 'fetch', '1.6')
mock_get_ver.assert_called_once_with(self.res, self.session, 'fetch')
class TestTagMixin(base.TestCase):
def setUp(self):
super(TestTagMixin, self).setUp()
self.service_name = "service"
self.base_path = "base_path"
class Test(resource.Resource, resource.TagMixin):
service = self.service_name
base_path = self.base_path
resources_key = 'resources'
allow_create = True
allow_fetch = True
allow_head = True
allow_commit = True
allow_delete = True
allow_list = True
self.test_class = Test
self.request = mock.Mock(spec=resource._Request)
self.request.url = "uri"
self.request.body = "body"
self.request.headers = "headers"
self.response = FakeResponse({})
self.sot = Test.new(id="id", tags=[])
self.sot._prepare_request = mock.Mock(return_value=self.request)
self.sot._translate_response = mock.Mock()
self.session = mock.Mock(spec=adapter.Adapter)
self.session.get = mock.Mock(return_value=self.response)
self.session.put = mock.Mock(return_value=self.response)
self.session.delete = mock.Mock(return_value=self.response)
def test_tags_attribute(self):
res = self.sot
self.assertTrue(hasattr(res, 'tags'))
self.assertIsInstance(res.tags, list)
def test_fetch_tags(self):
res = self.sot
sess = self.session
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.links = {}
mock_response.json.return_value = {'tags': ['blue1', 'green1']}
sess.get.side_effect = [mock_response]
result = res.fetch_tags(sess)
# Check tags attribute is updated
self.assertEqual(['blue1', 'green1'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.get.assert_called_once_with(url)
def test_set_tags(self):
res = self.sot
sess = self.session
# Set some initial value to check rewrite
res.tags.extend(['blue_old', 'green_old'])
result = res.set_tags(sess, ['blue', 'green'])
# Check tags attribute is updated
self.assertEqual(['blue', 'green'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.put.assert_called_once_with(
url,
json={'tags': ['blue', 'green']}
)
def test_remove_all_tags(self):
res = self.sot
sess = self.session
# Set some initial value to check removal
res.tags.extend(['blue_old', 'green_old'])
result = res.remove_all_tags(sess)
# Check tags attribute is updated
self.assertEqual([], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags'
sess.delete.assert_called_once_with(url)
def test_remove_single_tag(self):
res = self.sot
sess = self.session
res.tags.extend(['blue', 'dummy'])
result = res.remove_tag(sess, 'dummy')
# Check tags attribute is updated
self.assertEqual(['blue'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/dummy'
sess.delete.assert_called_once_with(url)
def test_check_tag_exists(self):
res = self.sot
sess = self.session
sess.get.side_effect = [FakeResponse(None, 202)]
result = res.check_tag(sess, 'blue')
# Check tags attribute is updated
self.assertEqual([], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/blue'
sess.get.assert_called_once_with(url)
def test_check_tag_not_exists(self):
res = self.sot
sess = self.session
mock_response = mock.Mock()
mock_response.status_code = 404
mock_response.links = {}
mock_response.content = None
sess.get.side_effect = [mock_response]
# ensure we get 404
self.assertRaises(
exceptions.NotFoundException,
res.check_tag,
sess,
'dummy',
)
def test_add_tag(self):
res = self.sot
sess = self.session
# Set some initial value to check add
res.tags.extend(['blue', 'green'])
result = res.add_tag(sess, 'lila')
# Check tags attribute is updated
self.assertEqual(['blue', 'green', 'lila'], res.tags)
# Check the passed resource is returned
self.assertEqual(res, result)
url = self.base_path + '/' + res.id + '/tags/lila'
sess.put.assert_called_once_with(url)
def test_tagged_resource_always_created_with_empty_tag_list(self):
res = self.sot
self.assertIsNotNone(res.tags)
self.assertEqual(res.tags, list())