Create Volume v3 functional tests
Until now-ish Volume v3 has been a pass-through to v2. In order to prepare to make the Volume v3 commands stand-alone copy the v2 functional tests to v3. This is the first of a series of reviews to completely separate Volume v2 and v3 commands. Once these are split we can begin to implement v3 microversion support and/or start using the OpenStack SDK as the REST library. Change-Id: Iefd78d8ef6bb851d7360596337a88ee8f8476767 Signed-off-by: Dean Troyer <dtroyer@gmail.com>
This commit is contained in:
Dean Troyer
parent
c6a171f010
commit
1c0160c8aa
|
|
@ -10,9 +10,209 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from openstackclient.tests.functional.volume.v2 import test_qos as v2
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.volume.v3 import common
|
||||
|
||||
|
||||
class QosTests(common.BaseVolumeTests, v2.QosTests):
|
||||
class QosTests(common.BaseVolumeTests):
|
||||
"""Functional tests for volume qos. """
|
||||
|
||||
def test_volume_qos_create_delete_list(self):
|
||||
"""Test create, list, delete multiple"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos create -f json ' +
|
||||
name1
|
||||
))
|
||||
self.assertEqual(
|
||||
name1,
|
||||
cmd_output['name']
|
||||
)
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos create -f json ' +
|
||||
name2
|
||||
))
|
||||
self.assertEqual(
|
||||
name2,
|
||||
cmd_output['name']
|
||||
)
|
||||
|
||||
# Test list
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos list -f json'
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test delete multiple
|
||||
del_output = self.openstack('volume qos delete ' + name1 + ' ' + name2)
|
||||
self.assertOutput('', del_output)
|
||||
|
||||
def test_volume_qos_set_show_unset(self):
|
||||
"""Tests create volume qos, set, unset, show, delete"""
|
||||
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos create -f json ' +
|
||||
'--consumer front-end '
|
||||
'--property Alpha=a ' +
|
||||
name
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume qos delete ' + name)
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output['name']
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
"front-end",
|
||||
cmd_output['consumer']
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Alpha': 'a'},
|
||||
cmd_output['properties']
|
||||
)
|
||||
|
||||
# Test volume qos set
|
||||
raw_output = self.openstack(
|
||||
'volume qos set ' +
|
||||
'--property Alpha=c ' +
|
||||
'--property Beta=b ' +
|
||||
name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# Test volume qos show
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output['name']
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Alpha': 'c', 'Beta': 'b'},
|
||||
cmd_output['properties']
|
||||
)
|
||||
|
||||
# Test volume qos unset
|
||||
raw_output = self.openstack(
|
||||
'volume qos unset ' +
|
||||
'--property Alpha ' +
|
||||
name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output['name']
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Beta': 'b'},
|
||||
cmd_output['properties']
|
||||
)
|
||||
|
||||
def test_volume_qos_asso_disasso(self):
|
||||
"""Tests associate and disassociate qos with volume type"""
|
||||
vol_type1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json ' +
|
||||
vol_type1
|
||||
))
|
||||
self.assertEqual(
|
||||
vol_type1,
|
||||
cmd_output['name']
|
||||
)
|
||||
self.addCleanup(self.openstack, 'volume type delete ' + vol_type1)
|
||||
|
||||
vol_type2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json ' +
|
||||
vol_type2
|
||||
))
|
||||
self.assertEqual(
|
||||
vol_type2,
|
||||
cmd_output['name']
|
||||
)
|
||||
self.addCleanup(self.openstack, 'volume type delete ' + vol_type2)
|
||||
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos create -f json ' +
|
||||
name
|
||||
))
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output['name']
|
||||
)
|
||||
self.addCleanup(self.openstack, 'volume qos delete ' + name)
|
||||
|
||||
# Test associate
|
||||
raw_output = self.openstack(
|
||||
'volume qos associate ' +
|
||||
name + ' ' + vol_type1
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
raw_output = self.openstack(
|
||||
'volume qos associate ' +
|
||||
name + ' ' + vol_type2
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
types = cmd_output["associations"]
|
||||
self.assertIn(vol_type1, types)
|
||||
self.assertIn(vol_type2, types)
|
||||
|
||||
# Test disassociate
|
||||
raw_output = self.openstack(
|
||||
'volume qos disassociate ' +
|
||||
'--volume-type ' + vol_type1 +
|
||||
' ' + name
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
types = cmd_output["associations"]
|
||||
self.assertNotIn(vol_type1, types)
|
||||
self.assertIn(vol_type2, types)
|
||||
|
||||
# Test disassociate --all
|
||||
raw_output = self.openstack(
|
||||
'volume qos associate ' +
|
||||
name + ' ' + vol_type1
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
types = cmd_output["associations"]
|
||||
self.assertIn(vol_type1, types)
|
||||
self.assertIn(vol_type2, types)
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume qos disassociate ' +
|
||||
'--all ' + name
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume qos show -f json ' +
|
||||
name
|
||||
))
|
||||
self.assertNotIn("associations", cmd_output.keys())
|
||||
|
|
|
|||
|
|
@ -10,12 +10,112 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from openstackclient.tests.functional.volume.v2 import test_transfer_request \
|
||||
as v2
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.volume.v3 import common
|
||||
|
||||
|
||||
class TransferRequestTests(common.BaseVolumeTests, v2.TransferRequestTests):
|
||||
class TransferRequestTests(common.BaseVolumeTests):
|
||||
"""Functional tests for transfer request. """
|
||||
|
||||
API_VERSION = '3'
|
||||
|
||||
def test_volume_transfer_request_accept(self):
|
||||
volume_name = uuid.uuid4().hex
|
||||
xfer_name = uuid.uuid4().hex
|
||||
|
||||
# create a volume
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
volume_name
|
||||
))
|
||||
self.assertEqual(volume_name, cmd_output['name'])
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume delete ' +
|
||||
volume_name
|
||||
)
|
||||
self.wait_for_status("volume", volume_name, "available")
|
||||
|
||||
# create volume transfer request for the volume
|
||||
# and get the auth_key of the new transfer request
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request create -f json ' +
|
||||
' --name ' + xfer_name + ' ' +
|
||||
volume_name
|
||||
))
|
||||
self.assertEqual(xfer_name, cmd_output['name'])
|
||||
xfer_id = cmd_output['id']
|
||||
auth_key = cmd_output['auth_key']
|
||||
self.assertTrue(auth_key)
|
||||
self.wait_for_status("volume", volume_name, "awaiting-transfer")
|
||||
|
||||
# accept the volume transfer request
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request accept -f json ' +
|
||||
'--auth-key ' + auth_key + ' ' +
|
||||
xfer_id
|
||||
))
|
||||
self.assertEqual(xfer_name, cmd_output['name'])
|
||||
self.wait_for_status("volume", volume_name, "available")
|
||||
|
||||
def test_volume_transfer_request_list_show(self):
|
||||
volume_name = uuid.uuid4().hex
|
||||
xfer_name = uuid.uuid4().hex
|
||||
|
||||
# create a volume
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
volume_name
|
||||
))
|
||||
self.assertEqual(volume_name, cmd_output['name'])
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume delete ' +
|
||||
volume_name
|
||||
)
|
||||
self.wait_for_status("volume", volume_name, "available")
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request create -f json ' +
|
||||
' --name ' + xfer_name + ' ' +
|
||||
volume_name
|
||||
))
|
||||
self.assertEqual(xfer_name, cmd_output['name'])
|
||||
xfer_id = cmd_output['id']
|
||||
auth_key = cmd_output['auth_key']
|
||||
self.assertTrue(auth_key)
|
||||
self.wait_for_status("volume", volume_name, "awaiting-transfer")
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request list -f json'
|
||||
))
|
||||
self.assertIn(xfer_name, [req['Name'] for req in cmd_output])
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request show -f json ' +
|
||||
xfer_id
|
||||
))
|
||||
self.assertEqual(xfer_name, cmd_output['name'])
|
||||
|
||||
# NOTE(dtroyer): We need to delete the transfer request to allow the
|
||||
# volume to be deleted. The addCleanup() route does
|
||||
# not have a mechanism to wait for the volume status
|
||||
# to become 'available' before attempting to delete
|
||||
# the volume.
|
||||
cmd_output = self.openstack(
|
||||
'--os-volume-api-version ' + self.API_VERSION + ' ' +
|
||||
'volume transfer request delete ' +
|
||||
xfer_id
|
||||
)
|
||||
self.wait_for_status("volume", volume_name, "available")
|
||||
|
|
|
|||
|
|
@ -10,9 +10,273 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from openstackclient.tests.functional.volume.v2 import test_volume as v2
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.volume.v3 import common
|
||||
|
||||
|
||||
class VolumeTests(common.BaseVolumeTests, v2.VolumeTests):
|
||||
class VolumeTests(common.BaseVolumeTests):
|
||||
"""Functional tests for volume. """
|
||||
|
||||
def test_volume_delete(self):
|
||||
"""Test create, delete multiple"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
name1
|
||||
))
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 2 ' +
|
||||
name2
|
||||
))
|
||||
self.assertEqual(
|
||||
2,
|
||||
cmd_output["size"],
|
||||
)
|
||||
|
||||
self.wait_for_status("volume", name1, "available")
|
||||
self.wait_for_status("volume", name2, "available")
|
||||
del_output = self.openstack('volume delete ' + name1 + ' ' + name2)
|
||||
self.assertOutput('', del_output)
|
||||
|
||||
def test_volume_list(self):
|
||||
"""Test create, list filter"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
name1
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume delete ' + name1)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.wait_for_status("volume", name1, "available")
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 2 ' +
|
||||
name2
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume delete ' + name2)
|
||||
self.assertEqual(
|
||||
2,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.wait_for_status("volume", name2, "available")
|
||||
raw_output = self.openstack(
|
||||
'volume set ' +
|
||||
'--state error ' +
|
||||
name2
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# Test list --long
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume list -f json ' +
|
||||
'--long'
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test list --status
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume list -f json ' +
|
||||
'--status error'
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertNotIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# TODO(qiangjiahui): Add project option to filter tests when we can
|
||||
# specify volume with project
|
||||
|
||||
def test_volume_set_and_unset(self):
|
||||
"""Tests create volume, set, unset, show, delete"""
|
||||
name = uuid.uuid4().hex
|
||||
new_name = name + "_"
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
'--description aaaa ' +
|
||||
'--property Alpha=a ' +
|
||||
name
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume delete ' + new_name)
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'aaaa',
|
||||
cmd_output["description"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Alpha': 'a'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'false',
|
||||
cmd_output["bootable"],
|
||||
)
|
||||
self.wait_for_status("volume", name, "available")
|
||||
|
||||
# Test volume set
|
||||
raw_output = self.openstack(
|
||||
'volume set ' +
|
||||
'--name ' + new_name +
|
||||
' --size 2 ' +
|
||||
'--description bbbb ' +
|
||||
'--no-property ' +
|
||||
'--property Beta=b ' +
|
||||
'--property Gamma=c ' +
|
||||
'--image-property a=b ' +
|
||||
'--image-property c=d ' +
|
||||
'--bootable ' +
|
||||
name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume show -f json ' +
|
||||
new_name
|
||||
))
|
||||
self.assertEqual(
|
||||
new_name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
2,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'bbbb',
|
||||
cmd_output["description"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Beta': 'b', 'Gamma': 'c'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'a': 'b', 'c': 'd'},
|
||||
cmd_output["volume_image_metadata"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'true',
|
||||
cmd_output["bootable"],
|
||||
)
|
||||
|
||||
# Test volume unset
|
||||
raw_output = self.openstack(
|
||||
'volume unset ' +
|
||||
'--property Beta ' +
|
||||
'--image-property a ' +
|
||||
new_name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume show -f json ' +
|
||||
new_name
|
||||
))
|
||||
self.assertEqual(
|
||||
{'Gamma': 'c'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'c': 'd'},
|
||||
cmd_output["volume_image_metadata"],
|
||||
)
|
||||
|
||||
def test_volume_snapshot(self):
|
||||
"""Tests volume create from snapshot"""
|
||||
|
||||
volume_name = uuid.uuid4().hex
|
||||
snapshot_name = uuid.uuid4().hex
|
||||
# Make a snapshot
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
volume_name
|
||||
))
|
||||
self.wait_for_status("volume", volume_name, "available")
|
||||
self.assertEqual(
|
||||
volume_name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
snapshot_name +
|
||||
' --volume ' + volume_name
|
||||
))
|
||||
self.wait_for_status("volume snapshot", snapshot_name, "available")
|
||||
|
||||
name = uuid.uuid4().hex
|
||||
# Create volume from snapshot
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--snapshot ' + snapshot_name +
|
||||
' ' + name
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume delete ' + name)
|
||||
self.addCleanup(self.openstack, 'volume delete ' + volume_name)
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.wait_for_status("volume", name, "available")
|
||||
|
||||
# Delete snapshot
|
||||
raw_output = self.openstack(
|
||||
'volume snapshot delete ' + snapshot_name)
|
||||
self.assertOutput('', raw_output)
|
||||
# Deleting snapshot may take time. If volume snapshot still exists when
|
||||
# a parent volume delete is requested, the volume deletion will fail.
|
||||
self.wait_for_delete('volume snapshot', snapshot_name)
|
||||
|
||||
def test_volume_list_backward_compatibility(self):
|
||||
"""Test backward compatibility of list command"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
name1
|
||||
))
|
||||
self.addCleanup(self.openstack, 'volume delete ' + name1)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.wait_for_status("volume", name1, "available")
|
||||
|
||||
# Test list -c "Display Name"
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume list -f json ' +
|
||||
'-c "Display Name"'
|
||||
))
|
||||
for each_volume in cmd_output:
|
||||
self.assertIn('Display Name', each_volume)
|
||||
|
||||
# Test list -c "Name"
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume list -f json ' +
|
||||
'-c "Name"'
|
||||
))
|
||||
for each_volume in cmd_output:
|
||||
self.assertIn('Name', each_volume)
|
||||
|
|
|
|||
|
|
@ -10,9 +10,245 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from openstackclient.tests.functional.volume.v2 import test_volume_snapshot as v2 # noqa
|
||||
import json
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.volume.v3 import common
|
||||
|
||||
|
||||
class VolumeSnapshotTests(common.BaseVolumeTests, v2.VolumeSnapshotTests):
|
||||
class VolumeSnapshotTests(common.BaseVolumeTests):
|
||||
"""Functional tests for volume snapshot. """
|
||||
|
||||
VOLLY = uuid.uuid4().hex
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super(VolumeSnapshotTests, cls).setUpClass()
|
||||
# create a volume for all tests to create snapshot
|
||||
cmd_output = json.loads(cls.openstack(
|
||||
'volume create -f json ' +
|
||||
'--size 1 ' +
|
||||
cls.VOLLY
|
||||
))
|
||||
cls.wait_for_status('volume', cls.VOLLY, 'available')
|
||||
cls.VOLUME_ID = cmd_output['id']
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
try:
|
||||
cls.wait_for_status('volume', cls.VOLLY, 'available')
|
||||
raw_output = cls.openstack(
|
||||
'volume delete --force ' + cls.VOLLY)
|
||||
cls.assertOutput('', raw_output)
|
||||
finally:
|
||||
super(VolumeSnapshotTests, cls).tearDownClass()
|
||||
|
||||
def test_volume_snapshot_delete(self):
|
||||
"""Test create, delete multiple"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
name1 +
|
||||
' --volume ' + self.VOLLY
|
||||
))
|
||||
self.assertEqual(
|
||||
name1,
|
||||
cmd_output["name"],
|
||||
)
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
name2 +
|
||||
' --volume ' + self.VOLLY
|
||||
))
|
||||
self.assertEqual(
|
||||
name2,
|
||||
cmd_output["name"],
|
||||
)
|
||||
|
||||
self.wait_for_status('volume snapshot', name1, 'available')
|
||||
self.wait_for_status('volume snapshot', name2, 'available')
|
||||
|
||||
del_output = self.openstack(
|
||||
'volume snapshot delete ' + name1 + ' ' + name2)
|
||||
self.assertOutput('', del_output)
|
||||
self.wait_for_delete('volume snapshot', name1)
|
||||
self.wait_for_delete('volume snapshot', name2)
|
||||
|
||||
def test_volume_snapshot_list(self):
|
||||
"""Test create, list filter"""
|
||||
name1 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
name1 +
|
||||
' --volume ' + self.VOLLY
|
||||
))
|
||||
self.addCleanup(self.wait_for_delete, 'volume snapshot', name1)
|
||||
self.addCleanup(self.openstack, 'volume snapshot delete ' + name1)
|
||||
self.assertEqual(
|
||||
name1,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
self.VOLUME_ID,
|
||||
cmd_output["volume_id"],
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.wait_for_status('volume snapshot', name1, 'available')
|
||||
|
||||
name2 = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
name2 +
|
||||
' --volume ' + self.VOLLY
|
||||
))
|
||||
self.addCleanup(self.wait_for_delete, 'volume snapshot', name2)
|
||||
self.addCleanup(self.openstack, 'volume snapshot delete ' + name2)
|
||||
self.assertEqual(
|
||||
name2,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
self.VOLUME_ID,
|
||||
cmd_output["volume_id"],
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.wait_for_status('volume snapshot', name2, 'available')
|
||||
raw_output = self.openstack(
|
||||
'volume snapshot set ' +
|
||||
'--state error ' +
|
||||
name2
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# Test list --long, --status
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot list -f json ' +
|
||||
'--long ' +
|
||||
'--status error'
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertNotIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test list --volume
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot list -f json ' +
|
||||
'--volume ' + self.VOLLY
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertIn(name2, names)
|
||||
|
||||
# Test list --name
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot list -f json ' +
|
||||
'--name ' + name1
|
||||
))
|
||||
names = [x["Name"] for x in cmd_output]
|
||||
self.assertIn(name1, names)
|
||||
self.assertNotIn(name2, names)
|
||||
|
||||
def test_volume_snapshot_set(self):
|
||||
"""Test create, set, unset, show, delete volume snapshot"""
|
||||
name = uuid.uuid4().hex
|
||||
new_name = name + "_"
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot create -f json ' +
|
||||
'--volume ' + self.VOLLY +
|
||||
' --description aaaa ' +
|
||||
'--property Alpha=a ' +
|
||||
name
|
||||
))
|
||||
self.addCleanup(self.wait_for_delete, 'volume snapshot', new_name)
|
||||
self.addCleanup(self.openstack, 'volume snapshot delete ' + new_name)
|
||||
self.assertEqual(
|
||||
name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'aaaa',
|
||||
cmd_output["description"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Alpha': 'a'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
self.wait_for_status('volume snapshot', name, 'available')
|
||||
|
||||
# Test volume snapshot set
|
||||
raw_output = self.openstack(
|
||||
'volume snapshot set ' +
|
||||
'--name ' + new_name +
|
||||
' --description bbbb ' +
|
||||
'--property Alpha=c ' +
|
||||
'--property Beta=b ' +
|
||||
name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# Show snapshot set result
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot show -f json ' +
|
||||
new_name
|
||||
))
|
||||
self.assertEqual(
|
||||
new_name,
|
||||
cmd_output["name"],
|
||||
)
|
||||
self.assertEqual(
|
||||
1,
|
||||
cmd_output["size"],
|
||||
)
|
||||
self.assertEqual(
|
||||
'bbbb',
|
||||
cmd_output["description"],
|
||||
)
|
||||
self.assertEqual(
|
||||
{'Alpha': 'c', 'Beta': 'b'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
|
||||
# Test volume snapshot unset
|
||||
raw_output = self.openstack(
|
||||
'volume snapshot unset ' +
|
||||
'--property Alpha ' +
|
||||
new_name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot show -f json ' +
|
||||
new_name
|
||||
))
|
||||
self.assertEqual(
|
||||
{'Beta': 'b'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
|
||||
# Test volume snapshot set --no-property
|
||||
raw_output = self.openstack(
|
||||
'volume snapshot set ' +
|
||||
'--no-property ' +
|
||||
new_name,
|
||||
)
|
||||
self.assertOutput('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume snapshot show -f json ' +
|
||||
new_name
|
||||
))
|
||||
self.assertNotIn(
|
||||
{'Beta': 'b'},
|
||||
cmd_output["properties"],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -10,9 +10,229 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from openstackclient.tests.functional.volume.v2 import test_volume_type as v2
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from openstackclient.tests.functional.volume.v3 import common
|
||||
|
||||
|
||||
class VolumeTypeTests(common.BaseVolumeTests, v2.VolumeTypeTests):
|
||||
class VolumeTypeTests(common.BaseVolumeTests):
|
||||
"""Functional tests for volume type. """
|
||||
|
||||
def test_volume_type_create_list(self):
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json --private ' +
|
||||
name,
|
||||
))
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'volume type delete ' + name,
|
||||
)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json %s' % name
|
||||
))
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
cmd_output = json.loads(self.openstack('volume type list -f json'))
|
||||
self.assertIn(name, [t['Name'] for t in cmd_output])
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type list -f json --default'
|
||||
))
|
||||
self.assertEqual(1, len(cmd_output))
|
||||
self.assertEqual('lvmdriver-1', cmd_output[0]['Name'])
|
||||
|
||||
def test_volume_type_set_unset_properties(self):
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json --private ' +
|
||||
name,
|
||||
))
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'volume type delete ' + name
|
||||
)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type set --property a=b --property c=d %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json %s' % name
|
||||
))
|
||||
self.assertEqual({'a': 'b', 'c': 'd'}, cmd_output['properties'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type unset --property a %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json %s' % name
|
||||
))
|
||||
self.assertEqual({'c': 'd'}, cmd_output['properties'])
|
||||
|
||||
def test_volume_type_set_unset_multiple_properties(self):
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json --private ' +
|
||||
name,
|
||||
))
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'volume type delete ' + name
|
||||
)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type set --property a=b --property c=d %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json %s' % name
|
||||
))
|
||||
self.assertEqual({'a': 'b', 'c': 'd'}, cmd_output['properties'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type unset --property a --property c %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json %s' % name
|
||||
))
|
||||
self.assertEqual({}, cmd_output['properties'])
|
||||
|
||||
def test_volume_type_set_unset_project(self):
|
||||
name = uuid.uuid4().hex
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json --private ' +
|
||||
name,
|
||||
))
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'volume type delete ' + name
|
||||
)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type set --project admin %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type unset --project admin %s' % name
|
||||
)
|
||||
self.assertEqual("", raw_output)
|
||||
|
||||
def test_multi_delete(self):
|
||||
vol_type1 = uuid.uuid4().hex
|
||||
vol_type2 = uuid.uuid4().hex
|
||||
self.openstack('volume type create %s' % vol_type1)
|
||||
time.sleep(5)
|
||||
self.openstack('volume type create %s' % vol_type2)
|
||||
time.sleep(5)
|
||||
cmd = 'volume type delete %s %s' % (vol_type1, vol_type2)
|
||||
raw_output = self.openstack(cmd)
|
||||
self.assertOutput('', raw_output)
|
||||
|
||||
# NOTE: Add some basic funtional tests with the old format to
|
||||
# make sure the command works properly, need to change
|
||||
# these to new test format when beef up all tests for
|
||||
# volume tye commands.
|
||||
def test_encryption_type(self):
|
||||
name = uuid.uuid4().hex
|
||||
encryption_type = uuid.uuid4().hex
|
||||
# test create new encryption type
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json '
|
||||
'--encryption-provider LuksEncryptor '
|
||||
'--encryption-cipher aes-xts-plain64 '
|
||||
'--encryption-key-size 128 '
|
||||
'--encryption-control-location front-end ' +
|
||||
encryption_type))
|
||||
expected = {'provider': 'LuksEncryptor',
|
||||
'cipher': 'aes-xts-plain64',
|
||||
'key_size': 128,
|
||||
'control_location': 'front-end'}
|
||||
for attr, value in expected.items():
|
||||
self.assertEqual(value, cmd_output['encryption'][attr])
|
||||
# test show encryption type
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json --encryption-type ' + encryption_type))
|
||||
expected = {'provider': 'LuksEncryptor',
|
||||
'cipher': 'aes-xts-plain64',
|
||||
'key_size': 128,
|
||||
'control_location': 'front-end'}
|
||||
for attr, value in expected.items():
|
||||
self.assertEqual(value, cmd_output['encryption'][attr])
|
||||
# test list encryption type
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type list -f json --encryption-type'))
|
||||
encryption_output = [t['Encryption'] for t in cmd_output
|
||||
if t['Name'] == encryption_type][0]
|
||||
expected = {'provider': 'LuksEncryptor',
|
||||
'cipher': 'aes-xts-plain64',
|
||||
'key_size': 128,
|
||||
'control_location': 'front-end'}
|
||||
for attr, value in expected.items():
|
||||
self.assertEqual(value, encryption_output[attr])
|
||||
# test set existing encryption type
|
||||
raw_output = self.openstack(
|
||||
'volume type set '
|
||||
'--encryption-key-size 256 '
|
||||
'--encryption-control-location back-end ' +
|
||||
encryption_type)
|
||||
self.assertEqual('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json --encryption-type ' + encryption_type))
|
||||
expected = {'provider': 'LuksEncryptor',
|
||||
'cipher': 'aes-xts-plain64',
|
||||
'key_size': 256,
|
||||
'control_location': 'back-end'}
|
||||
for attr, value in expected.items():
|
||||
self.assertEqual(value, cmd_output['encryption'][attr])
|
||||
# test set new encryption type
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type create -f json --private ' +
|
||||
name,
|
||||
))
|
||||
self.addCleanup(
|
||||
self.openstack,
|
||||
'volume type delete ' + name,
|
||||
)
|
||||
self.assertEqual(name, cmd_output['name'])
|
||||
|
||||
raw_output = self.openstack(
|
||||
'volume type set '
|
||||
'--encryption-provider LuksEncryptor '
|
||||
'--encryption-cipher aes-xts-plain64 '
|
||||
'--encryption-key-size 128 '
|
||||
'--encryption-control-location front-end ' +
|
||||
name)
|
||||
self.assertEqual('', raw_output)
|
||||
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json --encryption-type ' + name
|
||||
))
|
||||
expected = {'provider': 'LuksEncryptor',
|
||||
'cipher': 'aes-xts-plain64',
|
||||
'key_size': 128,
|
||||
'control_location': 'front-end'}
|
||||
for attr, value in expected.items():
|
||||
self.assertEqual(value, cmd_output['encryption'][attr])
|
||||
# test unset encryption type
|
||||
raw_output = self.openstack(
|
||||
'volume type unset --encryption-type ' + name
|
||||
)
|
||||
self.assertEqual('', raw_output)
|
||||
cmd_output = json.loads(self.openstack(
|
||||
'volume type show -f json --encryption-type ' + name
|
||||
))
|
||||
self.assertEqual({}, cmd_output['encryption'])
|
||||
# test delete encryption type
|
||||
raw_output = self.openstack('volume type delete ' + encryption_type)
|
||||
self.assertEqual('', raw_output)
|
||||
|
|
|
|||
Loading…
Reference in New Issue