Merge "Move common wait_for_image_status from compute images_client to waiters"
diff --git a/.testr.conf b/.testr.conf
index 05b12c4..c25ebec 100644
--- a/.testr.conf
+++ b/.testr.conf
@@ -2,7 +2,8 @@
test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-500} \
- ${PYTHON:-python} -m subunit.run discover -t ./ ./tempest $LISTOPT $IDOPTION
+ OS_TEST_PATH=${OS_TEST_PATH:-./tempest} \
+ ${PYTHON:-python} -m subunit.run discover -t ./ $OS_TEST_PATH $LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE
test_list_option=--list
group_regex=([^\.]*\.)*
diff --git a/HACKING.rst b/HACKING.rst
index a74ff73..3fa1ff5 100644
--- a/HACKING.rst
+++ b/HACKING.rst
@@ -9,8 +9,8 @@
------------------------------
- [T102] Cannot import OpenStack python clients in tempest/api tests
-- [T103] tempest/tests is deprecated
- [T104] Scenario tests require a services decorator
+- [T105] Unit tests cannot use setUpClass
Test Data/Configuration
-----------------------
@@ -39,7 +39,7 @@
when the additional operations leads to another exception.
Just letting an exception to propagate, is not bad idea in a test case,
- at all.
+at all.
Try to avoid using any exception handling construct which can hide the errors
origin.
@@ -53,10 +53,10 @@
test fails part way through.
Use the ``self.assert*`` methods provided by the unit test framework
- the signal failures early.
+the signal failures early.
Avoid using the ``self.fail`` alone, it's stack trace will signal
- the ``self.fail`` line as the origin of the error.
+the ``self.fail`` line as the origin of the error.
Avoid constructing complex boolean expressions for assertion.
The ``self.assertTrue`` or ``self.assertFalse`` without a ``msg`` argument,
@@ -192,3 +192,15 @@
The sample config file is autogenerated using a script. If any changes are made
to the config variables in tempest then the sample config file must be
regenerated. This can be done running the script: tools/generate_sample.sh
+
+Unit Tests
+----------
+Unit tests are a separate class of tests in tempest. They verify tempest
+itself, and thus have a different set of guidelines around them:
+
+1. They can not require anything running externally. All you should need to
+ run the unit tests is the git tree, python and the dependencies installed.
+ This includes running services, a config file, etc.
+
+2. The unit tests cannot use setUpClass, instead fixtures and testresources
+ should be used for shared state between tests.
diff --git a/tempest/api/compute/admin/test_aggregates.py b/tempest/api/compute/admin/test_aggregates.py
index 467a6f9..609d2c6 100644
--- a/tempest/api/compute/admin/test_aggregates.py
+++ b/tempest/api/compute/admin/test_aggregates.py
@@ -18,7 +18,6 @@
from tempest.api.compute import base
from tempest.common import tempest_fixtures as fixtures
from tempest.common.utils import data_utils
-from tempest import exceptions
from tempest.test import attr
@@ -35,7 +34,6 @@
def setUpClass(cls):
super(AggregatesAdminTestJSON, cls).setUpClass()
cls.client = cls.os_adm.aggregates_client
- cls.user_client = cls.aggregates_client
cls.aggregate_name_prefix = 'test_aggregate_'
cls.az_name_prefix = 'test_az_'
@@ -141,54 +139,6 @@
(x['id'], x['name'], x['availability_zone']),
aggregates))
- @attr(type=['negative', 'gate'])
- def test_aggregate_create_as_user(self):
- # Regular user is not allowed to create an aggregate.
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.create_aggregate,
- aggregate_name)
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_delete_as_user(self):
- # Regular user is not allowed to delete an aggregate.
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- resp, aggregate = self.client.create_aggregate(aggregate_name)
- self.addCleanup(self.client.delete_aggregate, aggregate['id'])
-
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.delete_aggregate,
- aggregate['id'])
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_list_as_user(self):
- # Regular user is not allowed to list aggregates.
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.list_aggregates)
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_get_details_as_user(self):
- # Regular user is not allowed to get aggregate details.
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- resp, aggregate = self.client.create_aggregate(aggregate_name)
- self.addCleanup(self.client.delete_aggregate, aggregate['id'])
-
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.get_aggregate,
- aggregate['id'])
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_delete_with_invalid_id(self):
- # Delete an aggregate with invalid id should raise exceptions.
- self.assertRaises(exceptions.NotFound,
- self.client.delete_aggregate, -1)
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_get_details_with_invalid_id(self):
- # Get aggregate details with invalid id should raise exceptions.
- self.assertRaises(exceptions.NotFound,
- self.client.get_aggregate, -1)
-
@attr(type='gate')
def test_aggregate_add_remove_host(self):
# Add an host to the given aggregate and remove.
@@ -262,48 +212,6 @@
resp, body = admin_servers_client.get_server(server['id'])
self.assertEqual(self.host, body[self._host_key])
- @attr(type=['negative', 'gate'])
- def test_aggregate_add_non_exist_host(self):
- # Adding a non-exist host to an aggregate should raise exceptions.
- resp, hosts_all = self.os_adm.hosts_client.list_hosts()
- hosts = map(lambda x: x['host_name'], hosts_all)
- while True:
- non_exist_host = data_utils.rand_name('nonexist_host_')
- if non_exist_host not in hosts:
- break
-
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- resp, aggregate = self.client.create_aggregate(aggregate_name)
- self.addCleanup(self.client.delete_aggregate, aggregate['id'])
-
- self.assertRaises(exceptions.NotFound, self.client.add_host,
- aggregate['id'], non_exist_host)
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_add_host_as_user(self):
- # Regular user is not allowed to add a host to an aggregate.
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- resp, aggregate = self.client.create_aggregate(aggregate_name)
- self.addCleanup(self.client.delete_aggregate, aggregate['id'])
-
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.add_host,
- aggregate['id'], self.host)
-
- @attr(type=['negative', 'gate'])
- def test_aggregate_remove_host_as_user(self):
- # Regular user is not allowed to remove a host from an aggregate.
- self.useFixture(fixtures.LockFixture('availability_zone'))
- aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
- resp, aggregate = self.client.create_aggregate(aggregate_name)
- self.addCleanup(self.client.delete_aggregate, aggregate['id'])
- self.client.add_host(aggregate['id'], self.host)
- self.addCleanup(self.client.remove_host, aggregate['id'], self.host)
-
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.remove_host,
- aggregate['id'], self.host)
-
class AggregatesAdminTestXML(AggregatesAdminTestJSON):
_host_key = (
diff --git a/tempest/api/compute/admin/test_aggregates_negative.py b/tempest/api/compute/admin/test_aggregates_negative.py
new file mode 100644
index 0000000..8506206
--- /dev/null
+++ b/tempest/api/compute/admin/test_aggregates_negative.py
@@ -0,0 +1,196 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Huawei 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.
+
+from tempest.api.compute import base
+from tempest.common import tempest_fixtures as fixtures
+from tempest.common.utils import data_utils
+from tempest import exceptions
+from tempest.test import attr
+
+
+class AggregatesAdminNegativeTestJSON(base.BaseV2ComputeAdminTest):
+
+ """
+ Tests Aggregates API that require admin privileges
+ """
+
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(AggregatesAdminNegativeTestJSON, cls).setUpClass()
+ cls.client = cls.os_adm.aggregates_client
+ cls.user_client = cls.aggregates_client
+ cls.aggregate_name_prefix = 'test_aggregate_'
+ cls.az_name_prefix = 'test_az_'
+
+ resp, hosts_all = cls.os_adm.hosts_client.list_hosts()
+ hosts = map(lambda x: x['host_name'],
+ filter(lambda y: y['service'] == 'compute', hosts_all))
+ cls.host = hosts[0]
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_create_as_user(self):
+ # Regular user is not allowed to create an aggregate.
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.create_aggregate,
+ aggregate_name)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_create_aggregate_name_length_less_than_1(self):
+ # the length of aggregate name should >= 1 and <=255
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_aggregate,
+ '')
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_create_aggregate_name_length_exceeds_255(self):
+ # the length of aggregate name should >= 1 and <=255
+ aggregate_name = 'a' * 256
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_aggregate,
+ aggregate_name)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_create_with_existent_aggregate_name(self):
+ # creating an aggregate with existent aggregate name is forbidden
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.Conflict,
+ self.client.create_aggregate,
+ aggregate_name)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_delete_as_user(self):
+ # Regular user is not allowed to delete an aggregate.
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.delete_aggregate,
+ aggregate['id'])
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_list_as_user(self):
+ # Regular user is not allowed to list aggregates.
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.list_aggregates)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_get_details_as_user(self):
+ # Regular user is not allowed to get aggregate details.
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.get_aggregate,
+ aggregate['id'])
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_delete_with_invalid_id(self):
+ # Delete an aggregate with invalid id should raise exceptions.
+ self.assertRaises(exceptions.NotFound,
+ self.client.delete_aggregate, -1)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_get_details_with_invalid_id(self):
+ # Get aggregate details with invalid id should raise exceptions.
+ self.assertRaises(exceptions.NotFound,
+ self.client.get_aggregate, -1)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_add_non_exist_host(self):
+ # Adding a non-exist host to an aggregate should raise exceptions.
+ resp, hosts_all = self.os_adm.hosts_client.list_hosts()
+ hosts = map(lambda x: x['host_name'], hosts_all)
+ while True:
+ non_exist_host = data_utils.rand_name('nonexist_host_')
+ if non_exist_host not in hosts:
+ break
+
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.NotFound, self.client.add_host,
+ aggregate['id'], non_exist_host)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_add_host_as_user(self):
+ # Regular user is not allowed to add a host to an aggregate.
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.add_host,
+ aggregate['id'], self.host)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_add_existent_host(self):
+ self.useFixture(fixtures.LockFixture('availability_zone'))
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ resp, body = self.client.add_host(aggregate['id'], self.host)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.remove_host, aggregate['id'], self.host)
+
+ self.assertRaises(exceptions.Conflict, self.client.add_host,
+ aggregate['id'], self.host)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_remove_host_as_user(self):
+ # Regular user is not allowed to remove a host from an aggregate.
+ self.useFixture(fixtures.LockFixture('availability_zone'))
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+ resp, body = self.client.add_host(aggregate['id'], self.host)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.remove_host, aggregate['id'], self.host)
+
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.remove_host,
+ aggregate['id'], self.host)
+
+ @attr(type=['negative', 'gate'])
+ def test_aggregate_remove_nonexistent_host(self):
+ non_exist_host = data_utils.rand_name('nonexist_host_')
+ aggregate_name = data_utils.rand_name(self.aggregate_name_prefix)
+ resp, aggregate = self.client.create_aggregate(aggregate_name)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.client.delete_aggregate, aggregate['id'])
+
+ self.assertRaises(exceptions.NotFound, self.client.remove_host,
+ aggregate['id'], non_exist_host)
+
+
+class AggregatesAdminNegativeTestXML(AggregatesAdminNegativeTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/admin/test_availability_zone.py b/tempest/api/compute/admin/test_availability_zone.py
index d6488c4..e1a1a5d 100644
--- a/tempest/api/compute/admin/test_availability_zone.py
+++ b/tempest/api/compute/admin/test_availability_zone.py
@@ -16,21 +16,20 @@
# under the License.
from tempest.api.compute import base
-from tempest import exceptions
from tempest.test import attr
-class AvailabilityZoneAdminTestJSON(base.BaseV2ComputeAdminTest):
+class AZAdminTestJSON(base.BaseV2ComputeAdminTest):
"""
- Tests Availability Zone API List that require admin privileges
+ Tests Availability Zone API List
"""
_interface = 'json'
@classmethod
def setUpClass(cls):
- super(AvailabilityZoneAdminTestJSON, cls).setUpClass()
+ super(AZAdminTestJSON, cls).setUpClass()
cls.client = cls.os_adm.availability_zone_client
cls.non_adm_client = cls.availability_zone_client
@@ -57,14 +56,6 @@
self.assertEqual(200, resp.status)
self.assertTrue(len(availability_zone) > 0)
- @attr(type=['negative', 'gate'])
- def test_get_availability_zone_list_detail_with_non_admin_user(self):
- # List of availability zones and available services with
- # non-administrator user
- self.assertRaises(
- exceptions.Unauthorized,
- self.non_adm_client.get_availability_zone_list_detail)
-
-class AvailabilityZoneAdminTestXML(AvailabilityZoneAdminTestJSON):
+class AZAdminTestXML(AZAdminTestJSON):
_interface = 'xml'
diff --git a/tempest/api/compute/admin/test_availability_zone_negative.py b/tempest/api/compute/admin/test_availability_zone_negative.py
new file mode 100644
index 0000000..6ba8d58
--- /dev/null
+++ b/tempest/api/compute/admin/test_availability_zone_negative.py
@@ -0,0 +1,45 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright 2013 NEC Corporation. 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.
+
+from tempest.api.compute import base
+from tempest import exceptions
+from tempest.test import attr
+
+
+class AZAdminNegativeTestJSON(base.BaseV2ComputeAdminTest):
+
+ """
+ Tests Availability Zone API List
+ """
+
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(AZAdminNegativeTestJSON, cls).setUpClass()
+ cls.non_adm_client = cls.availability_zone_client
+
+ @attr(type=['negative', 'gate'])
+ def test_get_availability_zone_list_detail_with_non_admin_user(self):
+ # List of availability zones and available services with
+ # non-administrator user
+ self.assertRaises(
+ exceptions.Unauthorized,
+ self.non_adm_client.get_availability_zone_list_detail)
+
+
+class AZAdminNegativeTestXML(AZAdminNegativeTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/admin/test_flavors.py b/tempest/api/compute/admin/test_flavors.py
index cf72e49..0fb9460 100644
--- a/tempest/api/compute/admin/test_flavors.py
+++ b/tempest/api/compute/admin/test_flavors.py
@@ -130,40 +130,6 @@
flag = True
self.assertTrue(flag)
- @test.attr(type=['negative', 'gate'])
- def test_get_flavor_details_for_deleted_flavor(self):
- # Delete a flavor and ensure it is not listed
- # Create a test flavor
- flavor_name = data_utils.rand_name(self.flavor_name_prefix)
- new_flavor_id = data_utils.rand_int_id(start=1000)
-
- resp, flavor = self.client.create_flavor(flavor_name,
- self.ram,
- self.vcpus, self.disk,
- new_flavor_id,
- ephemeral=self.ephemeral,
- swap=self.swap,
- rxtx=self.rxtx)
- # Delete the flavor
- new_flavor_id = flavor['id']
- resp_delete, body = self.client.delete_flavor(new_flavor_id)
- self.assertEqual(200, resp.status)
- self.assertEqual(202, resp_delete.status)
-
- # Deleted flavors can be seen via detailed GET
- resp, flavor = self.client.get_flavor_details(new_flavor_id)
- self.assertEqual(resp.status, 200)
- self.assertEqual(flavor['name'], flavor_name)
-
- # Deleted flavors should not show up in a list however
- resp, flavors = self.client.list_flavors_with_detail()
- self.assertEqual(resp.status, 200)
- flag = True
- for flavor in flavors:
- if flavor['name'] == flavor_name:
- flag = False
- self.assertTrue(flag)
-
@test.attr(type='gate')
def test_create_list_flavor_without_extra_data(self):
# Create a flavor and ensure it is listed
@@ -347,49 +313,6 @@
self.assertEqual(flavor['ram'], int(ram))
self.assertEqual(int(flavor['id']), new_flavor_id)
- @test.attr(type=['negative', 'gate'])
- def test_invalid_is_public_string(self):
- self.assertRaises(exceptions.BadRequest,
- self.client.list_flavors_with_detail,
- {'is_public': 'invalid'})
-
- @test.attr(type=['negative', 'gate'])
- def test_create_flavor_as_user(self):
- flavor_name = data_utils.rand_name(self.flavor_name_prefix)
- new_flavor_id = data_utils.rand_int_id(start=1000)
-
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.create_flavor,
- flavor_name, self.ram, self.vcpus, self.disk,
- new_flavor_id, ephemeral=self.ephemeral,
- swap=self.swap, rxtx=self.rxtx)
-
- @test.attr(type=['negative', 'gate'])
- def test_delete_flavor_as_user(self):
- self.assertRaises(exceptions.Unauthorized,
- self.user_client.delete_flavor,
- self.flavor_ref_alt)
-
- @test.attr(type=['negative', 'gate'])
- def test_create_flavor_using_invalid_ram(self):
- flavor_name = data_utils.rand_name(self.flavor_name_prefix)
- new_flavor_id = data_utils.rand_int_id(start=1000)
-
- self.assertRaises(exceptions.BadRequest,
- self.client.create_flavor,
- flavor_name, -1, self.vcpus,
- self.disk, new_flavor_id)
-
- @test.attr(type=['negative', 'gate'])
- def test_create_flavor_using_invalid_vcpus(self):
- flavor_name = data_utils.rand_name(self.flavor_name_prefix)
- new_flavor_id = data_utils.rand_int_id(start=1000)
-
- self.assertRaises(exceptions.BadRequest,
- self.client.create_flavor,
- flavor_name, self.ram, 0,
- self.disk, new_flavor_id)
-
class FlavorsAdminTestXML(FlavorsAdminTestJSON):
_interface = 'xml'
diff --git a/tempest/api/compute/admin/test_flavors_negative.py b/tempest/api/compute/admin/test_flavors_negative.py
new file mode 100644
index 0000000..1d9198a
--- /dev/null
+++ b/tempest/api/compute/admin/test_flavors_negative.py
@@ -0,0 +1,344 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 OpenStack Foundation
+# 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 uuid
+
+from tempest.api.compute import base
+from tempest.common.utils import data_utils
+from tempest import exceptions
+from tempest import test
+
+
+class FlavorsAdminNegativeTestJSON(base.BaseV2ComputeAdminTest):
+
+ """
+ Tests Flavors API Create and Delete that require admin privileges
+ """
+
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(FlavorsAdminNegativeTestJSON, cls).setUpClass()
+ if not test.is_extension_enabled('FlavorExtraData', 'compute'):
+ msg = "FlavorExtraData extension not enabled."
+ raise cls.skipException(msg)
+
+ cls.client = cls.os_adm.flavors_client
+ cls.user_client = cls.os.flavors_client
+ cls.flavor_name_prefix = 'test_flavor_'
+ cls.ram = 512
+ cls.vcpus = 1
+ cls.disk = 10
+ cls.ephemeral = 10
+ cls.swap = 1024
+ cls.rxtx = 2
+
+ def flavor_clean_up(self, flavor_id):
+ resp, body = self.client.delete_flavor(flavor_id)
+ self.assertEqual(resp.status, 202)
+ self.client.wait_for_resource_deletion(flavor_id)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_get_flavor_details_for_deleted_flavor(self):
+ # Delete a flavor and ensure it is not listed
+ # Create a test flavor
+ flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+
+ # no need to specify flavor_id, we can get the flavor_id from a
+ # response of create_flavor() call.
+ resp, flavor = self.client.create_flavor(flavor_name,
+ self.ram,
+ self.vcpus, self.disk,
+ '',
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx)
+ # Delete the flavor
+ new_flavor_id = flavor['id']
+ resp_delete, body = self.client.delete_flavor(new_flavor_id)
+ self.assertEqual(200, resp.status)
+ self.assertEqual(202, resp_delete.status)
+
+ # Deleted flavors can be seen via detailed GET
+ resp, flavor = self.client.get_flavor_details(new_flavor_id)
+ self.assertEqual(resp.status, 200)
+ self.assertEqual(flavor['name'], flavor_name)
+
+ # Deleted flavors should not show up in a list however
+ resp, flavors = self.client.list_flavors_with_detail()
+ self.assertEqual(resp.status, 200)
+ flag = True
+ for flavor in flavors:
+ if flavor['name'] == flavor_name:
+ flag = False
+ self.assertTrue(flag)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_invalid_is_public_string(self):
+ # the 'is_public' parameter can be 'none/true/false' if it exists
+ self.assertRaises(exceptions.BadRequest,
+ self.client.list_flavors_with_detail,
+ {'is_public': 'invalid'})
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_as_user(self):
+ # only admin user can create a flavor
+ flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.create_flavor,
+ flavor_name, self.ram, self.vcpus, self.disk,
+ new_flavor_id, ephemeral=self.ephemeral,
+ swap=self.swap, rxtx=self.rxtx)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_delete_flavor_as_user(self):
+ # only admin user can delete a flavor
+ self.assertRaises(exceptions.Unauthorized,
+ self.user_client.delete_flavor,
+ self.flavor_ref_alt)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_using_invalid_ram(self):
+ # the 'ram' attribute must be positive integer
+ flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ flavor_name, -1, self.vcpus,
+ self.disk, new_flavor_id)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_using_invalid_vcpus(self):
+ # the 'vcpu' attribute must be positive integer
+ flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ flavor_name, self.ram, -1,
+ self.disk, new_flavor_id)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_name_length_less_than_1(self):
+ # ensure name length >= 1
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ '',
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_name_length_exceeds_255(self):
+ # ensure name do not exceed 255 characters
+ new_flavor_name = 'a' * 256
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_name(self):
+ # the regex of flavor_name is '^[\w\.\- ]*$'
+ invalid_flavor_name = data_utils.rand_name('invalid-!@#$%-')
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ invalid_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_flavor_id(self):
+ # the regex of flavor_id is '^[\w\.\- ]*$', and it cannot contain
+ # leading and/or trailing whitespace
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ invalid_flavor_id = '!@#$%'
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ invalid_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_id_length_exceeds_255(self):
+ # the length of flavor_id should not exceed 255 characters
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ invalid_flavor_id = 'a' * 256
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ invalid_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_root_gb(self):
+ # root_gb attribute should be non-negative ( >= 0) integer
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ -1,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_ephemeral_gb(self):
+ # ephemeral_gb attribute should be non-negative ( >= 0) integer
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=-1,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_swap(self):
+ # swap attribute should be non-negative ( >= 0) integer
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=-1,
+ rxtx=self.rxtx,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_rxtx_factor(self):
+ # rxtx_factor attribute should be a positive float
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=-1.5,
+ is_public='False')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_with_invalid_is_public(self):
+ # is_public attribute should be boolean
+ new_flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.BadRequest,
+ self.client.create_flavor,
+ new_flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx,
+ is_public='Invalid')
+
+ @test.attr(type=['negative', 'gate'])
+ def test_create_flavor_already_exists(self):
+ flavor_name = data_utils.rand_name(self.flavor_name_prefix)
+ new_flavor_id = str(uuid.uuid4())
+
+ resp, flavor = self.client.create_flavor(flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx)
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self.flavor_clean_up, flavor['id'])
+
+ self.assertRaises(exceptions.Conflict,
+ self.client.create_flavor,
+ flavor_name,
+ self.ram, self.vcpus,
+ self.disk,
+ new_flavor_id,
+ ephemeral=self.ephemeral,
+ swap=self.swap,
+ rxtx=self.rxtx)
+
+ @test.attr(type=['negative', 'gate'])
+ def test_delete_nonexistent_flavor(self):
+ nonexistent_flavor_id = str(uuid.uuid4())
+
+ self.assertRaises(exceptions.NotFound,
+ self.client.delete_flavor,
+ nonexistent_flavor_id)
+
+
+class FlavorsAdminNegativeTestXML(FlavorsAdminNegativeTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/admin/test_security_groups.py b/tempest/api/compute/admin/test_security_groups.py
new file mode 100644
index 0000000..5ed4823
--- /dev/null
+++ b/tempest/api/compute/admin/test_security_groups.py
@@ -0,0 +1,96 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 NTT Data
+# 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 testtools
+
+from tempest.api.compute import base
+from tempest.common.utils import data_utils
+from tempest import config
+from tempest.test import attr
+
+
+class SecurityGroupsTestAdminJSON(base.BaseV2ComputeAdminTest):
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(SecurityGroupsTestAdminJSON, cls).setUpClass()
+ cls.adm_client = cls.os_adm.security_groups_client
+ cls.client = cls.security_groups_client
+
+ def _delete_security_group(self, securitygroup_id, admin=True):
+ if admin:
+ resp, _ = self.adm_client.delete_security_group(securitygroup_id)
+ else:
+ resp, _ = self.client.delete_security_group(securitygroup_id)
+
+ self.assertEqual(202, resp.status)
+
+ @testtools.skipIf(config.TempestConfig().service_available.neutron,
+ "Skipped because neutron do not support all_tenants"
+ "search filter.")
+ @attr(type='smoke')
+ def test_list_security_groups_list_all_tenants_filter(self):
+ # Admin can list security groups of all tenants
+ # List of all security groups created
+ security_group_list = []
+ # Create two security groups for a non-admin tenant
+ for i in range(2):
+ name = data_utils.rand_name('securitygroup-')
+ description = data_utils.rand_name('description-')
+ resp, securitygroup = (self.client
+ .create_security_group(name, description))
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self._delete_security_group,
+ securitygroup['id'], admin=False)
+ security_group_list.append(securitygroup)
+
+ client_tenant_id = securitygroup['tenant_id']
+ # Create two security groups for admin tenant
+ for i in range(2):
+ name = data_utils.rand_name('securitygroup-')
+ description = data_utils.rand_name('description-')
+ resp, adm_securitygroup = (self.adm_client
+ .create_security_group(name,
+ description))
+ self.assertEqual(200, resp.status)
+ self.addCleanup(self._delete_security_group,
+ adm_securitygroup['id'])
+ security_group_list.append(adm_securitygroup)
+
+ # Fetch all security groups based on 'all_tenants' search filter
+ param = {'all_tenants': 'true'}
+ resp, fetched_list = self.adm_client.list_security_groups(params=param)
+ self.assertEqual(200, resp.status)
+ sec_group_id_list = map(lambda sg: sg['id'], fetched_list)
+ # Now check if all created Security Groups are present in fetched list
+ for sec_group in security_group_list:
+ self.assertIn(sec_group['id'], sec_group_id_list)
+
+ # Fetch all security groups for non-admin user with 'all_tenants'
+ # search filter
+ resp, fetched_list = self.client.list_security_groups(params=param)
+ self.assertEqual(200, resp.status)
+ # Now check if all created Security Groups are present in fetched list
+ for sec_group in fetched_list:
+ self.assertEqual(sec_group['tenant_id'], client_tenant_id,
+ "Failed to get all security groups for "
+ "non admin user.")
+
+
+class SecurityGroupsTestAdminXML(SecurityGroupsTestAdminJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/flavors/test_flavors_negative.py b/tempest/api/compute/flavors/test_flavors_negative.py
index 81e4f87..0b20e90 100644
--- a/tempest/api/compute/flavors/test_flavors_negative.py
+++ b/tempest/api/compute/flavors/test_flavors_negative.py
@@ -18,7 +18,6 @@
import uuid
from tempest.api.compute import base
-from tempest.common.utils import data_utils
from tempest import exceptions
from tempest.test import attr
@@ -31,14 +30,6 @@
super(FlavorsNegativeTestJSON, cls).setUpClass()
cls.client = cls.flavors_client
- # Generating a nonexistent flavor id
- resp, flavors = cls.client.list_flavors()
- flavor_ids = [flavor['id'] for flavor in flavors]
- while True:
- cls.nonexistent_flavor_id = data_utils.rand_int_id(start=999)
- if cls.nonexistent_flavor_id not in flavor_ids:
- break
-
@attr(type=['negative', 'gate'])
def test_invalid_minRam_filter(self):
self.assertRaises(exceptions.BadRequest,
@@ -52,17 +43,11 @@
{'minDisk': 'invalid'})
@attr(type=['negative', 'gate'])
- def test_get_flavor_details_for_invalid_flavor_id(self):
- # Ensure 404 returned for invalid flavor ID
- invalid_flavor_id = str(uuid.uuid4())
- self.assertRaises(exceptions.NotFound, self.client.get_flavor_details,
- invalid_flavor_id)
-
- @attr(type=['negative', 'gate'])
def test_non_existent_flavor_id(self):
# flavor details are not returned for non-existent flavors
+ nonexistent_flavor_id = str(uuid.uuid4())
self.assertRaises(exceptions.NotFound, self.client.get_flavor_details,
- self.nonexistent_flavor_id)
+ nonexistent_flavor_id)
class FlavorsNegativeTestXML(FlavorsNegativeTestJSON):
diff --git a/tempest/api/compute/images/test_images_oneserver.py b/tempest/api/compute/images/test_images_oneserver.py
index b0ff7ab..6e4c8cb 100644
--- a/tempest/api/compute/images/test_images_oneserver.py
+++ b/tempest/api/compute/images/test_images_oneserver.py
@@ -117,6 +117,20 @@
self.assertEqual('204', resp['status'])
self.client.wait_for_resource_deletion(image_id)
+ @attr(type=['gate'])
+ def test_create_image_specify_multibyte_character_image_name(self):
+ if self.__class__._interface == "xml":
+ # NOTE(sdague): not entirely accurage, but we'd need a ton of work
+ # in our XML client to make this good
+ raise self.skipException("Not testable in XML")
+ # prefix character is:
+ # http://www.fileformat.info/info/unicode/char/1F4A9/index.htm
+ utf8_name = data_utils.rand_name(u'\xF0\x9F\x92\xA9')
+ resp, body = self.client.create_image(self.server_id, utf8_name)
+ image_id = data_utils.parse_image_id(resp['location'])
+ self.addCleanup(self.client.delete_image, image_id)
+ self.assertEqual('202', resp['status'])
+
class ImagesOneServerTestXML(ImagesOneServerTestJSON):
_interface = 'xml'
diff --git a/tempest/api/compute/images/test_images_oneserver_negative.py b/tempest/api/compute/images/test_images_oneserver_negative.py
index 864f445..b8a4304 100644
--- a/tempest/api/compute/images/test_images_oneserver_negative.py
+++ b/tempest/api/compute/images/test_images_oneserver_negative.py
@@ -49,7 +49,10 @@
LOG.exception(exc)
# Rebuild server if cannot reach the ACTIVE state
# Usually it means the server had a serius accident
- self.__class__.server_id = self.rebuild_server(self.server_id)
+ self._reset_server()
+
+ def _reset_server(self):
+ self.__class__.server_id = self.rebuild_server(self.server_id)
@classmethod
def setUpClass(cls):
@@ -83,11 +86,15 @@
@skip_because(bug="1006725")
@attr(type=['negative', 'gate'])
def test_create_image_specify_multibyte_character_image_name(self):
- # Return an error if the image name has multi-byte characters
- snapshot_name = data_utils.rand_name('\xef\xbb\xbf')
+ if self.__class__._interface == "xml":
+ raise self.skipException("Not testable in XML")
+ # invalid multibyte sequence from:
+ # http://stackoverflow.com/questions/1301402/
+ # example-invalid-utf8-string
+ invalid_name = data_utils.rand_name(u'\xc3\x28')
self.assertRaises(exceptions.BadRequest,
self.client.create_image, self.server_id,
- snapshot_name)
+ invalid_name)
@attr(type=['negative', 'gate'])
def test_create_image_specify_invalid_metadata(self):
@@ -116,6 +123,7 @@
self.assertEqual(202, resp.status)
image_id = data_utils.parse_image_id(resp['location'])
self.image_ids.append(image_id)
+ self.addCleanup(self._reset_server)
# Create second snapshot
alt_snapshot_name = data_utils.rand_name('test-snap-')
@@ -139,6 +147,7 @@
self.assertEqual(202, resp.status)
image_id = data_utils.parse_image_id(resp['location'])
self.image_ids.append(image_id)
+ self.addCleanup(self._reset_server)
# Do not wait, attempt to delete the image, ensure it's successful
resp, body = self.client.delete_image(image_id)
diff --git a/tempest/api/compute/keypairs/test_keypairs.py b/tempest/api/compute/keypairs/test_keypairs.py
index 50b6c77..b36595c 100644
--- a/tempest/api/compute/keypairs/test_keypairs.py
+++ b/tempest/api/compute/keypairs/test_keypairs.py
@@ -17,7 +17,7 @@
from tempest.api.compute import base
from tempest.common.utils import data_utils
-from tempest.test import attr
+from tempest import test
class KeyPairsTestJSON(base.BaseV2ComputeTest):
@@ -28,14 +28,23 @@
super(KeyPairsTestJSON, cls).setUpClass()
cls.client = cls.keypairs_client
- @attr(type='gate')
+ def _delete_keypair(self, keypair_name):
+ resp, _ = self.client.delete_keypair(keypair_name)
+ self.assertEqual(202, resp.status)
+
+ def _create_keypair(self, keypair_name, pub_key=None):
+ resp, body = self.client.create_keypair(keypair_name, pub_key)
+ self.addCleanup(self._delete_keypair, keypair_name)
+ return resp, body
+
+ @test.attr(type='gate')
def test_keypairs_create_list_delete(self):
# Keypairs created should be available in the response list
# Create 3 keypairs
key_list = list()
for i in range(3):
k_name = data_utils.rand_name('keypair-')
- resp, keypair = self.client.create_keypair(k_name)
+ resp, keypair = self._create_keypair(k_name)
# Need to pop these keys so that our compare doesn't fail later,
# as the keypair dicts from list API doesn't have them.
keypair.pop('private_key')
@@ -57,16 +66,12 @@
self.assertFalse(missing_kps,
"Failed to find keypairs %s in fetched list"
% ', '.join(m_key['name'] for m_key in missing_kps))
- # Delete all the keypairs created
- for keypair in key_list:
- resp, _ = self.client.delete_keypair(keypair['name'])
- self.assertEqual(202, resp.status)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_keypair_create_delete(self):
# Keypair should be created, verified and deleted
k_name = data_utils.rand_name('keypair-')
- resp, keypair = self.client.create_keypair(k_name)
+ resp, keypair = self._create_keypair(k_name)
self.assertEqual(200, resp.status)
private_key = keypair['private_key']
key_name = keypair['name']
@@ -75,15 +80,12 @@
"to the requested name")
self.assertTrue(private_key is not None,
"Field private_key is empty or not found.")
- resp, _ = self.client.delete_keypair(k_name)
- self.assertEqual(202, resp.status)
- @attr(type='gate')
+ @test.attr(type='gate')
def test_get_keypair_detail(self):
# Keypair should be created, Got details by name and deleted
k_name = data_utils.rand_name('keypair-')
- resp, keypair = self.client.create_keypair(k_name)
- self.addCleanup(self.client.delete_keypair, k_name)
+ resp, keypair = self._create_keypair(k_name)
resp, keypair_detail = self.client.get_keypair(k_name)
self.assertEqual(200, resp.status)
self.assertIn('name', keypair_detail)
@@ -95,7 +97,7 @@
self.assertTrue(public_key is not None,
"Field public_key is empty or not found.")
- @attr(type='gate')
+ @test.attr(type='gate')
def test_keypair_create_with_pub_key(self):
# Keypair should be created with a given public key
k_name = data_utils.rand_name('keypair-')
@@ -108,7 +110,7 @@
"LOeB1kYMOBaiUPLQTWXR3JpckqFIQwhIH0zoHlJvZE8hh90"
"XcPojYN56tI0OlrGqojbediJYD0rUsJu4weZpbn8vilb3JuDY+jws"
"snSA8wzBx3A/8y9Pp1B nova@ubuntu")
- resp, keypair = self.client.create_keypair(k_name, pub_key)
+ resp, keypair = self._create_keypair(k_name, pub_key)
self.assertEqual(200, resp.status)
self.assertFalse('private_key' in keypair,
"Field private_key is not empty!")
@@ -116,8 +118,6 @@
self.assertEqual(key_name, k_name,
"The created keypair name is not equal "
"to the requested name!")
- resp, _ = self.client.delete_keypair(k_name)
- self.assertEqual(202, resp.status)
class KeyPairsTestXML(KeyPairsTestJSON):
diff --git a/tempest/api/compute/keypairs/test_keypairs_negative.py b/tempest/api/compute/keypairs/test_keypairs_negative.py
index fad985e..621487c 100644
--- a/tempest/api/compute/keypairs/test_keypairs_negative.py
+++ b/tempest/api/compute/keypairs/test_keypairs_negative.py
@@ -19,7 +19,7 @@
from tempest.api.compute import base
from tempest.common.utils import data_utils
from tempest import exceptions
-from tempest.test import attr
+from tempest import test
class KeyPairsNegativeTestJSON(base.BaseV2ComputeTest):
@@ -30,67 +30,71 @@
super(KeyPairsNegativeTestJSON, cls).setUpClass()
cls.client = cls.keypairs_client
- @attr(type=['negative', 'gate'])
+ def _create_keypair(self, keypair_name, pub_key=None):
+ self.client.create_keypair(keypair_name, pub_key)
+ self.addCleanup(self.client.delete_keypair, keypair_name)
+
+ @test.attr(type=['negative', 'gate'])
def test_keypair_create_with_invalid_pub_key(self):
# Keypair should not be created with a non RSA public key
k_name = data_utils.rand_name('keypair-')
pub_key = "ssh-rsa JUNK nova@ubuntu"
self.assertRaises(exceptions.BadRequest,
- self.client.create_keypair, k_name, pub_key)
+ self._create_keypair, k_name, pub_key)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_keypair_delete_nonexistant_key(self):
# Non-existant key deletion should throw a proper error
k_name = data_utils.rand_name("keypair-non-existant-")
self.assertRaises(exceptions.NotFound, self.client.delete_keypair,
k_name)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_with_empty_public_key(self):
# Keypair should not be created with an empty public key
k_name = data_utils.rand_name("keypair-")
pub_key = ' '
- self.assertRaises(exceptions.BadRequest, self.client.create_keypair,
+ self.assertRaises(exceptions.BadRequest, self._create_keypair,
k_name, pub_key)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_when_public_key_bits_exceeds_maximum(self):
# Keypair should not be created when public key bits are too long
k_name = data_utils.rand_name("keypair-")
pub_key = 'ssh-rsa ' + 'A' * 2048 + ' openstack@ubuntu'
- self.assertRaises(exceptions.BadRequest, self.client.create_keypair,
+ self.assertRaises(exceptions.BadRequest, self._create_keypair,
k_name, pub_key)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_with_duplicate_name(self):
# Keypairs with duplicate names should not be created
k_name = data_utils.rand_name('keypair-')
resp, _ = self.client.create_keypair(k_name)
self.assertEqual(200, resp.status)
# Now try the same keyname to create another key
- self.assertRaises(exceptions.Conflict, self.client.create_keypair,
+ self.assertRaises(exceptions.Conflict, self._create_keypair,
k_name)
resp, _ = self.client.delete_keypair(k_name)
self.assertEqual(202, resp.status)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_with_empty_name_string(self):
# Keypairs with name being an empty string should not be created
- self.assertRaises(exceptions.BadRequest, self.client.create_keypair,
+ self.assertRaises(exceptions.BadRequest, self._create_keypair,
'')
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_with_long_keynames(self):
# Keypairs with name longer than 255 chars should not be created
k_name = 'keypair-'.ljust(260, '0')
- self.assertRaises(exceptions.BadRequest, self.client.create_keypair,
+ self.assertRaises(exceptions.BadRequest, self._create_keypair,
k_name)
- @attr(type=['negative', 'gate'])
+ @test.attr(type=['negative', 'gate'])
def test_create_keypair_invalid_name(self):
# Keypairs with name being an invalid name should not be created
k_name = 'key_/.\@:'
- self.assertRaises(exceptions.BadRequest, self.client.create_keypair,
+ self.assertRaises(exceptions.BadRequest, self._create_keypair,
k_name)
diff --git a/tempest/api/compute/v3/admin/test_availability_zone.py b/tempest/api/compute/v3/admin/test_availability_zone.py
index ff2765c..dca556f 100644
--- a/tempest/api/compute/v3/admin/test_availability_zone.py
+++ b/tempest/api/compute/v3/admin/test_availability_zone.py
@@ -16,21 +16,20 @@
# under the License.
from tempest.api.compute import base
-from tempest import exceptions
from tempest.test import attr
-class AvailabilityZoneAdminV3TestJSON(base.BaseV3ComputeAdminTest):
+class AZAdminV3TestJSON(base.BaseV3ComputeAdminTest):
"""
- Tests Availability Zone API List that require admin privileges
+ Tests Availability Zone API List
"""
_interface = 'json'
@classmethod
def setUpClass(cls):
- super(AvailabilityZoneAdminV3TestJSON, cls).setUpClass()
+ super(AZAdminV3TestJSON, cls).setUpClass()
cls.client = cls.availability_zone_admin_client
cls.non_adm_client = cls.availability_zone_client
@@ -57,14 +56,6 @@
self.assertEqual(200, resp.status)
self.assertTrue(len(availability_zone) > 0)
- @attr(type=['negative', 'gate'])
- def test_get_availability_zone_list_detail_with_non_admin_user(self):
- # List of availability zones and available services with
- # non-administrator user
- self.assertRaises(
- exceptions.Unauthorized,
- self.non_adm_client.get_availability_zone_list_detail)
-
-class AvailabilityZoneAdminV3TestXML(AvailabilityZoneAdminV3TestJSON):
+class AZAdminV3TestXML(AZAdminV3TestJSON):
_interface = 'xml'
diff --git a/tempest/api/compute/v3/admin/test_availability_zone_negative.py b/tempest/api/compute/v3/admin/test_availability_zone_negative.py
new file mode 100644
index 0000000..93a57e3
--- /dev/null
+++ b/tempest/api/compute/v3/admin/test_availability_zone_negative.py
@@ -0,0 +1,47 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 NEC Corporation
+# 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.
+
+from tempest.api.compute import base
+from tempest import exceptions
+from tempest.test import attr
+
+
+class AZAdminNegativeV3TestJSON(base.BaseV3ComputeAdminTest):
+
+ """
+ Tests Availability Zone API List
+ """
+
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(AZAdminNegativeV3TestJSON, cls).setUpClass()
+ cls.client = cls.availability_zone_admin_client
+ cls.non_adm_client = cls.availability_zone_client
+
+ @attr(type=['negative', 'gate'])
+ def test_get_availability_zone_list_detail_with_non_admin_user(self):
+ # List of availability zones and available services with
+ # non-administrator user
+ self.assertRaises(
+ exceptions.Unauthorized,
+ self.non_adm_client.get_availability_zone_list_detail)
+
+
+class AZAdminNegativeV3TestXML(AZAdminNegativeV3TestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/compute/volumes/test_volumes_list.py b/tempest/api/compute/volumes/test_volumes_list.py
index f54e9b3..b57dcfe 100644
--- a/tempest/api/compute/volumes/test_volumes_list.py
+++ b/tempest/api/compute/volumes/test_volumes_list.py
@@ -109,6 +109,65 @@
', '.join(m_vol['displayName']
for m_vol in missing_volumes))
+ @attr(type='gate')
+ def test_volume_list_param_limit(self):
+ # Return the list of volumes based on limit set
+ params = {'limit': 2}
+ resp, fetched_vol_list = self.client.list_volumes(params=params)
+ self.assertEqual(200, resp.status)
+
+ self.assertEqual(len(fetched_vol_list), params['limit'],
+ "Failed to list volumes by limit set")
+
+ @attr(type='gate')
+ def test_volume_list_with_detail_param_limit(self):
+ # Return the list of volumes with details based on limit set.
+ params = {'limit': 2}
+ resp, fetched_vol_list = \
+ self.client.list_volumes_with_detail(params=params)
+ self.assertEqual(200, resp.status)
+
+ self.assertEqual(len(fetched_vol_list), params['limit'],
+ "Failed to list volume details by limit set")
+
+ @attr(type='gate')
+ def test_volume_list_param_offset_and_limit(self):
+ # Return the list of volumes based on offset and limit set.
+ # get all volumes list
+ response, all_vol_list = self.client.list_volumes()
+ params = {'offset': 1, 'limit': 1}
+ resp, fetched_vol_list = self.client.list_volumes(params=params)
+ self.assertEqual(200, resp.status)
+
+ # Validating length of the fetched volumes
+ self.assertEqual(len(fetched_vol_list), params['limit'],
+ "Failed to list volumes by offset and limit")
+ # Validating offset of fetched volume
+ for index, volume in enumerate(fetched_vol_list):
+ self.assertEqual(volume['id'],
+ all_vol_list[index + params['offset']]['id'],
+ "Failed to list volumes by offset and limit")
+
+ @attr(type='gate')
+ def test_volume_list_with_detail_param_offset_and_limit(self):
+ # Return the list of volumes details based on offset and limit set.
+ # get all volumes list
+ response, all_vol_list = self.client.list_volumes_with_detail()
+ params = {'offset': 1, 'limit': 1}
+ resp, fetched_vol_list = \
+ self.client.list_volumes_with_detail(params=params)
+ self.assertEqual(200, resp.status)
+
+ # Validating length of the fetched volumes
+ self.assertEqual(len(fetched_vol_list), params['limit'],
+ "Failed to list volume details by offset and limit")
+ # Validating offset of fetched volume
+ for index, volume in enumerate(fetched_vol_list):
+ self.assertEqual(volume['id'],
+ all_vol_list[index + params['offset']]['id'],
+ "Failed to list volume details by "
+ "offset and limit")
+
class VolumesTestXML(VolumesTestJSON):
_interface = 'xml'
diff --git a/tempest/api/identity/admin/v3/test_trusts.py b/tempest/api/identity/admin/v3/test_trusts.py
new file mode 100644
index 0000000..5e13a5a
--- /dev/null
+++ b/tempest/api/identity/admin/v3/test_trusts.py
@@ -0,0 +1,222 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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 tempest.api.identity import base
+from tempest import clients
+from tempest.common.utils.data_utils import rand_name
+from tempest import exceptions
+from tempest.test import attr
+
+
+class BaseTrustsV3Test(base.BaseIdentityAdminTest):
+
+ def setUp(self):
+ super(BaseTrustsV3Test, self).setUp()
+ # Use alt_username as the trustee
+ self.trustee_username = self.config.identity.alt_username
+
+ self.trust_id = None
+ self.create_trustor_and_roles()
+ self.addCleanup(self.cleanup_trust_user_and_roles)
+
+ def create_trustor_and_roles(self):
+ # Get trustor project ID, use the admin project
+ self.trustor_project_name = self.v3_client.tenant_name
+ self.trustor_project_id = self.get_tenant_by_name(
+ self.trustor_project_name)['id']
+ self.assertIsNotNone(self.trustor_project_id)
+
+ # Create a trustor User
+ self.trustor_username = rand_name('user-')
+ u_desc = self.trustor_username + 'description'
+ u_email = self.trustor_username + '@testmail.tm'
+ self.trustor_password = rand_name('pass-')
+ resp, user = self.v3_client.create_user(
+ self.trustor_username,
+ description=u_desc,
+ password=self.trustor_password,
+ email=u_email,
+ project_id=self.trustor_project_id)
+ self.assertEqual(resp['status'], '201')
+ self.trustor_user_id = user['id']
+
+ # And two roles, one we'll delegate and one we won't
+ self.delegated_role = rand_name('DelegatedRole-')
+ self.not_delegated_role = rand_name('NotDelegatedRole-')
+
+ resp, role = self.v3_client.create_role(self.delegated_role)
+ self.assertEqual(resp['status'], '201')
+ self.delegated_role_id = role['id']
+
+ resp, role = self.v3_client.create_role(self.not_delegated_role)
+ self.assertEqual(resp['status'], '201')
+ self.not_delegated_role_id = role['id']
+
+ # Assign roles to trustor
+ self.v3_client.assign_user_role(self.trustor_project_id,
+ self.trustor_user_id,
+ self.delegated_role_id)
+ self.v3_client.assign_user_role(self.trustor_project_id,
+ self.trustor_user_id,
+ self.not_delegated_role_id)
+
+ # Get trustee user ID, use the demo user
+ trustee_username = self.v3_non_admin_client.user
+ self.trustee_user_id = self.get_user_by_name(trustee_username)['id']
+ self.assertIsNotNone(self.trustee_user_id)
+
+ # Initialize a new client with the trustor credentials
+ os = clients.Manager(username=self.trustor_username,
+ password=self.trustor_password,
+ tenant_name=self.trustor_project_name,
+ interface=self._interface)
+ self.trustor_v3_client = os.identity_v3_client
+
+ def cleanup_trust_user_and_roles(self):
+ if self.trust_id:
+ try:
+ self.trustor_v3_client.delete_trust(self.trust_id)
+ except exceptions.NotFound:
+ pass
+ self.trust_id = None
+
+ if self.trustor_user_id:
+ self.v3_client.delete_user(self.trustor_user_id)
+ if self.delegated_role_id:
+ self.v3_client.delete_role(self.delegated_role_id)
+ if self.not_delegated_role_id:
+ self.v3_client.delete_role(self.not_delegated_role_id)
+
+ def create_trust(self, impersonate=True, expires=None):
+
+ resp, trust_create = self.trustor_v3_client.create_trust(
+ trustor_user_id=self.trustor_user_id,
+ trustee_user_id=self.trustee_user_id,
+ project_id=self.trustor_project_id,
+ role_names=[self.delegated_role],
+ impersonation=impersonate,
+ expires_at=expires)
+ self.assertEqual('201', resp['status'])
+ self.trust_id = trust_create['id']
+ return trust_create
+
+ def validate_trust(self, trust, impersonate=True, expires=None,
+ summary=False):
+ self.assertIsNotNone(trust['id'])
+ self.assertEqual(impersonate, trust['impersonation'])
+ self.assertEqual(expires, trust['expires_at'])
+ self.assertEqual(self.trustor_user_id, trust['trustor_user_id'])
+ self.assertEqual(self.trustee_user_id, trust['trustee_user_id'])
+ self.assertIn('v3/OS-TRUST/trusts', trust['links']['self'])
+ self.assertEqual(self.trustor_project_id, trust['project_id'])
+ if not summary:
+ self.assertEqual(self.delegated_role, trust['roles'][0]['name'])
+ self.assertEqual(1, len(trust['roles']))
+
+ def get_trust(self):
+ resp, trust_get = self.trustor_v3_client.get_trust(self.trust_id)
+ self.assertEqual('200', resp['status'])
+ return trust_get
+
+ def validate_role(self, role):
+ self.assertEqual(self.delegated_role_id, role['id'])
+ self.assertEqual(self.delegated_role, role['name'])
+ self.assertIn('v3/roles/%s' % self.delegated_role_id,
+ role['links']['self'])
+ self.assertNotEqual(self.not_delegated_role_id, role['id'])
+ self.assertNotEqual(self.not_delegated_role, role['name'])
+ self.assertNotIn('v3/roles/%s' % self.not_delegated_role_id,
+ role['links']['self'])
+
+ def check_trust_roles(self):
+ # Check we find the delegated role
+ resp, roles_get = self.trustor_v3_client.get_trust_roles(
+ self.trust_id)
+ self.assertEqual('200', resp['status'])
+ self.assertEqual(1, len(roles_get))
+ self.validate_role(roles_get[0])
+
+ resp, role_get = self.trustor_v3_client.get_trust_role(
+ self.trust_id, self.delegated_role_id)
+ self.assertEqual('200', resp['status'])
+ self.validate_role(role_get)
+
+ resp, role_get = self.trustor_v3_client.check_trust_role(
+ self.trust_id, self.delegated_role_id)
+ self.assertEqual('204', resp['status'])
+
+ # And that we don't find not_delegated_role
+ self.assertRaises(exceptions.NotFound,
+ self.trustor_v3_client.get_trust_role,
+ self.trust_id,
+ self.not_delegated_role_id)
+
+ self.assertRaises(exceptions.NotFound,
+ self.trustor_v3_client.check_trust_role,
+ self.trust_id,
+ self.not_delegated_role_id)
+
+ def delete_trust(self):
+ resp, trust_delete = self.trustor_v3_client.delete_trust(self.trust_id)
+ self.assertEqual('204', resp['status'])
+ self.assertRaises(exceptions.NotFound,
+ self.trustor_v3_client.get_trust,
+ self.trust_id)
+ self.trust_id = None
+
+
+class TrustsV3TestJSON(BaseTrustsV3Test):
+ _interface = 'json'
+
+ def setUp(self):
+ super(TrustsV3TestJSON, self).setUp()
+ self.create_trustor_and_roles()
+
+ @attr(type='smoke')
+ def test_trust_impersonate(self):
+ # Test case to check we can create, get and delete a trust
+ # updates are not supported for trusts
+ trust = self.create_trust()
+ self.validate_trust(trust)
+
+ trust_get = self.get_trust()
+ self.validate_trust(trust_get)
+
+ self.check_trust_roles()
+
+ self.delete_trust()
+
+ @attr(type='smoke')
+ def test_trust_noimpersonate(self):
+ # Test case to check we can create, get and delete a trust
+ # with impersonation=False
+ trust = self.create_trust(impersonate=False)
+ self.validate_trust(trust, impersonate=False)
+
+ trust_get = self.get_trust()
+ self.validate_trust(trust_get, impersonate=False)
+
+ self.check_trust_roles()
+
+ self.delete_trust()
+
+ @attr(type='smoke')
+ def test_trust_expire_invalid(self):
+ # Test case to check we can check an invlaid expiry time
+ # is rejected with the correct error
+ # with an expiry specified
+ expires_str = 'bad.123Z'
+ self.assertRaises(exceptions.BadRequest,
+ self.create_trust,
+ expires=expires_str)
diff --git a/tempest/api/network/admin/test_agent_management.py b/tempest/api/network/admin/test_agent_management.py
new file mode 100644
index 0000000..94659b2
--- /dev/null
+++ b/tempest/api/network/admin/test_agent_management.py
@@ -0,0 +1,82 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 IBM Corp.
+#
+# 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 tempest.api.network import base
+from tempest.common import tempest_fixtures as fixtures
+from tempest.test import attr
+
+
+class AgentManagementTestJSON(base.BaseAdminNetworkTest):
+ _interface = 'json'
+
+ @classmethod
+ def setUpClass(cls):
+ super(AgentManagementTestJSON, cls).setUpClass()
+ resp, body = cls.admin_client.list_agents()
+ agents = body['agents']
+ cls.agent = agents[0]
+
+ @attr(type='smoke')
+ def test_list_agent(self):
+ resp, body = self.admin_client.list_agents()
+ self.assertEqual('200', resp['status'])
+ agents = body['agents']
+ self.assertIn(self.agent, agents)
+
+ @attr(type='smoke')
+ def test_show_agent(self):
+ resp, body = self.admin_client.show_agent(self.agent['id'])
+ agent = body['agent']
+ self.assertEqual('200', resp['status'])
+ self.assertEqual(agent['id'], self.agent['id'])
+
+ @attr(type='smoke')
+ def test_update_agent_status(self):
+ origin_status = self.agent['admin_state_up']
+ # Try to update the 'admin_state_up' to the original
+ # one to avoid the negative effect.
+ agent_status = {'admin_state_up': origin_status}
+ resp, body = self.admin_client.update_agent(agent_id=self.agent['id'],
+ agent_info=agent_status)
+ updated_status = body['agent']['admin_state_up']
+ self.assertEqual('200', resp['status'])
+ self.assertEqual(origin_status, updated_status)
+
+ @attr(type='smoke')
+ def test_update_agent_description(self):
+ self.useFixture(fixtures.LockFixture('agent_description'))
+ description = 'description for update agent.'
+ agent_description = {'description': description}
+ resp, body = self.admin_client.update_agent(
+ agent_id=self.agent['id'],
+ agent_info=agent_description)
+ self.assertEqual('200', resp['status'])
+ self.addCleanup(self._restore_agent)
+ updated_description = body['agent']['description']
+ self.assertEqual(updated_description, description)
+
+ def _restore_agent(self):
+ """
+ Restore the agent description after update test.
+ """
+ description = self.agent['description'] or ''
+ origin_agent = {'description': description}
+ self.admin_client.update_agent(agent_id=self.agent['id'],
+ agent_info=origin_agent)
+
+
+class AgentManagementTestXML(AgentManagementTestJSON):
+ _interface = 'xml'
diff --git a/tempest/api/network/common.py b/tempest/api/network/common.py
index ab19fa8..528a204 100644
--- a/tempest/api/network/common.py
+++ b/tempest/api/network/common.py
@@ -47,6 +47,9 @@
def delete(self):
raise NotImplemented()
+ def __hash__(self):
+ return id(self)
+
class DeletableNetwork(DeletableResource):
@@ -86,6 +89,23 @@
class DeletableFloatingIp(DeletableResource):
+ def update(self, *args, **kwargs):
+ result = self.client.update_floatingip(floatingip=self.id,
+ body=dict(
+ floatingip=dict(*args,
+ **kwargs)
+ ))
+ super(DeletableFloatingIp, self).update(**result['floatingip'])
+
+ def __repr__(self):
+ return '<%s addr="%s">' % (self.__class__.__name__,
+ self.floating_ip_address)
+
+ def __str__(self):
+ return '<"FloatingIP" addr="%s" id="%s">' % (self.__class__.__name__,
+ self.floating_ip_address,
+ self.id)
+
def delete(self):
self.client.delete_floatingip(self.id)
diff --git a/tempest/api/network/test_routers.py b/tempest/api/network/test_routers.py
index 3cbe23f..06e07bb 100644
--- a/tempest/api/network/test_routers.py
+++ b/tempest/api/network/test_routers.py
@@ -15,6 +15,8 @@
# License for the specific language governing permissions and limitations
# under the License.
+import netaddr
+
from tempest.api.network import base
from tempest.common.utils import data_utils
from tempest.test import attr
@@ -229,3 +231,22 @@
{'network_id': self.network_cfg.public_network_id,
'enable_snat': False})
self._verify_gateway_port(router['id'])
+
+ @attr(type='smoke')
+ def test_update_extra_route(self):
+ self.network = self.create_network()
+ self.name = self.network['name']
+ self.subnet = self.create_subnet(self.network)
+ # Add router interface with subnet id
+ self.router = self.create_router(data_utils.rand_name('router-'), True)
+ self.create_router_interface(self.router['id'], self.subnet['id'])
+ self.addCleanup(
+ self._delete_extra_routes,
+ self.router['id'])
+ # Update router extra route
+ cidr = netaddr.IPNetwork(self.subnet['cidr'])
+ resp, extra_route = self.client.update_extra_routes(
+ self.router['id'], str(cidr[0]), str(self.subnet['cidr']))
+
+ def _delete_extra_routes(self, router_id):
+ resp, _ = self.client.delete_extra_routes(router_id)
diff --git a/tempest/api/volume/test_volumes_list.py b/tempest/api/volume/test_volumes_list.py
index 4dbc88a..def330e 100644
--- a/tempest/api/volume/test_volumes_list.py
+++ b/tempest/api/volume/test_volumes_list.py
@@ -57,13 +57,13 @@
# Create 3 test volumes
cls.volume_list = []
cls.volume_id_list = []
+ cls.metadata = {'Type': 'work'}
for i in range(3):
v_name = data_utils.rand_name('volume')
- metadata = {'Type': 'work'}
try:
resp, volume = cls.client.create_volume(size=1,
display_name=v_name,
- metadata=metadata)
+ metadata=cls.metadata)
cls.client.wait_for_volume_status(volume['id'], 'available')
resp, volume = cls.client.get_volume(volume['id'])
cls.volume_list.append(volume)
@@ -88,6 +88,25 @@
cls.client.wait_for_resource_deletion(volid)
super(VolumesListTest, cls).tearDownClass()
+ def _list_by_param_value_and_assert(self, params, with_detail=False):
+ """
+ Perform list or list_details action with given params
+ and validates result.
+ """
+ if with_detail:
+ resp, fetched_vol_list = \
+ self.client.list_volumes_with_detail(params=params)
+ else:
+ resp, fetched_vol_list = self.client.list_volumes(params=params)
+
+ self.assertEqual(200, resp.status)
+ # Validating params of fetched volumes
+ for volume in fetched_vol_list:
+ for key in params:
+ msg = "Failed to list volumes %s by %s" % \
+ ('details' if with_detail else '', key)
+ self.assertEqual(params[key], volume[key], msg)
+
@attr(type='smoke')
def test_volume_list(self):
# Get a list of Volumes
@@ -164,6 +183,34 @@
self.assertEqual(zone, volume['availability_zone'])
self.assertVolumesIn(fetched_list, self.volume_list)
+ @attr(type='gate')
+ def test_volume_list_with_param_metadata(self):
+ # Test to list volumes when metadata param is given
+ params = {'metadata': self.metadata}
+ self._list_by_param_value_and_assert(params)
+
+ @attr(type='gate')
+ def test_volume_list_with_detail_param_metadata(self):
+ # Test to list volumes details when metadata param is given
+ params = {'metadata': self.metadata}
+ self._list_by_param_value_and_assert(params, with_detail=True)
+
+ @attr(type='gate')
+ def test_volume_list_param_display_name_and_status(self):
+ # Test to list volume when display name and status param is given
+ volume = self.volume_list[data_utils.rand_int_id(0, 2)]
+ params = {'display_name': volume['display_name'],
+ 'status': 'available'}
+ self._list_by_param_value_and_assert(params)
+
+ @attr(type='gate')
+ def test_volume_list_with_detail_param_display_name_and_status(self):
+ # Test to list volume when name and status param is given
+ volume = self.volume_list[data_utils.rand_int_id(0, 2)]
+ params = {'display_name': volume['display_name'],
+ 'status': 'available'}
+ self._list_by_param_value_and_assert(params, with_detail=True)
+
class VolumeListTestXML(VolumesListTest):
_interface = 'xml'
diff --git a/tempest/api/volume/test_volumes_snapshots.py b/tempest/api/volume/test_volumes_snapshots.py
index 99e8de7..6c45c3d 100644
--- a/tempest/api/volume/test_volumes_snapshots.py
+++ b/tempest/api/volume/test_volumes_snapshots.py
@@ -37,6 +37,27 @@
def tearDownClass(cls):
super(VolumesSnapshotTest, cls).tearDownClass()
+ def _list_by_param_values_and_assert(self, params, with_detail=False):
+ """
+ Perform list or list_details action with given params
+ and validates result.
+ """
+ if with_detail:
+ resp, fetched_snap_list = \
+ self.snapshots_client.\
+ list_snapshots_with_detail(params=params)
+ else:
+ resp, fetched_snap_list = \
+ self.snapshots_client.list_snapshots(params=params)
+
+ self.assertEqual(200, resp.status)
+ # Validating params of fetched snapshots
+ for snap in fetched_snap_list:
+ for key in params:
+ msg = "Failed to list snapshots %s by %s" % \
+ ('details' if with_detail else '', key)
+ self.assertEqual(params[key], snap[key], msg)
+
@attr(type='gate')
def test_snapshot_create_get_list_update_delete(self):
# Create a snapshot
@@ -83,6 +104,46 @@
self.snapshots.remove(snapshot)
@attr(type='gate')
+ def test_snapshots_list_with_params(self):
+ """list snapshots with params."""
+ # Create a snapshot
+ display_name = data_utils.rand_name('snap')
+ snapshot = self.create_snapshot(self.volume_origin['id'],
+ display_name=display_name)
+
+ # Verify list snapshots by display_name filter
+ params = {'display_name': snapshot['display_name']}
+ self._list_by_param_values_and_assert(params)
+
+ # Verify list snapshots by status filter
+ params = {'status': 'available'}
+ self._list_by_param_values_and_assert(params)
+
+ # Verify list snapshots by status and display name filter
+ params = {'status': 'available',
+ 'display_name': snapshot['display_name']}
+ self._list_by_param_values_and_assert(params)
+
+ @attr(type='gate')
+ def test_snapshots_list_details_with_params(self):
+ """list snapshot details with params."""
+ # Create a snapshot
+ display_name = data_utils.rand_name('snap')
+ snapshot = self.create_snapshot(self.volume_origin['id'],
+ display_name=display_name)
+
+ # Verify list snapshot details by display_name filter
+ params = {'display_name': snapshot['display_name']}
+ self._list_by_param_values_and_assert(params, with_detail=True)
+ # Verify list snapshot details by status filter
+ params = {'status': 'available'}
+ self._list_by_param_values_and_assert(params, with_detail=True)
+ # Verify list snapshot details by status and display name filter
+ params = {'status': 'available',
+ 'display_name': snapshot['display_name']}
+ self._list_by_param_values_and_assert(params, with_detail=True)
+
+ @attr(type='gate')
def test_volume_from_snapshot(self):
# Create a temporary snap using wrapper method from base, then
# create a snap based volume, check resp code and deletes it
diff --git a/tempest/cli/simple_read_only/test_neutron.py b/tempest/cli/simple_read_only/test_neutron.py
index 047b17d..80376ab 100644
--- a/tempest/cli/simple_read_only/test_neutron.py
+++ b/tempest/cli/simple_read_only/test_neutron.py
@@ -44,35 +44,43 @@
raise cls.skipException(msg)
super(SimpleReadOnlyNeutronClientTest, cls).setUpClass()
+ @test.attr(type='smoke')
def test_neutron_fake_action(self):
self.assertRaises(subprocess.CalledProcessError,
self.neutron,
'this-does-not-exist')
+ @test.attr(type='smoke')
def test_neutron_net_list(self):
self.neutron('net-list')
+ @test.attr(type='smoke')
def test_neutron_ext_list(self):
ext = self.parser.listing(self.neutron('ext-list'))
self.assertTableStruct(ext, ['alias', 'name'])
+ @test.attr(type='smoke')
def test_neutron_dhcp_agent_list_hosting_net(self):
self.neutron('dhcp-agent-list-hosting-net',
params=CONF.compute.fixed_network_name)
+ @test.attr(type='smoke')
def test_neutron_agent_list(self):
agents = self.parser.listing(self.neutron('agent-list'))
field_names = ['id', 'agent_type', 'host', 'alive', 'admin_state_up']
self.assertTableStruct(agents, field_names)
+ @test.attr(type='smoke')
def test_neutron_floatingip_list(self):
self.neutron('floatingip-list')
@test.skip_because(bug="1240694")
+ @test.attr(type='smoke')
def test_neutron_meter_label_list(self):
self.neutron('meter-label-list')
@test.skip_because(bug="1240694")
+ @test.attr(type='smoke')
def test_neutron_meter_label_rule_list(self):
self.neutron('meter-label-rule-list')
@@ -83,40 +91,52 @@
if '404 Not Found' not in e.stderr:
self.fail('%s: Unexpected failure.' % command)
+ @test.attr(type='smoke')
def test_neutron_lb_healthmonitor_list(self):
self._test_neutron_lbaas_command('lb-healthmonitor-list')
+ @test.attr(type='smoke')
def test_neutron_lb_member_list(self):
self._test_neutron_lbaas_command('lb-member-list')
+ @test.attr(type='smoke')
def test_neutron_lb_pool_list(self):
self._test_neutron_lbaas_command('lb-pool-list')
+ @test.attr(type='smoke')
def test_neutron_lb_vip_list(self):
self._test_neutron_lbaas_command('lb-vip-list')
+ @test.attr(type='smoke')
def test_neutron_net_external_list(self):
self.neutron('net-external-list')
+ @test.attr(type='smoke')
def test_neutron_port_list(self):
self.neutron('port-list')
+ @test.attr(type='smoke')
def test_neutron_quota_list(self):
self.neutron('quota-list')
+ @test.attr(type='smoke')
def test_neutron_router_list(self):
self.neutron('router-list')
+ @test.attr(type='smoke')
def test_neutron_security_group_list(self):
security_grp = self.parser.listing(self.neutron('security-group-list'))
self.assertTableStruct(security_grp, ['id', 'name', 'description'])
+ @test.attr(type='smoke')
def test_neutron_security_group_rule_list(self):
self.neutron('security-group-rule-list')
+ @test.attr(type='smoke')
def test_neutron_subnet_list(self):
self.neutron('subnet-list')
+ @test.attr(type='smoke')
def test_neutron_help(self):
help_text = self.neutron('help')
lines = help_text.split('\n')
@@ -136,11 +156,14 @@
# Optional arguments:
+ @test.attr(type='smoke')
def test_neutron_version(self):
self.neutron('', flags='--version')
+ @test.attr(type='smoke')
def test_neutron_debug_net_list(self):
self.neutron('net-list', flags='--debug')
+ @test.attr(type='smoke')
def test_neutron_quiet_net_list(self):
self.neutron('net-list', flags='--quiet')
diff --git a/tempest/common/utils/linux/remote_client.py b/tempest/common/utils/linux/remote_client.py
index 0d0e794..e64a257 100644
--- a/tempest/common/utils/linux/remote_client.py
+++ b/tempest/common/utils/linux/remote_client.py
@@ -89,3 +89,7 @@
# usually to /dev/ttyS0
cmd = 'sudo sh -c "echo \\"%s\\" >/dev/console"' % message
return self.ssh_client.exec_command(cmd)
+
+ def ping_host(self, host):
+ cmd = 'ping -c1 -w1 %s' % host
+ return self.ssh_client.exec_command(cmd)
diff --git a/tempest/hacking/checks.py b/tempest/hacking/checks.py
index 4c1c107..2adc98e 100644
--- a/tempest/hacking/checks.py
+++ b/tempest/hacking/checks.py
@@ -21,6 +21,7 @@
PYTHON_CLIENT_RE = re.compile('import (%s)client' % '|'.join(PYTHON_CLIENTS))
TEST_DEFINITION = re.compile(r'^\s*def test.*')
+SETUPCLASS_DEFINITION = re.compile(r'^\s*def setUpClass')
SCENARIO_DECORATOR = re.compile(r'\s*@.*services\(')
@@ -52,6 +53,14 @@
"T104: Scenario tests require a service decorator")
+def no_setupclass_for_unit_tests(physical_line, filename):
+ if 'tempest/tests' in filename:
+ if SETUPCLASS_DEFINITION.match(physical_line):
+ return (physical_line.find('def'),
+ "T105: setUpClass can not be used with unit tests")
+
+
def factory(register):
register(import_no_clients_in_api)
register(scenario_tests_need_service_tags)
+ register(no_setupclass_for_unit_tests)
diff --git a/tempest/scenario/manager.py b/tempest/scenario/manager.py
index fa46cd7..b24d2b9 100644
--- a/tempest/scenario/manager.py
+++ b/tempest/scenario/manager.py
@@ -628,7 +628,15 @@
self.set_resource(data_utils.rand_name('floatingip-'), floating_ip)
return floating_ip
- def _ping_ip_address(self, ip_address):
+ def _disassociate_floating_ip(self, floating_ip):
+ """
+ :param floating_ip: type DeletableFloatingIp
+ """
+ floating_ip.update(port_id=None)
+ self.assertEqual(None, floating_ip.port_id)
+ return floating_ip
+
+ def _ping_ip_address(self, ip_address, should_succeed=True):
cmd = ['ping', '-c1', '-w1', ip_address]
def ping():
@@ -636,8 +644,7 @@
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
proc.wait()
- if proc.returncode == 0:
- return True
+ return (proc.returncode == 0) == should_succeed
return tempest.test.call_until_true(
ping, self.config.compute.ping_timeout, 1)
@@ -649,17 +656,37 @@
timeout=timeout)
return ssh_client.test_connection_auth()
- def _check_vm_connectivity(self, ip_address, username, private_key):
- self.assertTrue(self._ping_ip_address(ip_address),
- "Timed out waiting for %s to become "
- "reachable" % ip_address)
- self.assertTrue(self._is_reachable_via_ssh(
- ip_address,
- username,
- private_key,
- timeout=self.config.compute.ssh_timeout),
- 'Auth failure in connecting to %s@%s via ssh' %
- (username, ip_address))
+ def _check_vm_connectivity(self, ip_address,
+ username=None,
+ private_key=None,
+ should_connect=True):
+ """
+ :param ip_address: server to test against
+ :param username: server's ssh username
+ :param private_key: server's ssh private key to be used
+ :param should_connect: True/False indicates positive/negative test
+ positive - attempt ping and ssh
+ negative - attempt ping and fail if succeed
+
+ :raises: AssertError if the result of the connectivity check does
+ not match the value of the should_connect param
+ """
+ if should_connect:
+ msg = "Timed out waiting for %s to become reachable" % ip_address
+ else:
+ msg = "ip address %s is reachable" % ip_address
+ self.assertTrue(self._ping_ip_address(ip_address,
+ should_succeed=should_connect),
+ msg=msg)
+ if should_connect:
+ # no need to check ssh for negative connectivity
+ self.assertTrue(self._is_reachable_via_ssh(
+ ip_address,
+ username,
+ private_key,
+ timeout=self.config.compute.ssh_timeout),
+ 'Auth failure in connecting to %s@%s via ssh' %
+ (username, ip_address))
def _create_security_group_nova(self, client=None,
namestart='secgroup-smoke-',
@@ -826,6 +853,12 @@
return rules
+ def _ssh_to_server(self, server, private_key):
+ ssh_login = self.config.compute.image_ssh_user
+ return self.get_remote_client(server,
+ username=ssh_login,
+ private_key=private_key)
+
def _show_quota_network(self, tenant_id):
quota = self.network_client.show_quota(tenant_id)
return quota['quota']['network']
diff --git a/tempest/scenario/test_network_basic_ops.py b/tempest/scenario/test_network_basic_ops.py
index 1418b75..d605dff 100644
--- a/tempest/scenario/test_network_basic_ops.py
+++ b/tempest/scenario/test_network_basic_ops.py
@@ -41,10 +41,7 @@
def __init__(self, compute_client, floating_ip_map):
self.compute_client = compute_client
- self.unchecked = {}
- for k in floating_ip_map.keys():
- self.unchecked[k] = [f.floating_ip_address
- for f in floating_ip_map[k]]
+ self.unchecked = floating_ip_map.copy()
def run_checks(self):
"""Check for any remaining unverified floating IPs
@@ -56,16 +53,14 @@
"""
to_delete = []
loggable_map = {}
- for k, check_addrs in self.unchecked.iteritems():
- serverdata = self.compute_client.servers.get(k.id)
- for net_name, ip_addr in serverdata.networks.iteritems():
- for addr in ip_addr:
- if addr in check_addrs:
- check_addrs.remove(addr)
- if len(check_addrs) == 0:
- to_delete.append(k)
+ for check_addr, server in self.unchecked.iteritems():
+ serverdata = self.compute_client.servers.get(server.id)
+ ip_addr = [addr for sublist in serverdata.networks.values() for
+ addr in sublist]
+ if check_addr.floating_ip_address in ip_addr:
+ to_delete.append(check_addr)
else:
- loggable_map[k.id] = check_addrs
+ loggable_map[server.id] = check_addr
for to_del in to_delete:
del self.unchecked[to_del]
@@ -93,6 +88,9 @@
ssh server hosted at the IP address. This check guarantees
that the IP address is associated with the target VM.
+ - detach the floating-ip from the VM and verify that it becomes
+ unreachable
+
# TODO(mnewby) - Need to implement the following:
- the Tempest host can ssh into the VM via the IP address and
successfully execute the following:
@@ -297,30 +295,34 @@
"Timed out while waiting for the floating IP assignments "
"to propagate")
- def _assign_floating_ips(self):
+ def _create_and_associate_floating_ips(self):
public_network_id = self.config.network.public_network_id
for server in self.servers:
floating_ip = self._create_floating_ip(server, public_network_id)
- self.floating_ips.setdefault(server, [])
- self.floating_ips[server].append(floating_ip)
+ self.floating_ips[floating_ip] = server
- def _check_public_network_connectivity(self):
+ def _check_public_network_connectivity(self, should_connect=True):
# The target login is assumed to have been configured for
# key-based authentication by cloud-init.
ssh_login = self.config.compute.image_ssh_user
private_key = self.keypairs[self.tenant_id].private_key
try:
- for server, floating_ips in self.floating_ips.iteritems():
- for floating_ip in floating_ips:
- ip_address = floating_ip.floating_ip_address
- self._check_vm_connectivity(ip_address,
- ssh_login,
- private_key)
+ for floating_ip, server in self.floating_ips.iteritems():
+ ip_address = floating_ip.floating_ip_address
+ self._check_vm_connectivity(ip_address,
+ ssh_login,
+ private_key,
+ should_connect=should_connect)
except Exception as exc:
LOG.exception(exc)
debug.log_ip_ns()
raise exc
+ def _disassociate_floating_ips(self):
+ for floating_ip, server in self.floating_ips.iteritems():
+ self._disassociate_floating_ip(floating_ip)
+ self.floating_ips[floating_ip] = None
+
@attr(type='smoke')
@services('compute', 'network')
def test_network_basic_ops(self):
@@ -329,7 +331,9 @@
self._create_networks()
self._check_networks()
self._create_servers()
- self._assign_floating_ips()
+ self._create_and_associate_floating_ips()
self._wait_for_floating_ip_association()
self._check_tenant_network_connectivity()
- self._check_public_network_connectivity()
+ self._check_public_network_connectivity(should_connect=True)
+ self._disassociate_floating_ips()
+ self._check_public_network_connectivity(should_connect=False)
diff --git a/tempest/services/identity/v3/json/identity_client.py b/tempest/services/identity/v3/json/identity_client.py
index ec99d37..e457c1f 100644
--- a/tempest/services/identity/v3/json/identity_client.py
+++ b/tempest/services/identity/v3/json/identity_client.py
@@ -359,6 +359,66 @@
(domain_id, group_id, role_id))
return resp, body
+ def create_trust(self, trustor_user_id, trustee_user_id, project_id,
+ role_names, impersonation, expires_at):
+ """Creates a trust."""
+ roles = [{'name': n} for n in role_names]
+ post_body = {
+ 'trustor_user_id': trustor_user_id,
+ 'trustee_user_id': trustee_user_id,
+ 'project_id': project_id,
+ 'impersonation': impersonation,
+ 'roles': roles,
+ 'expires_at': expires_at
+ }
+ post_body = json.dumps({'trust': post_body})
+ resp, body = self.post('OS-TRUST/trusts', post_body, self.headers)
+ body = json.loads(body)
+ return resp, body['trust']
+
+ def delete_trust(self, trust_id):
+ """Deletes a trust."""
+ resp, body = self.delete("OS-TRUST/trusts/%s" % trust_id)
+ return resp, body
+
+ def get_trusts(self, trustor_user_id=None, trustee_user_id=None):
+ """GET trusts."""
+ if trustor_user_id:
+ resp, body = self.get("OS-TRUST/trusts?trustor_user_id=%s"
+ % trustor_user_id)
+ elif trustee_user_id:
+ resp, body = self.get("OS-TRUST/trusts?trustee_user_id=%s"
+ % trustee_user_id)
+ else:
+ resp, body = self.get("OS-TRUST/trusts")
+ body = json.loads(body)
+ return resp, body['trusts']
+
+ def get_trust(self, trust_id):
+ """GET trust."""
+ resp, body = self.get("OS-TRUST/trusts/%s" % trust_id)
+ body = json.loads(body)
+ return resp, body['trust']
+
+ def get_trust_roles(self, trust_id):
+ """GET roles delegated by a trust."""
+ resp, body = self.get("OS-TRUST/trusts/%s/roles" % trust_id)
+ body = json.loads(body)
+ return resp, body['roles']
+
+ def get_trust_role(self, trust_id, role_id):
+ """GET role delegated by a trust."""
+ resp, body = self.get("OS-TRUST/trusts/%s/roles/%s"
+ % (trust_id, role_id))
+ body = json.loads(body)
+ return resp, body['role']
+
+ def check_trust_role(self, trust_id, role_id):
+ """HEAD Check if role is delegated by a trust."""
+ resp, body = self.head("OS-TRUST/trusts/%s/roles/%s"
+ % (trust_id, role_id))
+ return resp, body
+
class V3TokenClientJSON(RestClient):
diff --git a/tempest/services/network/json/network_client.py b/tempest/services/network/json/network_client.py
index c6bd423..3199537 100644
--- a/tempest/services/network/json/network_client.py
+++ b/tempest/services/network/json/network_client.py
@@ -680,6 +680,24 @@
body = json.loads(body)
return resp, body
+ def show_agent(self, agent_id):
+ uri = '%s/agents/%s' % (self.uri_prefix, agent_id)
+ resp, body = self.get(uri, self.headers)
+ body = json.loads(body)
+ return resp, body
+
+ def update_agent(self, agent_id, agent_info):
+ """
+ :param agent_info: Agent update information.
+ E.g {"admin_state_up": True}
+ """
+ uri = '%s/agents/%s' % (self.uri_prefix, agent_id)
+ agent = {"agent": agent_info}
+ body = json.dumps(agent)
+ resp, body = self.put(uri, body=body, headers=self.headers)
+ body = json.loads(body)
+ return resp, body
+
def list_routers_on_l3_agent(self, agent_id):
uri = '%s/agents/%s/l3-routers' % (self.uri_prefix, agent_id)
resp, body = self.get(uri, self.headers)
@@ -754,3 +772,29 @@
resp, body = self.put(uri, body=body, headers=self.headers)
body = json.loads(body)
return resp, body
+
+ def update_extra_routes(self, router_id, nexthop, destination):
+ uri = '%s/routers/%s' % (self.uri_prefix, router_id)
+ put_body = {
+ 'router': {
+ 'routes': [{'nexthop': nexthop,
+ "destination": destination}]
+ }
+ }
+ body = json.dumps(put_body)
+ resp, body = self.put(uri, body=body, headers=self.headers)
+ body = json.loads(body)
+ return resp, body
+
+ def delete_extra_routes(self, router_id):
+ uri = '%s/routers/%s' % (self.uri_prefix, router_id)
+ null_routes = None
+ put_body = {
+ 'router': {
+ 'routes': null_routes
+ }
+ }
+ body = json.dumps(put_body)
+ resp, body = self.put(uri, body=body, headers=self.headers)
+ body = json.loads(body)
+ return resp, body
diff --git a/tempest/services/network/xml/network_client.py b/tempest/services/network/xml/network_client.py
index e11d4c1..c46a523 100755
--- a/tempest/services/network/xml/network_client.py
+++ b/tempest/services/network/xml/network_client.py
@@ -561,6 +561,23 @@
agents = {'agents': agents}
return resp, agents
+ def show_agent(self, agent_id):
+ uri = '%s/agents/%s' % (self.uri_prefix, agent_id)
+ resp, body = self.get(uri, self.headers)
+ body = _root_tag_fetcher_and_xml_to_json_parse(body)
+ return resp, body
+
+ def update_agent(self, agent_id, agent_info):
+ uri = '%s/agents/%s' % (self.uri_prefix, agent_id)
+ agent = Element('agent')
+ for (key, value) in agent_info.items():
+ p = Element(key, value)
+ agent.append(p)
+ resp, body = self.put(uri, body=str(Document(agent)),
+ headers=self.headers)
+ body = _root_tag_fetcher_and_xml_to_json_parse(body)
+ return resp, body
+
def list_routers_on_l3_agent(self, agent_id):
uri = '%s/agents/%s/l3-routers' % (self.uri_prefix, agent_id)
resp, body = self.get(uri, self.headers)
diff --git a/tempest/services/volume/json/snapshots_client.py b/tempest/services/volume/json/snapshots_client.py
index 5d980eb..9435122 100644
--- a/tempest/services/volume/json/snapshots_client.py
+++ b/tempest/services/volume/json/snapshots_client.py
@@ -44,7 +44,7 @@
body = json.loads(body)
return resp, body['snapshots']
- def list_snapshot_with_detail(self, params=None):
+ def list_snapshots_with_detail(self, params=None):
"""List the details of all snapshots."""
url = 'snapshots/detail'
if params:
diff --git a/tools/check_logs.py b/tools/check_logs.py
index 0c8fd21..963709b 100755
--- a/tools/check_logs.py
+++ b/tools/check_logs.py
@@ -33,7 +33,7 @@
def process_files(file_specs, url_specs, whitelists):
- regexp = re.compile(r"^.*(ERROR|CRITICAL).*\[.*\-.*\]")
+ regexp = re.compile(r"^.* (ERROR|CRITICAL) .*\[.*\-.*\]")
had_errors = False
for (name, filename) in file_specs:
whitelist = whitelists.get(name, [])
diff --git a/tox.ini b/tox.ini
index 1d7e1b7..c7f92ae 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,12 +12,15 @@
install_command = pip install -U {opts} {packages}
[testenv:py26]
+setenv = OS_TEST_PATH=./tempest/tests
commands = python setup.py test --slowest --testr-arg='tempest\.tests {posargs}'
[testenv:py33]
+setenv = OS_TEST_PATH=./tempest/tests
commands = python setup.py test --slowest --testr-arg='tempest\.tests {posargs}'
[testenv:py27]
+setenv = OS_TEST_PATH=./tempest/tests
commands = python setup.py test --slowest --testr-arg='tempest\.tests {posargs}'
[testenv:all]