Merge "Using revision_number to ensure no overlap in *one* network"
diff --git a/neutron/tests/tempest/api/admin/test_shared_network_extension.py b/neutron/tests/tempest/api/admin/test_shared_network_extension.py
index a637c3e..bc10745 100644
--- a/neutron/tests/tempest/api/admin/test_shared_network_extension.py
+++ b/neutron/tests/tempest/api/admin/test_shared_network_extension.py
@@ -192,6 +192,22 @@
)['rbac_policy']
return {'network': net, 'subnet': subnet, 'policy': pol}
+ @test.attr(type='smoke')
+ @test.idempotent_id('86c3529b-1231-40de-803c-bfffffff1eee')
+ def test_create_rbac_policy_with_target_tenant_none(self):
+ with testtools.ExpectedException(lib_exc.BadRequest):
+ self._make_admin_net_and_subnet_shared_to_tenant_id(
+ tenant_id=None)
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('86c3529b-1231-40de-803c-bfffffff1fff')
+ def test_create_rbac_policy_with_target_tenant_too_long_id(self):
+ with testtools.ExpectedException(lib_exc.BadRequest):
+ target_tenant = '1234' * 100
+ self._make_admin_net_and_subnet_shared_to_tenant_id(
+ tenant_id=target_tenant)
+
+ @test.attr(type='smoke')
@test.idempotent_id('86c3529b-1231-40de-803c-afffffff1fff')
def test_network_only_visible_to_policy_target(self):
net = self._make_admin_net_and_subnet_shared_to_tenant_id(
diff --git a/neutron/tests/tempest/api/clients.py b/neutron/tests/tempest/api/clients.py
index eb39882..c6f41d0 100644
--- a/neutron/tests/tempest/api/clients.py
+++ b/neutron/tests/tempest/api/clients.py
@@ -13,8 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
-from oslo_config import cfg
-
from tempest.lib.services.compute import keypairs_client
from tempest.lib.services.compute import servers_client
from tempest.lib.services.identity.v2 import tenants_client
@@ -30,22 +28,12 @@
"""
Top level manager for OpenStack tempest clients
"""
- try:
- default_params = {
- 'disable_ssl_certificate_validation':
- CONF.service_clients.disable_ssl_certificate_validation,
- 'ca_certs': CONF.service_clients.ca_certificates_file,
- 'trace_requests': CONF.debug.trace_requests
- }
- except cfg.NoSuchOptError:
- # TODO(armax): remove this except when a new tempest release
- # > 12.1.0 includes change 1afca56b05
- default_params = {
- 'disable_ssl_certificate_validation':
- CONF.identity.disable_ssl_certificate_validation,
- 'ca_certs': CONF.identity.ca_certificates_file,
- 'trace_requests': CONF.debug.trace_requests
- }
+ default_params = {
+ 'disable_ssl_certificate_validation':
+ CONF.identity.disable_ssl_certificate_validation,
+ 'ca_certs': CONF.identity.ca_certificates_file,
+ 'trace_requests': CONF.debug.trace_requests
+ }
# NOTE: Tempest uses timeout values of compute API if project specific
# timeout values don't exist.
diff --git a/neutron/tests/tempest/api/test_allowed_address_pair.py b/neutron/tests/tempest/api/test_allowed_address_pair.py
index 88818b2..e4c499c 100644
--- a/neutron/tests/tempest/api/test_allowed_address_pair.py
+++ b/neutron/tests/tempest/api/test_allowed_address_pair.py
@@ -19,8 +19,6 @@
from neutron.tests.tempest.api import base
from neutron.tests.tempest import config
-CONF = config.CONF
-
class AllowedAddressPairTestJSON(base.BaseNetworkTest):
diff --git a/neutron/tests/tempest/api/test_auto_allocated_topology.py b/neutron/tests/tempest/api/test_auto_allocated_topology.py
index afdfe8c..65c3057 100644
--- a/neutron/tests/tempest/api/test_auto_allocated_topology.py
+++ b/neutron/tests/tempest/api/test_auto_allocated_topology.py
@@ -22,11 +22,14 @@
class TestAutoAllocatedTopology(base.BaseAdminNetworkTest):
"""
- NOTE: This test may eventually migrate to Tempest.
-
- Tests the Get-Me-A-Network operation in the Neutron API
+ Tests the Get-Me-A-Network operations in the Neutron API
using the REST client for Neutron.
"""
+ # NOTE(armax): this is a precaution to avoid interference
+ # from other tests exercising this extension. So long as
+ # all tests are added under TestAutoAllocatedTopology,
+ # nothing bad should happen.
+ force_tenant_isolation = True
@classmethod
@test.requires_ext(extension="auto-allocated-topology", service="network")
@@ -101,3 +104,14 @@
# After the initial GET, the API should be idempotent
self.assertEqual(network_id1, network_id2)
self.assertEqual(resources_after1, resources_after2)
+
+ @test.idempotent_id('aabc0b02-cee4-11e5-9f3c-091127605a2b')
+ def test_delete_allocated_net_topology_as_tenant(self):
+ resources_before = self._count_topology_resources()
+ self.assertEqual((0, 0, 0), resources_before)
+ body = self.client.get_auto_allocated_topology()
+ topology = body['auto_allocated_topology']
+ self.assertIsNotNone(topology)
+ self.client.delete_auto_allocated_topology()
+ resources_after = self._count_topology_resources()
+ self.assertEqual((0, 0, 0), resources_after)
diff --git a/neutron/tests/tempest/api/test_extensions.py b/neutron/tests/tempest/api/test_extensions.py
index 1defb33..28029d8 100644
--- a/neutron/tests/tempest/api/test_extensions.py
+++ b/neutron/tests/tempest/api/test_extensions.py
@@ -13,9 +13,6 @@
from tempest import test
from neutron.tests.tempest.api import base
-from neutron.tests.tempest import config
-
-CONF = config.CONF
class ExtensionsTest(base.BaseNetworkTest):
diff --git a/neutron/tests/tempest/api/test_network_ip_availability.py b/neutron/tests/tempest/api/test_network_ip_availability.py
index 6a81128..324c3fd 100644
--- a/neutron/tests/tempest/api/test_network_ip_availability.py
+++ b/neutron/tests/tempest/api/test_network_ip_availability.py
@@ -24,8 +24,6 @@
from neutron_lib import constants as lib_constants
-CONF = config.CONF
-
# 3 IP addresses are taken from every total for IPv4 these are reserved
DEFAULT_IP4_RESERVED = 3
# 2 IP addresses are taken from every total for IPv6 these are reserved
diff --git a/neutron/tests/tempest/api/test_networks.py b/neutron/tests/tempest/api/test_networks.py
index 16fe81b..2c2991a 100644
--- a/neutron/tests/tempest/api/test_networks.py
+++ b/neutron/tests/tempest/api/test_networks.py
@@ -16,9 +16,6 @@
from tempest import test
from neutron.tests.tempest.api import base
-from neutron.tests.tempest import config
-
-CONF = config.CONF
class NetworksTestJSON(base.BaseNetworkTest):
@@ -95,7 +92,7 @@
resource = 'network'
- list_kwargs = {'shared': False}
+ list_kwargs = {'shared': False, 'router:external': False}
@classmethod
def resource_setup(cls):
diff --git a/neutron/tests/tempest/api/test_qos.py b/neutron/tests/tempest/api/test_qos.py
index 313715d..ca49758 100644
--- a/neutron/tests/tempest/api/test_qos.py
+++ b/neutron/tests/tempest/api/test_qos.py
@@ -75,6 +75,28 @@
self.assertTrue(retrieved_policy['shared'])
self.assertEqual([], retrieved_policy['rules'])
+ @test.idempotent_id('6e880e0f-bbfc-4e54-87c6-680f90e1b618')
+ def test_policy_update_forbidden_for_regular_tenants_own_policy(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='',
+ shared=False,
+ tenant_id=self.client.tenant_id)
+ self.assertRaises(
+ exceptions.Forbidden,
+ self.client.update_qos_policy,
+ policy['id'], description='test policy')
+
+ @test.idempotent_id('4ecfd7e7-47b6-4702-be38-be9235901a87')
+ def test_policy_update_forbidden_for_regular_tenants_foreign_policy(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='',
+ shared=False,
+ tenant_id=self.admin_client.tenant_id)
+ self.assertRaises(
+ exceptions.NotFound,
+ self.client.update_qos_policy,
+ policy['id'], description='test policy')
+
@test.idempotent_id('ee263db4-009a-4641-83e5-d0e83506ba4c')
def test_shared_policy_update(self):
policy = self.create_qos_policy(name='test-policy',
@@ -315,6 +337,13 @@
obtained_policy = self.client.show_qos_policy(policy['id'])['policy']
self.assertEqual(obtained_policy, policy)
+ @test.idempotent_id('aed8e2a6-22da-421b-89b9-935a2c1a1b50')
+ def test_policy_create_forbidden_for_regular_tenants(self):
+ self.assertRaises(
+ exceptions.Forbidden,
+ self.client.create_qos_policy,
+ 'test-policy', 'test policy', False)
+
class QosBandwidthLimitRuleTestJSON(base.BaseAdminNetworkTest):
@classmethod
@@ -412,13 +441,6 @@
self.create_qos_bandwidth_limit_rule,
'policy', 200, 1337)
- @test.idempotent_id('eed8e2a6-22da-421b-89b9-935a2c1a1b50')
- def test_policy_create_forbidden_for_regular_tenants(self):
- self.assertRaises(
- exceptions.Forbidden,
- self.client.create_qos_policy,
- 'test-policy', 'test policy', False)
-
@test.idempotent_id('a4a2e7ad-786f-4927-a85a-e545a93bd274')
def test_rule_create_forbidden_for_regular_tenants(self):
self.assertRaises(
@@ -426,6 +448,34 @@
self.client.create_bandwidth_limit_rule,
'policy', 1, 2)
+ @test.idempotent_id('1bfc55d9-6fd8-4293-ab3a-b1d69bf7cd2e')
+ def test_rule_update_forbidden_for_regular_tenants_own_policy(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False,
+ tenant_id=self.client.tenant_id)
+ rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],
+ max_kbps=1,
+ max_burst_kbps=1)
+ self.assertRaises(
+ exceptions.NotFound,
+ self.client.update_bandwidth_limit_rule,
+ policy['id'], rule['id'], max_kbps=2, max_burst_kbps=4)
+
+ @test.idempotent_id('9a607936-4b6f-4c2f-ad21-bd5b3d4fc91f')
+ def test_rule_update_forbidden_for_regular_tenants_foreign_policy(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False,
+ tenant_id=self.admin_client.tenant_id)
+ rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],
+ max_kbps=1,
+ max_burst_kbps=1)
+ self.assertRaises(
+ exceptions.NotFound,
+ self.client.update_bandwidth_limit_rule,
+ policy['id'], rule['id'], max_kbps=2, max_burst_kbps=4)
+
@test.idempotent_id('ce0bd0c2-54d9-4e29-85f1-cfb36ac3ebe2')
def test_get_rules_by_policy(self):
policy1 = self.create_qos_policy(name='test-policy1',
@@ -833,6 +883,147 @@
self.assertNotIn(rule2['id'], rules_ids)
+class QosMinimumBandwidthRuleTestJSON(base.BaseAdminNetworkTest):
+ DIRECTION_EGRESS = "egress"
+ DIRECTION_INGRESS = "ingress"
+ RULE_NAME = qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH + "_rule"
+ RULES_NAME = RULE_NAME + "s"
+
+ @classmethod
+ @test.requires_ext(extension="qos", service="network")
+ def resource_setup(cls):
+ super(QosMinimumBandwidthRuleTestJSON, cls).resource_setup()
+
+ @test.idempotent_id('aa59b00b-3e9c-4787-92f8-93a5cdf5e378')
+ def test_rule_create(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False)
+ rule = self.admin_client.create_minimum_bandwidth_rule(
+ policy_id=policy['id'], min_kbps=1138,
+ direction=self.DIRECTION_EGRESS)[self.RULE_NAME]
+
+ # Test 'show rule'
+ retrieved_rule = self.admin_client.show_minimum_bandwidth_rule(
+ policy['id'], rule['id'])
+ retrieved_rule = retrieved_rule[self.RULE_NAME]
+ self.assertEqual(rule['id'], retrieved_rule['id'])
+ self.assertEqual(1138, retrieved_rule['min_kbps'])
+ self.assertEqual(self.DIRECTION_EGRESS, retrieved_rule['direction'])
+
+ # Test 'list rules'
+ rules = self.admin_client.list_minimum_bandwidth_rules(policy['id'])
+ rules = rules[self.RULES_NAME]
+ rules_ids = [r['id'] for r in rules]
+ self.assertIn(rule['id'], rules_ids)
+
+ # Test 'show policy'
+ retrieved_policy = self.admin_client.show_qos_policy(policy['id'])
+ policy_rules = retrieved_policy['policy']['rules']
+ self.assertEqual(1, len(policy_rules))
+ self.assertEqual(rule['id'], policy_rules[0]['id'])
+ self.assertEqual(qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH,
+ policy_rules[0]['type'])
+
+ @test.idempotent_id('aa59b00b-ab01-4787-92f8-93a5cdf5e378')
+ def test_rule_create_fail_for_the_same_type(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False)
+ self.admin_client.create_minimum_bandwidth_rule(
+ policy_id=policy['id'], min_kbps=200,
+ direction=self.DIRECTION_EGRESS)
+
+ self.assertRaises(exceptions.Conflict,
+ self.admin_client.create_minimum_bandwidth_rule,
+ policy_id=policy['id'],
+ min_kbps=201, direction=self.DIRECTION_EGRESS)
+
+ @test.idempotent_id('d6fce764-e511-4fa6-9f86-f4b41cf142cf')
+ def test_rule_create_fail_for_direction_ingress(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False)
+ self.assertRaises(exceptions.BadRequest,
+ self.admin_client.create_minimum_bandwidth_rule,
+ policy_id=policy['id'],
+ min_kbps=201, direction=self.DIRECTION_INGRESS)
+
+ @test.idempotent_id('a49a6988-2568-47d2-931e-2dbc858943b3')
+ def test_rule_update(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False)
+ rule = self.admin_client.create_minimum_bandwidth_rule(
+ policy_id=policy['id'], min_kbps=300,
+ direction=self.DIRECTION_EGRESS)[self.RULE_NAME]
+
+ self.admin_client.update_minimum_bandwidth_rule(policy['id'],
+ rule['id'], min_kbps=350, direction=self.DIRECTION_EGRESS)
+
+ retrieved_policy = self.admin_client.show_minimum_bandwidth_rule(
+ policy['id'], rule['id'])
+ retrieved_policy = retrieved_policy[self.RULE_NAME]
+ self.assertEqual(350, retrieved_policy['min_kbps'])
+ self.assertEqual(self.DIRECTION_EGRESS, retrieved_policy['direction'])
+
+ @test.idempotent_id('a7ee6efd-7b33-4a68-927d-275b4f8ba958')
+ def test_rule_delete(self):
+ policy = self.create_qos_policy(name='test-policy',
+ description='test policy',
+ shared=False)
+ rule = self.admin_client.create_minimum_bandwidth_rule(
+ policy['id'], 200, self.DIRECTION_EGRESS)[self.RULE_NAME]
+
+ retrieved_policy = self.admin_client.show_minimum_bandwidth_rule(
+ policy['id'], rule['id'])
+ retrieved_policy = retrieved_policy[self.RULE_NAME]
+ self.assertEqual(rule['id'], retrieved_policy['id'])
+
+ self.admin_client.delete_minimum_bandwidth_rule(policy['id'],
+ rule['id'])
+ self.assertRaises(exceptions.NotFound,
+ self.admin_client.show_minimum_bandwidth_rule,
+ policy['id'], rule['id'])
+
+ @test.idempotent_id('a211222c-5808-46cb-a961-983bbab6b852')
+ def test_rule_create_rule_nonexistent_policy(self):
+ self.assertRaises(
+ exceptions.NotFound,
+ self.admin_client.create_minimum_bandwidth_rule,
+ 'policy', 200, self.DIRECTION_EGRESS)
+
+ @test.idempotent_id('b4a2e7ad-786f-4927-a85a-e545a93bd274')
+ def test_rule_create_forbidden_for_regular_tenants(self):
+ self.assertRaises(
+ exceptions.Forbidden,
+ self.client.create_minimum_bandwidth_rule,
+ 'policy', 300, self.DIRECTION_EGRESS)
+
+ @test.idempotent_id('de0bd0c2-54d9-4e29-85f1-cfb36ac3ebe2')
+ def test_get_rules_by_policy(self):
+ policy1 = self.create_qos_policy(name='test-policy1',
+ description='test policy1',
+ shared=False)
+ rule1 = self.admin_client.create_minimum_bandwidth_rule(
+ policy_id=policy1['id'], min_kbps=200,
+ direction=self.DIRECTION_EGRESS)[self.RULE_NAME]
+
+ policy2 = self.create_qos_policy(name='test-policy2',
+ description='test policy2',
+ shared=False)
+ rule2 = self.admin_client.create_minimum_bandwidth_rule(
+ policy_id=policy2['id'], min_kbps=5000,
+ direction=self.DIRECTION_EGRESS)[self.RULE_NAME]
+
+ # Test 'list rules'
+ rules = self.admin_client.list_minimum_bandwidth_rules(policy1['id'])
+ rules = rules[self.RULES_NAME]
+ rules_ids = [r['id'] for r in rules]
+ self.assertIn(rule1['id'], rules_ids)
+ self.assertNotIn(rule2['id'], rules_ids)
+
+
class QosSearchCriteriaTest(base.BaseSearchCriteriaTest,
base.BaseAdminNetworkTest):
diff --git a/neutron/tests/tempest/api/test_security_groups_negative.py b/neutron/tests/tempest/api/test_security_groups_negative.py
index 68b5fd0..43bc88c 100644
--- a/neutron/tests/tempest/api/test_security_groups_negative.py
+++ b/neutron/tests/tempest/api/test_security_groups_negative.py
@@ -17,9 +17,6 @@
from tempest import test
from neutron.tests.tempest.api import base_security_groups as base
-from neutron.tests.tempest import config
-
-CONF = config.CONF
class NegativeSecGroupTest(base.BaseSecGroupTest):
diff --git a/neutron/tests/tempest/api/test_trunk.py b/neutron/tests/tempest/api/test_trunk.py
index c22c334..44adb7e 100644
--- a/neutron/tests/tempest/api/test_trunk.py
+++ b/neutron/tests/tempest/api/test_trunk.py
@@ -58,46 +58,69 @@
trunks_cleanup(cls.client, cls.trunks)
super(TrunkTestJSONBase, cls).resource_cleanup()
+ def _remove_timestamps(self, trunk):
+ # Let's make sure these fields exist, but let's not
+ # use them in the comparison, in case skews or
+ # roundups get in the way.
+ created_at = trunk.pop('created_at')
+ updated_at = trunk.pop('updated_at')
+ self.assertIsNotNone(created_at)
+ self.assertIsNotNone(updated_at)
+
def _create_trunk_with_network_and_parent(self, subports, **kwargs):
network = self.create_network()
parent_port = self.create_port(network)
trunk = self.client.create_trunk(parent_port['id'], subports, **kwargs)
self.trunks.append(trunk['trunk'])
+ self._remove_timestamps(trunk['trunk'])
return trunk
+ def _show_trunk(self, trunk_id):
+ trunk = self.client.show_trunk(trunk_id)
+ self._remove_timestamps(trunk['trunk'])
+ return trunk
+
+ def _list_trunks(self):
+ trunks = self.client.list_trunks()
+ for t in trunks['trunks']:
+ self._remove_timestamps(t)
+ return trunks
+
class TrunkTestJSON(TrunkTestJSONBase):
+ def _test_create_trunk(self, subports):
+ trunk = self._create_trunk_with_network_and_parent(subports)
+ observed_trunk = self._show_trunk(trunk['trunk']['id'])
+ self.assertEqual(trunk, observed_trunk)
+
@test.idempotent_id('e1a6355c-4768-41f3-9bf8-0f1d192bd501')
def test_create_trunk_empty_subports_list(self):
- trunk = self._create_trunk_with_network_and_parent([])
- observed_trunk = self.client.show_trunk(trunk['trunk']['id'])
- self.assertEqual(trunk, observed_trunk)
+ self._test_create_trunk([])
@test.idempotent_id('382dfa39-ca03-4bd3-9a1c-91e36d2e3796')
def test_create_trunk_subports_not_specified(self):
- trunk = self._create_trunk_with_network_and_parent(None)
- observed_trunk = self.client.show_trunk(trunk['trunk']['id'])
- self.assertEqual(trunk, observed_trunk)
+ self._test_create_trunk(None)
@test.idempotent_id('7de46c22-e2b6-4959-ac5a-0e624632ab32')
def test_create_show_delete_trunk(self):
trunk = self._create_trunk_with_network_and_parent(None)
trunk_id = trunk['trunk']['id']
parent_port_id = trunk['trunk']['port_id']
- res = self.client.show_trunk(trunk_id)
+ res = self._show_trunk(trunk_id)
self.assertEqual(trunk_id, res['trunk']['id'])
self.assertEqual(parent_port_id, res['trunk']['port_id'])
self.client.delete_trunk(trunk_id)
- self.assertRaises(lib_exc.NotFound, self.client.show_trunk, trunk_id)
+ self.assertRaises(lib_exc.NotFound, self._show_trunk, trunk_id)
@test.idempotent_id('4ce46c22-a2b6-4659-bc5a-0ef2463cab32')
def test_create_update_trunk(self):
trunk = self._create_trunk_with_network_and_parent(None)
trunk_id = trunk['trunk']['id']
- res = self.client.show_trunk(trunk_id)
+ res = self._show_trunk(trunk_id)
self.assertTrue(res['trunk']['admin_state_up'])
self.assertEqual("", res['trunk']['name'])
+ self.assertEqual("", res['trunk']['description'])
res = self.client.update_trunk(
trunk_id, name='foo', admin_state_up=False)
self.assertFalse(res['trunk']['admin_state_up'])
@@ -105,13 +128,22 @@
# enable the trunk so that it can be managed
self.client.update_trunk(trunk_id, admin_state_up=True)
+ @test.idempotent_id('5ff46c22-a2b6-5559-bc5a-0ef2463cab32')
+ def test_create_update_trunk_with_description(self):
+ trunk = self._create_trunk_with_network_and_parent(
+ None, description="foo description")
+ trunk_id = trunk['trunk']['id']
+ self.assertEqual("foo description", trunk['trunk']['description'])
+ trunk = self.client.update_trunk(trunk_id, description='')
+ self.assertEqual('', trunk['trunk']['description'])
+
@test.idempotent_id('73365f73-bed6-42cd-960b-ec04e0c99d85')
def test_list_trunks(self):
trunk1 = self._create_trunk_with_network_and_parent(None)
trunk2 = self._create_trunk_with_network_and_parent(None)
expected_trunks = {trunk1['trunk']['id']: trunk1['trunk'],
trunk2['trunk']['id']: trunk2['trunk']}
- trunk_list = self.client.list_trunks()['trunks']
+ trunk_list = self._list_trunks()['trunks']
matched_trunks = [x for x in trunk_list if x['id'] in expected_trunks]
self.assertEqual(2, len(matched_trunks))
for trunk in matched_trunks:
@@ -126,7 +158,7 @@
'segmentation_type': 'vlan',
'segmentation_id': 2}]
self.client.add_subports(trunk['trunk']['id'], subports)
- trunk = self.client.show_trunk(trunk['trunk']['id'])
+ trunk = self._show_trunk(trunk['trunk']['id'])
observed_subports = trunk['trunk']['sub_ports']
self.assertEqual(1, len(observed_subports))
created_subport = observed_subports[0]
@@ -168,7 +200,7 @@
self.assertEqual(expected_subport, res['sub_ports'][0])
# Validate the results of a subport list
- trunk = self.client.show_trunk(trunk['trunk']['id'])
+ trunk = self._show_trunk(trunk['trunk']['id'])
observed_subports = trunk['trunk']['sub_ports']
self.assertEqual(1, len(observed_subports))
self.assertEqual(expected_subport, observed_subports[0])
diff --git a/neutron/tests/tempest/services/network/json/network_client.py b/neutron/tests/tempest/services/network/json/network_client.py
index 9c8cb05..00e9b76 100644
--- a/neutron/tests/tempest/services/network/json/network_client.py
+++ b/neutron/tests/tempest/services/network/json/network_client.py
@@ -55,6 +55,7 @@
'metering_label_rules': 'metering',
'policies': 'qos',
'bandwidth_limit_rules': 'qos',
+ 'minimum_bandwidth_rules': 'qos',
'rule_types': 'qos',
'rbac-policies': '',
}
@@ -652,6 +653,52 @@
self.expected_success(204, resp.status)
return service_client.ResponseBody(resp, body)
+ def create_minimum_bandwidth_rule(self, policy_id, min_kbps, direction):
+ uri = '%s/qos/policies/%s/minimum_bandwidth_rules' % (
+ self.uri_prefix, policy_id)
+ post_data = self.serialize({
+ 'minimum_bandwidth_rule': {
+ 'min_kbps': min_kbps,
+ 'direction': direction
+ }
+ })
+ resp, body = self.post(uri, post_data)
+ self.expected_success(201, resp.status)
+ body = jsonutils.loads(body)
+ return service_client.ResponseBody(resp, body)
+
+ def list_minimum_bandwidth_rules(self, policy_id):
+ uri = '%s/qos/policies/%s/minimum_bandwidth_rules' % (
+ self.uri_prefix, policy_id)
+ resp, body = self.get(uri)
+ body = self.deserialize_single(body)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBody(resp, body)
+
+ def show_minimum_bandwidth_rule(self, policy_id, rule_id):
+ uri = '%s/qos/policies/%s/minimum_bandwidth_rules/%s' % (
+ self.uri_prefix, policy_id, rule_id)
+ resp, body = self.get(uri)
+ body = self.deserialize_single(body)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBody(resp, body)
+
+ def update_minimum_bandwidth_rule(self, policy_id, rule_id, **kwargs):
+ uri = '%s/qos/policies/%s/minimum_bandwidth_rules/%s' % (
+ self.uri_prefix, policy_id, rule_id)
+ post_data = {'minimum_bandwidth_rule': kwargs}
+ resp, body = self.put(uri, jsonutils.dumps(post_data))
+ body = self.deserialize_single(body)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBody(resp, body)
+
+ def delete_minimum_bandwidth_rule(self, policy_id, rule_id):
+ uri = '%s/qos/policies/%s/minimum_bandwidth_rules/%s' % (
+ self.uri_prefix, policy_id, rule_id)
+ resp, body = self.delete(uri)
+ self.expected_success(204, resp.status)
+ return service_client.ResponseBody(resp, body)
+
def list_qos_rule_types(self):
uri = '%s/qos/rule-types' % self.uri_prefix
resp, body = self.get(uri)
@@ -660,7 +707,8 @@
return service_client.ResponseBody(resp, body)
def create_trunk(self, parent_port_id, subports,
- tenant_id=None, name=None, admin_state_up=None):
+ tenant_id=None, name=None, admin_state_up=None,
+ description=None):
uri = '%s/trunks' % self.uri_prefix
post_data = {
'trunk': {
@@ -673,6 +721,8 @@
post_data['trunk']['tenant_id'] = tenant_id
if name is not None:
post_data['trunk']['name'] = name
+ if description is not None:
+ post_data['trunk']['description'] = description
if admin_state_up is not None:
post_data['trunk']['admin_state_up'] = admin_state_up
resp, body = self.post(uri, self.serialize(post_data))
@@ -738,6 +788,12 @@
body = jsonutils.loads(body)
return service_client.ResponseBody(resp, body)
+ def delete_auto_allocated_topology(self, tenant_id=None):
+ uri = '%s/auto-allocated-topology/%s' % (self.uri_prefix, tenant_id)
+ resp, body = self.delete(uri)
+ self.expected_success(204, resp.status)
+ return service_client.ResponseBody(resp, body)
+
def create_security_group_rule(self, direction, security_group_id,
**kwargs):
post_body = {'security_group_rule': kwargs}