openstack/openstacksdk
Code Issues Proposed changes

Merge "Move tag mixin into the common" into feature/r1

Browse Source
This commit is contained in:
Zuul 2021-07-16 08:54:03 +00:00 committed by Gerrit Code Review
commit 21c3c2d6ec
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())