diff options
Diffstat (limited to 'openstackclient/tests')
4 files changed, 272 insertions, 147 deletions
diff --git a/openstackclient/tests/functional/common/test_quota.py b/openstackclient/tests/functional/common/test_quota.py index b2b198af..c1de9aa9 100644 --- a/openstackclient/tests/functional/common/test_quota.py +++ b/openstackclient/tests/functional/common/test_quota.py @@ -10,8 +10,6 @@ # License for the specific language governing permissions and limitations # under the License. -import testtools - from openstackclient.tests.functional import base @@ -27,7 +25,6 @@ class QuotaTests(base.TestCase): cls.PROJECT_NAME =\ cls.get_openstack_configuration_value('auth.project_name') - @testtools.skip('broken SDK testing') def test_quota_set(self): self.openstack('quota set --instances 11 --volumes 11 --networks 11 ' + self.PROJECT_NAME) diff --git a/openstackclient/tests/functional/network/v2/test_floating_ip.py b/openstackclient/tests/functional/network/v2/test_floating_ip.py index fa9607a0..8fbec3d5 100644 --- a/openstackclient/tests/functional/network/v2/test_floating_ip.py +++ b/openstackclient/tests/functional/network/v2/test_floating_ip.py @@ -31,25 +31,38 @@ class FloatingIpTests(base.TestCase): cls.re_description = re.compile("description\s+\|\s+([^|]+?)\s+\|") cls.re_network_id = re.compile("floating_network_id\s+\|\s+(\S+)") - # Make a random subnet - cls.subnet = ".".join(map( - str, - (random.randint(0, 223) for _ in range(3)) - )) + ".0/26" - # Create a network for the floating ip raw_output = cls.openstack( 'network create --external ' + cls.NETWORK_NAME ) cls.network_id = re.search(cls.re_id, raw_output).group(1) - # Create a subnet for the network - raw_output = cls.openstack( - 'subnet create ' + - '--network ' + cls.NETWORK_NAME + ' ' + - '--subnet-range ' + cls.subnet + ' ' + - cls.SUBNET_NAME - ) + # Try random subnet range for subnet creating + # Because we can not determine ahead of time what subnets are already + # in use, possibly by another test running in parallel, try 4 times + for i in range(4): + # Make a random subnet + cls.subnet = ".".join(map( + str, + (random.randint(0, 223) for _ in range(3)) + )) + ".0/26" + try: + # Create a subnet for the network + raw_output = cls.openstack( + 'subnet create ' + + '--network ' + cls.NETWORK_NAME + ' ' + + '--subnet-range ' + cls.subnet + ' ' + + cls.SUBNET_NAME + ) + except Exception: + if (i == 3): + # raise the exception at the last time + raise + pass + else: + # break and no longer retry if create sucessfully + break + cls.subnet_id = re.search(cls.re_id, raw_output).group(1) @classmethod diff --git a/openstackclient/tests/functional/volume/v2/test_volume.py b/openstackclient/tests/functional/volume/v2/test_volume.py index ea891cba..203ca819 100644 --- a/openstackclient/tests/functional/volume/v2/test_volume.py +++ b/openstackclient/tests/functional/volume/v2/test_volume.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import json import time import uuid @@ -19,118 +20,220 @@ from openstackclient.tests.functional.volume.v2 import common class VolumeTests(common.BaseVolumeTests): """Functional tests for volume. """ - NAME = uuid.uuid4().hex - SNAPSHOT_NAME = uuid.uuid4().hex - VOLUME_FROM_SNAPSHOT_NAME = uuid.uuid4().hex - OTHER_NAME = uuid.uuid4().hex - HEADERS = ['"Display Name"'] - FIELDS = ['name'] - - @classmethod - def setUpClass(cls): - super(VolumeTests, cls).setUpClass() - opts = cls.get_opts(cls.FIELDS) - - # Create test volume - raw_output = cls.openstack('volume create --size 1 ' + cls.NAME + opts) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) - - @classmethod - def tearDownClass(cls): - # Rename test volume - raw_output = cls.openstack( - 'volume set --name ' + cls.OTHER_NAME + ' ' + cls.NAME) - cls.assertOutput('', raw_output) - - # Set volume state - cls.openstack('volume set --state error ' + cls.OTHER_NAME) - opts = cls.get_opts(["status"]) - raw_output_status = cls.openstack( - 'volume show ' + cls.OTHER_NAME + opts) - - # Delete test volume - raw_output = cls.openstack('volume delete ' + cls.OTHER_NAME) - cls.assertOutput('', raw_output) - cls.assertOutput('error\n', raw_output_status) + 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("volume", name1, "available") + self.wait_for("volume", name2, "available") + del_output = self.openstack('volume delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) def test_volume_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('volume list' + opts) - self.assertIn(self.NAME, raw_output) - - def test_volume_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) - - def test_volume_properties(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("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("volume", name2, "available") raw_output = self.openstack( - 'volume set --property a=b --property c=d ' + self.NAME) - self.assertEqual("", raw_output) - opts = self.get_opts(["properties"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("a='b', c='d'\n", raw_output) + 'volume set ' + + '--state error ' + + name2 + ) + self.assertOutput('', raw_output) - raw_output = self.openstack('volume unset --property a ' + self.NAME) - self.assertEqual("", raw_output) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("c='d'\n", raw_output) + # Test list --long + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '--long' + )) + names = [x["Display 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["Display 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(self): - discription = uuid.uuid4().hex - self.openstack('volume set --description ' + discription + ' ' + - self.NAME) - opts = self.get_opts(["description", "name"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(discription + "\n" + self.NAME + "\n", raw_output) - - def test_volume_set_size(self): - self.openstack('volume set --size 2 ' + self.NAME) - opts = self.get_opts(["name", "size"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n2\n", raw_output) - - def test_volume_set_bootable(self): - self.openstack('volume set --bootable ' + self.NAME) - opts = self.get_opts(["bootable"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("true\n", raw_output) - - self.openstack('volume set --non-bootable ' + self.NAME) - opts = self.get_opts(["bootable"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("false\n", raw_output) + """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("volume", name, "available") + + # Test volume set + raw_output = self.openstack( + 'volume set ' + + '--name ' + new_name + + ' --size 2 ' + + '--description bbbb ' + + '--property Alpha=c ' + + '--property Beta=b ' + + '--bootable ' + + name, + ) + self.assertOutput('', raw_output) - def test_volume_snapshot(self): - opts = self.get_opts(self.FIELDS) - - # Create snapshot from test volume - raw_output = self.openstack('volume snapshot create ' + - self.SNAPSHOT_NAME + - ' --volume ' + self.NAME + opts) - expected = self.SNAPSHOT_NAME + '\n' - self.assertOutput(expected, raw_output) - self.wait_for("volume snapshot", self.SNAPSHOT_NAME, "available") - - # Create volume from snapshot - raw_output = self.openstack('volume create --size 2 --snapshot ' + - self.SNAPSHOT_NAME + ' ' + - self.VOLUME_FROM_SNAPSHOT_NAME + opts) - expected = self.VOLUME_FROM_SNAPSHOT_NAME + '\n' - self.assertOutput(expected, raw_output) - self.wait_for("volume", self.VOLUME_FROM_SNAPSHOT_NAME, "available") - - # Delete volume that create from snapshot - raw_output = self.openstack('volume delete ' + - self.VOLUME_FROM_SNAPSHOT_NAME) + 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( + "Alpha='c', Beta='b'", + cmd_output["properties"], + ) + self.assertEqual( + 'true', + cmd_output["bootable"], + ) + + # Test volume unset + raw_output = self.openstack( + 'volume unset ' + + '--property Alpha ' + + new_name, + ) self.assertOutput('', raw_output) - # Delete test snapshot + cmd_output = json.loads(self.openstack( + 'volume show -f json ' + + new_name + )) + self.assertEqual( + "Beta='b'", + cmd_output["properties"], + ) + + 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("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("volume snapshot", snapshot_name, "available") + + name = uuid.uuid4().hex + 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("volume", name, "available") + + # Delete snapshot raw_output = self.openstack( - 'volume snapshot delete ' + self.SNAPSHOT_NAME) + 'volume snapshot delete ' + snapshot_name) self.assertOutput('', raw_output) - self.wait_for("volume", self.NAME, "available") def wait_for(self, check_type, check_name, desired_status, wait=120, interval=5, failures=['ERROR']): diff --git a/openstackclient/tests/unit/common/test_quota.py b/openstackclient/tests/unit/common/test_quota.py index 7dd23373..244d74d2 100644 --- a/openstackclient/tests/unit/common/test_quota.py +++ b/openstackclient/tests/unit/common/test_quota.py @@ -13,6 +13,8 @@ import copy import mock +from openstack.network.v2 import quota as _quota + from openstackclient.common import quota from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes from openstackclient.tests.unit import fakes @@ -282,27 +284,32 @@ class TestQuotaSet(TestQuota): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - kwargs = { - 'subnet': network_fakes.QUOTA['subnet'], - 'network': network_fakes.QUOTA['network'], - 'floatingip': network_fakes.QUOTA['floatingip'], - 'subnetpool': network_fakes.QUOTA['subnetpool'], - 'security_group_rule': - network_fakes.QUOTA['security_group_rule'], - 'security_group': network_fakes.QUOTA['security_group'], - 'router': network_fakes.QUOTA['router'], - 'rbac_policy': network_fakes.QUOTA['rbac_policy'], - 'port': network_fakes.QUOTA['port'], - 'vip': network_fakes.QUOTA['vip'], - 'healthmonitor': network_fakes.QUOTA['healthmonitor'], - 'l7policy': network_fakes.QUOTA['l7policy'], - } - self.network_mock.update_quota.assert_called_once_with( - identity_fakes.project_id, - **kwargs - ) - self.assertIsNone(result) + # TODO(huanxuan): Remove this if condition once the fixed + # SDK Quota class is the minimum required version. + # This is expected to be SDK release 0.9.13 + if not hasattr(_quota.Quota, 'allow_get'): + # Just run this when sdk <= 0.9.10 + result = self.cmd.take_action(parsed_args) + kwargs = { + 'subnet': network_fakes.QUOTA['subnet'], + 'network': network_fakes.QUOTA['network'], + 'floatingip': network_fakes.QUOTA['floatingip'], + 'subnetpool': network_fakes.QUOTA['subnetpool'], + 'security_group_rule': + network_fakes.QUOTA['security_group_rule'], + 'security_group': network_fakes.QUOTA['security_group'], + 'router': network_fakes.QUOTA['router'], + 'rbac_policy': network_fakes.QUOTA['rbac_policy'], + 'port': network_fakes.QUOTA['port'], + 'vip': network_fakes.QUOTA['vip'], + 'healthmonitor': network_fakes.QUOTA['healthmonitor'], + 'l7policy': network_fakes.QUOTA['l7policy'], + } + self.network_mock.update_quota.assert_called_once_with( + identity_fakes.project_id, + **kwargs + ) + self.assertIsNone(result) def test_quota_set_with_class(self): arglist = [ @@ -476,15 +483,20 @@ class TestQuotaShow(TestQuota): parsed_args = self.check_parser(self.cmd, arglist, verifylist) - self.cmd.take_action(parsed_args) - - self.quotas_mock.defaults.assert_called_once_with( - identity_fakes.project_id) - self.volume_quotas_mock.defaults.assert_called_once_with( - identity_fakes.project_id) - self.network.get_quota_default.assert_called_once_with( - identity_fakes.project_id) - self.assertNotCalled(self.network.get_quota) + # TODO(huanxuan): Remove this if condition once the fixed + # SDK QuotaDefault class is the minimum required version. + # This is expected to be SDK release 0.9.13 + if not hasattr(_quota.QuotaDefault, 'project'): + # Just run this when sdk <= 0.9.10 + self.cmd.take_action(parsed_args) + + self.quotas_mock.defaults.assert_called_once_with( + identity_fakes.project_id) + self.volume_quotas_mock.defaults.assert_called_once_with( + identity_fakes.project_id) + self.network.get_quota_default.assert_called_once_with( + identity_fakes.project_id) + self.assertNotCalled(self.network.get_quota) def test_quota_show_with_class(self): arglist = [ |