openstack/python-openstackclient
Code Issues Proposed changes

tests: Add volume v1, v2, v3 FakeClientMixin

Browse Source 
This ensures we are speccing the image proxy API, as we did previously
for the network tests in Ic203964c7dede7dd80ae2d93b8fa1b7e6634a758.

Change-Id: I132ccd1170cc903f6edc505926b071170aeaa08c
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2023-09-06 11:22:49 +01:00
parent 4cabf6f7a2
commit 98fb1678bc
9 changed files with 113 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
openstackclient/tests/unit/common/test_availability_zone.py
View File

@ -78,7 +78,11 @@ def _build_network_az_datalist(network_az, long_datalist=False):
return (datalist,)
class TestAvailabilityZone(network_fakes.FakeClientMixin, utils.TestCommand):
class TestAvailabilityZone(
network_fakes.FakeClientMixin,
volume_fakes.FakeClientMixin,
utils.TestCommand,
):
def setUp(self):
super().setUp()
@ -86,10 +90,6 @@ class TestAvailabilityZone(network_fakes.FakeClientMixin, utils.TestCommand):
self.compute_client = self.app.client_manager.sdk_connection.compute
self.compute_client.availability_zones = mock.Mock()
self.app.client_manager.sdk_connection.volume = mock.Mock()
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
self.volume_sdk_client.availability_zones = mock.Mock()
class TestAvailabilityZoneList(TestAvailabilityZone):
compute_azs = compute_fakes.create_availability_zones()

9
openstackclient/tests/unit/compute/v2/fakes.py
View File

