openstack
/
tempest
Code Issues Proposed changes

Merge "Add unit tests for volume and snapshot manage clients"

This commit is contained in:
Jenkins 2017-07-20 11:19:48 +00:00 committed by Gerrit Code Review
parent 724c86b02d bf549a4432
commit d3fa46495a
2 changed files with 194 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
tempest/tests/lib/services/volume/v2/test_snapshot_manage_client.py Normal file
View File

@ -0,0 +1,83 @@
# Copyright 2017 FiberHome Telecommunication Technologies CO.,LTD
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_serialization import jsonutils as json
from tempest.lib.services.volume.v2 import snapshot_manage_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestSnapshotManageClient(base.BaseServiceTest):
SNAPSHOT_MANAGE_REQUEST = {
"snapshot": {
"description": "snapshot-manage-description",
"metadata": None,
"ref": {
"source-name": "_snapshot-22b71da0-94f9-4aca-ad45-7522b3fa96bb"
},
"name": "snapshot-managed",
"volume_id": "7c064b34-1e4b-40bd-93ca-4ac5a973661b"
}
}
SNAPSHOT_MANAGE_RESPONSE = {
"snapshot": {
"status": "creating",
"description": "snapshot-manage-description",
"updated_at": None,
"volume_id": "32bafcc8-7109-42cd-8342-70d8de2bedef",
"id": "8fd6eb9d-0a82-456d-b1ec-dea4ac7f1ee2",
"size": 1,
"name": "snapshot-managed",
"created_at": "2017-07-11T10:07:58.000000",
"metadata": {}
}
}
def setUp(self):
super(TestSnapshotManageClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = snapshot_manage_client.SnapshotManageClient(fake_auth,
'volume',
'regionOne')
def _test_manage_snapshot(self, bytes_body=False):
payload = json.dumps(self.SNAPSHOT_MANAGE_REQUEST, sort_keys=True)
json_dumps = json.dumps
# NOTE: Use sort_keys for json.dumps so that the expected and actual
# payloads are guaranteed to be identical for mock_args assert check.
with mock.patch.object(snapshot_manage_client.json,
'dumps') as mock_dumps:
mock_dumps.side_effect = lambda d: json_dumps(d, sort_keys=True)
self.check_service_client_function(
self.client.manage_snapshot,
'tempest.lib.common.rest_client.RestClient.post',
self.SNAPSHOT_MANAGE_RESPONSE,
to_utf=bytes_body,
status=202,
mock_args=['os-snapshot-manage', payload],
**self.SNAPSHOT_MANAGE_REQUEST['snapshot'])
def test_manage_snapshot_with_str_body(self):
self._test_manage_snapshot()
def test_manage_snapshot_with_bytes_body(self):
self._test_manage_snapshot(bytes_body=True)

111
tempest/tests/lib/services/volume/v2/test_volume_manage_client.py Normal file
View File

@ -0,0 +1,111 @@
# Copyright 2017 FiberHome Telecommunication Technologies CO.,LTD
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo_serialization import jsonutils as json
from tempest.lib.services.volume.v2 import volume_manage_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestVolumeManageClient(base.BaseServiceTest):
VOLUME_MANAGE_REQUEST = {
"volume": {
"host": "controller1@rbd#rbd",
"name": "volume-managed",
"availability_zone": "nova",
"bootable": False,
"metadata": None,
"ref": {
"source-name": "volume-2ce6ca46-e6c1-4fe5-8268-3a1c536fcbf3"
},
"volume_type": None,
"description": "volume-manage-description"
}
}
VOLUME_MANAGE_RESPONSE = {
"volume": {
"migration_status": None,
"attachments": [],
"links": [
{
"href": "fake-url-1",
"rel": "self"
},
{
"href": "fake-url-2",
"rel": "bookmark"
}
],
"availability_zone": "nova",
"os-vol-host-attr:host": "controller1@rbd#rbd",
"encrypted": False,
"updated_at": None,
"replication_status": None,
"snapshot_id": None,
"id": "c07cd4a4-b52b-4511-a176-fbaa2011a227",
"size": 0,
"user_id": "142d8663efce464c89811c63e45bd82e",
"os-vol-tenant-attr:tenant_id": "f21a9c86d7114bf99c711f4874d80474",
"os-vol-mig-status-attr:migstat": None,
"metadata": {},
"status": "creating",
"description": "volume-manage-description",
"multiattach": False,
"source_volid": None,
"consistencygroup_id": None,
"os-vol-mig-status-attr:name_id": None,
"name": "volume-managed",
"bootable": "false",
"created_at": "2017-07-11T09:14:01.000000",
"volume_type": None
}
}
def setUp(self):
super(TestVolumeManageClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = volume_manage_client.VolumeManageClient(fake_auth,
'volume',
'regionOne')
def _test_manage_volume(self, bytes_body=False):
payload = json.dumps(self.VOLUME_MANAGE_REQUEST, sort_keys=True)
json_dumps = json.dumps
# NOTE: Use sort_keys for json.dumps so that the expected and actual
# payloads are guaranteed to be identical for mock_args assert check.
with mock.patch.object(volume_manage_client.json,
'dumps') as mock_dumps:
mock_dumps.side_effect = lambda d: json_dumps(d, sort_keys=True)
self.check_service_client_function(
self.client.manage_volume,
'tempest.lib.common.rest_client.RestClient.post',
self.VOLUME_MANAGE_RESPONSE,
to_utf=bytes_body,
status=202,
mock_args=['os-volume-manage', payload],
**self.VOLUME_MANAGE_REQUEST['volume'])
def test_manage_volume_with_str_body(self):
self._test_manage_volume()
def test_manage_volume_with_bytes_body(self):
self._test_manage_volume(bytes_body=True)