@ -41,7 +41,7 @@ from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes
from openstackclient.tests.unit.image.v2 import fakes as image_fakes
from openstackclient.tests.unit.network.v2 import fakes as network_fakes
from openstackclient.tests.unit import utils
from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes
floating_ip_num = 100
fix_ip_num = 100
@ -151,6 +151,7 @@ class FakeComputev2Client(object):
class TestComputev2(
network_fakes.FakeClientMixin,
image_fakes.FakeClientMixin,
volume_fakes.FakeClientMixin,
utils.TestCommand,
):
def setUp(self):
@ -171,12 +172,6 @@ class TestComputev2(
token=fakes.AUTH_TOKEN,
)
self.app.client_manager.volume = volume_fakes.FakeVolumeClient(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,
)
self.volume_client = self.app.client_manager.volume
def create_one_aggregate(attrs=None):
"""Create a fake aggregate.

5
openstackclient/tests/unit/compute/v2/test_server.py
View File

@ -34,7 +34,7 @@ from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
from openstackclient.tests.unit.image.v2 import fakes as image_fakes
from openstackclient.tests.unit.network.v2 import fakes as network_fakes
from openstackclient.tests.unit import utils
from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes
class TestPowerStateColumn(utils.TestCase):
@ -95,9 +95,6 @@ class TestServer(compute_fakes.TestComputev2):
self.volumes_mock = self.volume_client.volumes
self.volumes_mock.reset_mock()
self.app.client_manager.sdk_connection.volume = mock.Mock()
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
# Get a shortcut to the volume client VolumeManager Mock
self.snapshots_mock = self.volume_client.volume_snapshots
self.snapshots_mock.reset_mock()

2
openstackclient/tests/unit/compute/v2/test_server_volume.py
View File

@ -27,9 +27,7 @@ class TestServerVolume(compute_fakes.TestComputev2):
super().setUp()
self.app.client_manager.sdk_connection.compute = mock.Mock()
self.app.client_manager.sdk_connection.volume = mock.Mock()
self.compute_client = self.app.client_manager.sdk_connection.compute
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
class TestServerVolumeList(TestServerVolume):

5
openstackclient/tests/unit/utils.py
View File

@ -12,9 +12,8 @@
# 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 io import StringIO
import io
import os
import fixtures
@ -75,7 +74,7 @@ class TestCommand(TestCase):
def check_parser(self, cmd, args, verify_args):
cmd_parser = cmd.get_parser('check_parser')
stderr = StringIO()
stderr = io.StringIO()
with fixtures.MonkeyPatch('sys.stderr', stderr):
try:
parsed_args = cmd_parser.parse_args(args)

7
openstackclient/tests/unit/volume/v1/fakes.py
View File

@ -53,7 +53,7 @@ class FakeVolumev1Client:
self.management_url = kwargs['endpoint']
class TestVolumev1(utils.TestCommand):
class FakeClientMixin:
def setUp(self):
super().setUp()
@ -63,6 +63,11 @@ class TestVolumev1(utils.TestCommand):
)
self.volume_client = self.app.client_manager.volume
class TestVolumev1(FakeClientMixin, utils.TestCommand):
def setUp(self):
super().setUp()
self.app.client_manager.identity = identity_fakes.FakeIdentityv2Client(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,

7
openstackclient/tests/unit/volume/v2/fakes.py
View File

@ -89,7 +89,7 @@ class FakeVolumeClient:
self.volumes.resource_class = fakes.FakeResource(None, {})
class TestVolume(utils.TestCommand):
class FakeClientMixin:
def setUp(self):
super().setUp()
@ -105,6 +105,11 @@ class TestVolume(utils.TestCommand):
)
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
class TestVolume(FakeClientMixin, utils.TestCommand):
def setUp(self):
super().setUp()
self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)

37
openstackclient/tests/unit/volume/v2/test_volume_backup.py
View File

@ -41,8 +41,6 @@ class TestBackup(volume_fakes.TestVolume):
def setUp(self):
super().setUp()
self.app.client_manager.sdk_connection.volume = mock.Mock()
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
patcher = mock.patch.object(
sdk_utils, 'supports_microversion', return_value=True
)
@ -325,24 +323,27 @@ class TestBackupDelete(TestBackup):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.backups[0], exceptions.CommandError]
with mock.patch.object(
self.volume_sdk_client, 'find_backup', side_effect=find_mock_result
) as find_mock:
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 backups failed to delete.', str(e))
self.volume_sdk_client.find_backup.side_effect = find_mock_result
find_mock.assert_any_call(self.backups[0].id, ignore_missing=False)
find_mock.assert_any_call('unexist_backup', ignore_missing=False)
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 backups failed to delete.', str(e))
self.assertEqual(2, find_mock.call_count)
self.volume_sdk_client.delete_backup.assert_called_once_with(
self.backups[0].id,
ignore_missing=False,
force=False,
)
self.volume_sdk_client.find_backup.assert_any_call(
self.backups[0].id, ignore_missing=False
)
self.volume_sdk_client.find_backup.assert_any_call(
'unexist_backup', ignore_missing=False
)
self.assertEqual(2, self.volume_sdk_client.find_backup.call_count)
self.volume_sdk_client.delete_backup.assert_called_once_with(
self.backups[0].id,
ignore_missing=False,
force=False,
)
class TestBackupList(TestBackup):

73
openstackclient/tests/unit/volume/v3/fakes.py
View File

@ -18,6 +18,7 @@ from cinderclient import api_versions
from openstack.block_storage.v3 import _proxy
from openstack.block_storage.v3 import availability_zone as _availability_zone
from openstack.block_storage.v3 import extension as _extension
from openstack.block_storage.v3 import volume as _volume
from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
from openstackclient.tests.unit import fakes
@ -32,6 +33,8 @@ class FakeVolumeClient:
self.management_url = kwargs['endpoint']
self.api_version = api_versions.APIVersion('3.0')
self.availability_zones = mock.Mock()
self.availability_zones.resource_class = fakes.FakeResource(None, {})
self.attachments = mock.Mock()
self.attachments.resource_class = fakes.FakeResource(None, {})
self.clusters = mock.Mock()
@ -44,10 +47,16 @@ class FakeVolumeClient:
self.group_types.resource_class = fakes.FakeResource(None, {})
self.messages = mock.Mock()
self.messages.resource_class = fakes.FakeResource(None, {})
self.quota_classes = mock.Mock()
self.quota_classes.resource_class = fakes.FakeResource(None, {})
self.quotas = mock.Mock()
self.quotas.resource_class = fakes.FakeResource(None, {})
self.resource_filters = mock.Mock()
self.resource_filters.resource_class = fakes.FakeResource(None, {})
self.volumes = mock.Mock()
self.volumes.resource_class = fakes.FakeResource(None, {})
self.volume_snapshots = mock.Mock()
self.volume_snapshots.resource_class = fakes.FakeResource(None, {})
self.volume_types = mock.Mock()
self.volume_types.resource_class = fakes.FakeResource(None, {})
self.services = mock.Mock()
@ -56,7 +65,7 @@ class FakeVolumeClient:
self.workers.resource_class = fakes.FakeResource(None, {})
class TestVolume(utils.TestCommand):
class FakeClientMixin:
def setUp(self):
super().setUp()
@ -72,6 +81,11 @@ class TestVolume(utils.TestCommand):
)
self.volume_sdk_client = self.app.client_manager.sdk_connection.volume
class TestVolume(FakeClientMixin, utils.TestCommand):
def setUp(self):
super().setUp()
self.app.client_manager.identity = identity_fakes.FakeIdentityv3Client(
endpoint=fakes.AUTH_URL, token=fakes.AUTH_TOKEN
)
@ -82,6 +96,7 @@ class TestVolume(utils.TestCommand):
# TODO(stephenfin): Check if the responses are actually the same
create_one_snapshot = volume_v2_fakes.create_one_snapshot
create_one_volume = volume_v2_fakes.create_one_volume
create_one_volume_type = volume_v2_fakes.create_one_volume_type
@ -243,6 +258,62 @@ def create_resource_filters(attrs=None, count=2):
return resource_filters
def create_one_sdk_volume(attrs=None):
"""Create a fake volume.
:param dict attrs:
A dictionary with all attributes of volume
:return:
A FakeResource object with id, name, status, etc.
"""
attrs = attrs or {}
# Set default attribute
volume_info = {
'id': 'volume-id' + uuid.uuid4().hex,
'name': 'volume-name' + uuid.uuid4().hex,
'description': 'description' + uuid.uuid4().hex,
'status': random.choice(['available', 'in_use']),
'size': random.randint(1, 20),
'volume_type': random.choice(['fake_lvmdriver-1', 'fake_lvmdriver-2']),
'bootable': random.choice(['true', 'false']),
'metadata': {
'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
'key' + uuid.uuid4().hex: 'val' + uuid.uuid4().hex,
},
'snapshot_id': random.randint(1, 5),
'availability_zone': 'zone' + uuid.uuid4().hex,
'attachments': [
{
'device': '/dev/' + uuid.uuid4().hex,
'server_id': uuid.uuid4().hex,
},
],
}
# Overwrite default attributes if there are some attributes set
volume_info.update(attrs)
return _volume.Volume(**volume_info)
def create_sdk_volumes(attrs=None, count=2):
"""Create multiple fake volumes.
:param dict attrs:
A dictionary with all attributes of volume
:param Integer count:
The number of volumes to be faked
:return:
A list of FakeResource objects
"""
volumes = []
for n in range(0, count):
volumes.append(create_one_sdk_volume(attrs))
return volumes
def create_one_volume_group(attrs=None):
"""Create a fake group.