openstack
/
neutron
Code Issues Proposed changes

Remove the bundled intree neutron tempest plugin

Browse Source 
The plugin has been split into its own repository[1] in accordance with
Queens Goal "Split Tempest Plugins into Separate Repos/Projects[2]".
This patch removes the local copy as well as the setuptools entry point.
We can also now remove the autodoc_tree_excludes pbr option since
there's no more plugin to exclude and it defaults to [setup.py].

The patch leaves still in-tree base classes and constants in order to
avoid breakage of projects depending on such code. In the future, we
will deprecate those and gradually move affected subprojects to the new
repo.

List of affected repositories:
 - neutron-vpnaas
 - openstack-ansible-os_neutron
 - tripleo-quickstart-extras
 - networking-midonet
 - networking-l2gw
 - neutron-dynamic-routing
 - networking-vsphere
 - networking-plumgrid
 - networking-zvm

[1] http://git.openstack.org/cgit/openstack/neutron-tempest-plugin
[2] https://governance.openstack.org/tc/goals/queens/split-tempest-plugins.html

Depends-On: I371aa4d5f043f695df04b98b0f485c8f0548f2b3

Change-Id: Ia21aad29d0bbf779583964db6f1665c9b3b83161
changes/72/506672/5
Chandan Kumar 5 years ago committed by Ihar Hrachyshka
parent 618ee8658f
commit 813295f8cf
66 changed files with 0 additions and 10001 deletions
Show Stats Download Patch File Download Diff File

9
neutron/tests/tempest/README.rst
Unescape View File

@ -1,9 +0,0 @@
WARNING
=======
Some files under this path were copied from tempest as part of the move of the
api tests, and they will be removed as required over time to minimize the
dependency on the tempest testing framework. While it exists, only
neutron.tests.tempest.* should be importing files from this path.
neutron.tests.tempest.config uses the global cfg.CONF instance and importing it
outside of the api tests has the potential to break Neutron's use of cfg.CONF.

0
neutron/tests/tempest/api/admin/__init__.py
Unescape View File

53
neutron/tests/tempest/api/admin/test_agent_availability_zone.py
Unescape View File

@ -1,53 +0,0 @@
# 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 neutron_lib import constants
from tempest.lib import decorators
import testtools
from neutron.tests.tempest.api import base
from neutron.tests.tempest import config
AZ_SUPPORTED_AGENTS = [constants.AGENT_TYPE_DHCP, constants.AGENT_TYPE_L3]
CONF = config.CONF
class AgentAvailabilityZoneTestCase(base.BaseAdminNetworkTest):
required_extensions = ['agent', 'availability_zone']
@classmethod
def resource_setup(cls):
super(AgentAvailabilityZoneTestCase, cls).resource_setup()
body = cls.admin_client.list_agents()
agents = body['agents']
agents_type = [agent.get('agent_type') for agent in agents]
for az_agent in AZ_SUPPORTED_AGENTS:
if az_agent in agents_type:
return
msg = 'availability_zone supported agent not found.'
raise cls.skipException(msg)
@decorators.idempotent_id('3ffa661e-cfcc-417d-8b63-1c5ec4a22e54')
@testtools.skipUnless(CONF.neutron_plugin_options.agent_availability_zone,
"Need a single availability_zone assumption.")
def test_agents_availability_zone(self):
"""
Test list agents availability_zone, only L3 and DHCP agent support
availability_zone, default availability_zone is "nova".
"""
body = self.admin_client.list_agents()
agents = body['agents']
for agent in agents:
if agent.get('agent_type') in AZ_SUPPORTED_AGENTS:
self.assertEqual(
CONF.neutron_plugin_options.agent_availability_zone,
agent.get('availability_zone'))

90
neutron/tests/tempest/api/admin/test_agent_management.py
Unescape View File

@ -1,90 +0,0 @@
# 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 neutron.tests.tempest.common import tempest_fixtures
from tempest.lib import decorators
from neutron.tests.tempest.api import base
class AgentManagementTestJSON(base.BaseAdminNetworkTest):
required_extensions = ['agent']
@classmethod
def resource_setup(cls):
super(AgentManagementTestJSON, cls).resource_setup()
body = cls.admin_client.list_agents()
agents = body['agents']
cls.agent = agents[0] # don't modify this agent
@decorators.idempotent_id('9c80f04d-11f3-44a4-8738-ed2f879b0ff4')
def test_list_agent(self):
body = self.admin_client.list_agents()
agents = body['agents']
# Heartbeats must be excluded from comparison
self.agent.pop('heartbeat_timestamp', None)
self.agent.pop('configurations', None)
for agent in agents:
agent.pop('heartbeat_timestamp', None)
agent.pop('configurations', None)
self.assertIn(self.agent, agents)
@decorators.idempotent_id('e335be47-b9a1-46fd-be30-0874c0b751e6')
def test_list_agents_non_admin(self):
body = self.client.list_agents()
self.assertEqual(len(body["agents"]), 0)
@decorators.idempotent_id('869bc8e8-0fda-4a30-9b71-f8a7cf58ca9f')
def test_show_agent(self):
body = self.admin_client.show_agent(self.agent['id'])
agent = body['agent']
self.assertEqual(agent['id'], self.agent['id'])
@decorators.idempotent_id('371dfc5b-55b9-4cb5-ac82-c40eadaac941')
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}
body = self.admin_client.update_agent(agent_id=self.agent['id'],
agent_info=agent_status)
updated_status = body['agent']['admin_state_up']
self.assertEqual(origin_status, updated_status)
@decorators.idempotent_id('68a94a14-1243-46e6-83bf-157627e31556')
def test_update_agent_description(self):
agents = self.admin_client.list_agents()['agents']
try:
dyn_agent = agents[1]
except IndexError:
raise self.skipException("This test requires at least two agents.")
self.useFixture(tempest_fixtures.LockFixture('agent_description'))
description = 'description for update agent.'
agent_description = {'description': description}
body = self.admin_client.update_agent(agent_id=dyn_agent['id'],
agent_info=agent_description)
self.addCleanup(self._restore_agent, dyn_agent)
updated_description = body['agent']['description']
self.assertEqual(updated_description, description)
def _restore_agent(self, dyn_agent):
"""
Restore the agent description after update test.
"""
description = dyn_agent['description']
origin_agent = {'description': description}
self.admin_client.update_agent(agent_id=dyn_agent['id'],
agent_info=origin_agent)

108
neutron/tests/tempest/api/admin/test_dhcp_agent_scheduler.py
Unescape View File

@ -1,108 +0,0 @@
# 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 neutron_lib import constants
from tempest.lib import decorators
from neutron.common import utils
from neutron.tests.tempest.api import base
class DHCPAgentSchedulersTestJSON(base.BaseAdminNetworkTest):
required_extensions = ['dhcp_agent_scheduler']
@classmethod
def resource_setup(cls):
super(DHCPAgentSchedulersTestJSON, cls).resource_setup()
# Create a network and make sure it will be hosted by a
# dhcp agent: this is done by creating a regular port
cls.network = cls.create_network()
cls.subnet = cls.create_subnet(cls.network)
cls.cidr = cls.subnet['cidr']
cls.port = cls.create_port(cls.network)
@decorators.idempotent_id('f164801e-1dd8-4b8b-b5d3-cc3ac77cfaa5')
def test_dhcp_port_status_active(self):
def dhcp_port_active():
for p in self.client.list_ports(
network_id=self.network['id'])['ports']:
if (p['device_owner'] == constants.DEVICE_OWNER_DHCP and
p['status'] == constants.PORT_STATUS_ACTIVE):
return True
return False
utils.wait_until_true(dhcp_port_active)
@decorators.idempotent_id('5032b1fe-eb42-4a64-8f3b-6e189d8b5c7d')
def test_list_dhcp_agent_hosting_network(self):
self.admin_client.list_dhcp_agent_hosting_network(
self.network['id'])
@decorators.idempotent_id('30c48f98-e45d-4ffb-841c-b8aad57c7587')
def test_list_networks_hosted_by_one_dhcp(self):
body = self.admin_client.list_dhcp_agent_hosting_network(
self.network['id'])
agents = body['agents']
self.assertIsNotNone(agents)
agent = agents[0]
self.assertTrue(self._check_network_in_dhcp_agent(
self.network['id'], agent))
def _check_network_in_dhcp_agent(self, network_id, agent):
network_ids = []
body = self.admin_client.list_networks_hosted_by_one_dhcp_agent(
agent['id'])
networks = body['networks']
for network in networks:
network_ids.append(network['id'])
return network_id in network_ids
@decorators.idempotent_id('a0856713-6549-470c-a656-e97c8df9a14d')
def test_add_remove_network_from_dhcp_agent(self):
# The agent is now bound to the network, we can free the port
self.client.delete_port(self.port['id'])
self.ports.remove(self.port)
agent = dict()
agent['agent_type'] = None
body = self.admin_client.list_agents()
agents = body['agents']
for a in agents:
if a['agent_type'] == 'DHCP agent':
agent = a
break
self.assertEqual(agent['agent_type'], 'DHCP agent', 'Could not find '
'DHCP agent in agent list though dhcp_agent_scheduler'
' is enabled.')
network = self.create_network()
network_id = network['id']
if self._check_network_in_dhcp_agent(network_id, agent):
self._remove_network_from_dhcp_agent(network_id, agent)
self._add_dhcp_agent_to_network(network_id, agent)
else:
self._add_dhcp_agent_to_network(network_id, agent)
self._remove_network_from_dhcp_agent(network_id, agent)
def _remove_network_from_dhcp_agent(self, network_id, agent):
self.admin_client.remove_network_from_dhcp_agent(
agent_id=agent['id'],
network_id=network_id)
self.assertFalse(self._check_network_in_dhcp_agent(
network_id, agent))
def _add_dhcp_agent_to_network(self, network_id, agent):
self.admin_client.add_dhcp_agent_to_network(agent['id'],
network_id)
self.assertTrue(self._check_network_in_dhcp_agent(
network_id, agent))

35
neutron/tests/tempest/api/admin/test_extension_driver_port_security_admin.py
Unescape View File

@ -1,35 +0,0 @@
# Copyright 2015 Cisco Systems, Inc.
# 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.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron.tests.tempest.api import base
from neutron.tests.tempest.api import base_security_groups as base_security
class PortSecurityAdminTests(base_security.BaseSecGroupTest,
base.BaseAdminNetworkTest):
required_extensions = ['port-security']
@decorators.attr(type='negative')
@decorators.idempotent_id('d39a96e2-2dea-4feb-8093-e7ac991ce6f8')
def test_create_port_security_false_on_shared_network(self):
network = self.create_shared_network()
self.assertTrue(network['shared'])
self.create_subnet(network, client=self.admin_client)
self.assertRaises(lib_exc.Forbidden, self.create_port,
network, port_security_enabled=False)

196
neutron/tests/tempest/api/admin/test_external_network_extension.py
Unescape View File

@ -1,196 +0,0 @@
# 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 oslo_config import cfg
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
import testtools
from neutron.tests.tempest.api import base
class ExternalNetworksRBACTestJSON(base.BaseAdminNetworkTest):
credentials = ['primary', 'alt', 'admin']
required_extensions = ['rbac-policies']
@classmethod
def resource_setup(cls):
super(ExternalNetworksRBACTestJSON, cls).resource_setup()
cls.client2 = cls.os_alt.network_client
def _create_network(self, external=True):
post_body = {'name': data_utils.rand_name('network')}
if external:
post_body['router:external'] = external
body = self.admin_client.create_network(**post_body)
network = body['network']
self.addCleanup(self.admin_client.delete_network, network['id'])
return network
@decorators.idempotent_id('afd8f1b7-a81e-4629-bca8-a367b3a144bb')
def test_regular_client_shares_with_another(self):
net = self.create_network()
self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant=self.client2.tenant_id)
body = self.client2.list_networks()
networks_list = [n['id'] for n in body['networks']]
self.assertIn(net['id'], networks_list)
r = self.client2.create_router(
data_utils.rand_name('router'),
external_gateway_info={'network_id': net['id']})['router']
self.addCleanup(self.admin_client.delete_router, r['id'])
@decorators.idempotent_id('eff9443a-2d04-48ee-840e-d955ac564bcd')
def test_regular_client_blocked_from_creating_external_wild_policies(self):
net = self.create_network()
with testtools.ExpectedException(lib_exc.Forbidden):
self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant='*')
@decorators.idempotent_id('a2e19f06-48a9-4e4c-b717-08cb2008707d')
def test_wildcard_policy_created_from_external_network_api(self):
# create external makes wildcard
net_id = self._create_network(external=True)['id']
self.assertEqual(1, len(self.admin_client.list_rbac_policies(
object_id=net_id, action='access_as_external',
target_tenant='*')['rbac_policies']))
# update to non-external clears wildcard
self.admin_client.update_network(net_id, **{'router:external': False})
self.assertEqual(0, len(self.admin_client.list_rbac_policies(
object_id=net_id, action='access_as_external',
target_tenant='*')['rbac_policies']))
# create non-external has no wildcard
net_id = self._create_network(external=False)['id']
self.assertEqual(0, len(self.admin_client.list_rbac_policies(
object_id=net_id, action='access_as_external',
target_tenant='*')['rbac_policies']))
# update to external makes wildcard
self.admin_client.update_network(net_id, **{'router:external': True})
self.assertEqual(1, len(self.admin_client.list_rbac_policies(
object_id=net_id, action='access_as_external',
target_tenant='*')['rbac_policies']))
@decorators.idempotent_id('a5539002-5bdb-48b5-b124-abcd12347865')
def test_external_update_policy_from_wildcard_to_specific_tenant(self):
net_id = self._create_network(external=True)['id']
rbac_pol = self.admin_client.list_rbac_policies(
object_id=net_id, action='access_as_external',
target_tenant='*')['rbac_policies'][0]
r = self.client2.create_router(
data_utils.rand_name('router'),
external_gateway_info={'network_id': net_id})['router']
self.addCleanup(self.admin_client.delete_router, r['id'])
# changing wildcard to specific tenant should be okay since its the
# only one using the network
self.admin_client.update_rbac_policy(
rbac_pol['id'], target_tenant=self.client2.tenant_id)
@decorators.idempotent_id('a5539002-5bdb-48b5-b124-e9eedd5975e6')
def test_external_conversion_on_policy_create(self):
net_id = self._create_network(external=False)['id']
self.admin_client.create_rbac_policy(
object_type='network', object_id=net_id,
action='access_as_external',
target_tenant=self.client2.tenant_id)
body = self.admin_client.show_network(net_id)['network']
self.assertTrue(body['router:external'])
@decorators.idempotent_id('01364c50-bfb6-46c4-b44c-edc4564d61cf')
def test_policy_allows_tenant_to_allocate_floatingip(self):
net = self._create_network(external=False)
# share to the admin client so it gets converted to external but
# not shared to everyone
self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant=self.admin_client.tenant_id)
self.create_subnet(net, client=self.admin_client, enable_dhcp=False)
with testtools.ExpectedException(lib_exc.NotFound):
self.client2.create_floatingip(
floating_network_id=net['id'])
self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant=self.client2.tenant_id)
self.client2.create_floatingip(
floating_network_id=net['id'])
@decorators.idempotent_id('476be1e0-f72e-47dc-9a14-4435926bbe82')
def test_policy_allows_tenant_to_attach_ext_gw(self):
net = self._create_network(external=False)
self.create_subnet(net, client=self.admin_client, enable_dhcp=False)
self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant=self.client2.tenant_id)
r = self.client2.create_router(
data_utils.rand_name('router'),
external_gateway_info={'network_id': net['id']})['router']
self.addCleanup(self.admin_client.delete_router, r['id'])
@decorators.idempotent_id('d54decee-4203-4ced-91a2-ea42ca63e154')
def test_delete_policies_while_tenant_attached_to_net(self):
net = self._create_network(external=False)
self.create_subnet(net, client=self.admin_client, enable_dhcp=False)
wildcard = self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant='*')['rbac_policy']
r = self.client2.create_router(
data_utils.rand_name('router'),
external_gateway_info={'network_id': net['id']})['router']
# delete should fail because the wildcard is required for the tenant's
# access
with testtools.ExpectedException(lib_exc.Conflict):
self.admin_client.delete_rbac_policy(wildcard['id'])
tenant = self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant=self.client2.tenant_id)['rbac_policy']
# now we can delete the policy because the tenant has its own policy
# to allow it access
self.admin_client.delete_rbac_policy(wildcard['id'])
# but now we can't delete the tenant's policy without the wildcard
with testtools.ExpectedException(lib_exc.Conflict):
self.admin_client.delete_rbac_policy(tenant['id'])
wildcard = self.admin_client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_external',
target_tenant='*')['rbac_policy']
# with the wildcard added back we can delete the tenant's policy
self.admin_client.delete_rbac_policy(tenant['id'])
self.admin_client.delete_router(r['id'])
# now without the tenant attached, the wildcard can be deleted
self.admin_client.delete_rbac_policy(wildcard['id'])
# finally we ensure that the tenant can't attach to the network since
# there are no policies allowing it
with testtools.ExpectedException(lib_exc.NotFound):
self.client2.create_router(
data_utils.rand_name('router'),
external_gateway_info={'network_id': net['id']})
@decorators.idempotent_id('7041cec7-d8fe-4c78-9b04-b51b2fd49dc9')
def test_wildcard_policy_delete_blocked_on_default_ext(self):
public_net_id = cfg.CONF.network.public_network_id
# ensure it is default before so we don't wipe out the policy
self.admin_client.update_network(public_net_id, is_default=True)
policy = self.admin_client.list_rbac_policies(
object_id=public_net_id, action='access_as_external',
target_tenant='*')['rbac_policies'][0]
with testtools.ExpectedException(lib_exc.Conflict):
self.admin_client.delete_rbac_policy(policy['id'])

83
neutron/tests/tempest/api/admin/test_floating_ips_admin_actions.py
Unescape View File

@ -1,83 +0,0 @@
# Copyright 2014 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.
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
import testtools
from neutron.tests.tempest.api import base
from neutron.tests.tempest import config
CONF = config.CONF
class FloatingIPAdminTestJSON(base.BaseAdminNetworkTest):
force_tenant_isolation = True
credentials = ['primary', 'alt', 'admin']
@classmethod
def resource_setup(cls):
super(FloatingIPAdminTestJSON, cls).resource_setup()
cls.ext_net_id = CONF.network.public_network_id
cls.floating_ip = cls.create_floatingip(cls.ext_net_id)
cls.alt_client = cls.os_alt.network_client
cls.network = cls.create_network()
cls.subnet = cls.create_subnet(cls.network)
cls.router = cls.create_router(data_utils.rand_name('router'),
external_network_id=cls.ext_net_id)
cls.create_router_interface(cls.router['id'], cls.subnet['id'])
cls.port = cls.create_port(cls.network)
@decorators.attr(type='negative')
@decorators.idempotent_id('11116ee9-4e99-5b15-b8e1-aa7df92ca589')
def test_associate_floating_ip_with_port_from_another_project(self):
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id)
floating_ip = body['floatingip']
project_id = self.create_project()['id']
port = self.admin_client.create_port(network_id=self.network['id'],
project_id=project_id)
self.addCleanup(self.admin_client.delete_port, port['port']['id'])
self.assertRaises(lib_exc.BadRequest,
self.client.update_floatingip,
floating_ip['id'], port_id=port['port']['id'])
@testtools.skipUnless(
CONF.neutron_plugin_options.specify_floating_ip_address_available,
"Feature for specifying floating IP address is disabled")
@decorators.idempotent_id('332a8ae4-402e-4b98-bb6f-532e5a87b8e0')
def test_create_floatingip_with_specified_ip_address(self):
# other tests may end up stealing the IP before we can use it
# since it's on the external network so we need to retry if it's
# in use.
for i in range(100):
fip = self.get_unused_ip(self.ext_net_id, ip_version=4)
try:
body = self.admin_client.create_floatingip(
floating_network_id=self.ext_net_id,
floating_ip_address=fip)
break
except lib_exc.Conflict:
pass
else:
self.fail("Could not get an unused IP after 100 attempts")
created_floating_ip = body['floatingip']
self.addCleanup(self.admin_client.delete_floatingip,
created_floating_ip['id'])
self.assertIsNotNone(created_floating_ip['id'])
self.assertIsNotNone(created_floating_ip['tenant_id'])
self.assertEqual(created_floating_ip['floating_ip_address'], fip)

85
neutron/tests/tempest/api/admin/test_l3_agent_scheduler.py
Unescape View File

@ -1,85 +0,0 @@
# 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.lib.common.utils import data_utils
from tempest.lib import decorators
from neutron.tests.tempest.api import base
from neutron.tests.tempest import exceptions
AGENT_TYPE = 'L3 agent'
AGENT_MODES = (
'legacy',
'dvr_snat'
)
class L3AgentSchedulerTestJSON(base.BaseAdminNetworkTest):
_agent_mode = 'legacy'
"""
Tests the following operations in the Neutron API using the REST client for
Neutron:
List routers that the given L3 agent is hosting.
List L3 agents hosting the given router.
Add and Remove Router to L3 agent
v2.0 of the Neutron API is assumed.
The l3_agent_scheduler extension is required for these tests.
"""
required_extensions = ['l3_agent_scheduler']
@classmethod
def resource_setup(cls):
super(L3AgentSchedulerTestJSON, cls).resource_setup()
body = cls.admin_client.list_agents()
agents = body['agents']
for agent in agents:
# TODO(armax): falling back on default _agent_mode can be
# dropped as soon as Icehouse is dropped.
agent_mode = (
agent['configurations'].get('agent_mode', cls._agent_mode))
if agent['agent_type'] == AGENT_TYPE and agent_mode in AGENT_MODES:
cls.agent = agent
break
else:
msg = "L3 Agent Scheduler enabled in conf, but L3 Agent not found"
raise exceptions.InvalidConfiguration(msg)
cls.router = cls.create_router(data_utils.rand_name('router'))
@decorators.idempotent_id('b7ce6e89-e837-4ded-9b78-9ed3c9c6a45a')
def test_list_routers_on_l3_agent(self):
self.admin_client.list_routers_on_l3_agent(self.agent['id'])
@decorators.idempotent_id('9464e5e7-8625-49c3-8fd1-89c52be59d66')
def test_add_list_remove_router_on_l3_agent(self):
l3_agent_ids = list()
self.admin_client.add_router_to_l3_agent(
self.agent['id'],
self.router['id'])
body = (
self.admin_client.list_l3_agents_hosting_router(self.router['id']))
for agent in body['agents']:
l3_agent_ids.append(agent['id'])
self.assertIn('agent_type', agent)
self.assertEqual('L3 agent', agent['agent_type'])
self.assertIn(self.agent['id'], l3_agent_ids)
body = self.admin_client.remove_router_from_l3_agent(
self.agent['id'],
self.router['id'])
# NOTE(afazekas): The deletion not asserted, because neutron
# is not forbidden to reschedule the router to the same agent

87
neutron/tests/tempest/api/admin/test_networks.py
Unescape View File

@ -1,87 +0,0 @@
# 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 oslo_utils import uuidutils
from tempest.common import utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron.tests.tempest.api import base
from neutron.tests.tempest import config
class NetworksTestAdmin(base.BaseAdminNetworkTest):
@decorators.idempotent_id('d3c76044-d067-4cb0-ae47-8cdd875c7f67')
@utils.requires_ext(extension="project-id", service="network")
def test_admin_create_network_keystone_v3(self):
project_id = self.client.tenant_id # non-admin
name = 'admin-created-with-project_id'
new_net = self.create_network_keystone_v3(name, project_id,
client=self.admin_client)
self.assertEqual(name, new_net['name'])
self.assertEqual(project_id, new_net['project_id'])
self.assertEqual(project_id, new_net['tenant_id'])
body = self.client.list_networks(id=new_net['id'])
lookup_net = body['networks'][0]
self.assertEqual(name, lookup_net['name'])
self.assertEqual(project_id, lookup_net['project_id'])
self.assertEqual(project_id, lookup_net['tenant_id'])
@decorators.idempotent_id('8d21aaca-4364-4eb9-8b79-44b4fff6373b')
@utils.requires_ext(extension="project-id", service="network")
def test_admin_create_network_keystone_v3_and_tenant(self):
project_id = self.client.tenant_id # non-admin
name = 'created-with-project-and-tenant'
new_net = self.create_network_keystone_v3(
name, project_id, tenant_id=project_id, client=self.admin_client)
self.assertEqual(name, new_net['name'])
self.assertEqual(project_id, new_net['project_id'])
self.assertEqual(project_id, new_net['tenant_id'])
body = self.client.list_networks(id=new_net['id'])
lookup_net = body['networks'][0]
self.assertEqual(name, lookup_net['name'])
self.assertEqual(project_id, lookup_net['project_id'])
self.assertEqual(project_id, lookup_net['tenant_id'])
@decorators.idempotent_id('08b92179-669d-45ee-8233-ef6611190809')
@utils.requires_ext(extension="project-id", service="network")
def test_admin_create_network_keystone_v3_and_other_tenant(self):
project_id = self.client.tenant_id # non-admin
other_tenant = uuidutils.generate_uuid()
name = 'created-with-project-and-other-tenant'
e = self.assertRaises(lib_exc.BadRequest,
self.create_network_keystone_v3, name,
project_id, tenant_id=other_tenant,
client=self.admin_client)
expected_message = "'project_id' and 'tenant_id' do not match"
self.assertEqual(expected_message, e.resp_body['message'])
@decorators.idempotent_id('571d0dde-0f84-11e7-b565-fa163e4fa634')
@testtools.skipUnless("vxlan" in config.CONF.neutron_plugin_options.
available_type_drivers,
'VXLAN type_driver is not enabled')
@utils.requires_ext(extension="provider", service="network")
def test_create_tenant_network_vxlan(self):
network = self.admin_client.create_network(
**{"provider:network_type": "vxlan"})['network']
self.addCleanup(self.admin_client.delete_network,
network['id'])
network = self.admin_client.show_network(
network['id'])['network']
self.assertEqual('vxlan', network['provider:network_type'])

150
neutron/tests/tempest/api/admin/test_quotas.py
Unescape View File

@ -1,150 +0,0 @@
# Copyright 2013 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 six
from tempest.common import utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron.tests.tempest.api import base
from neutron.tests.tempest import config
CONF = config.CONF
class QuotasTestBase(base.BaseAdminNetworkTest):
required_extensions = ['quotas']
@classmethod
def resource_setup(cls):
super(QuotasTestBase, cls).resource_setup()
def _setup_quotas(self, project_id, **new_quotas):
# Change quotas for tenant
quota_set = self.admin_client.update_quotas(project_id,
**new_quotas)
self.addCleanup(self._cleanup_quotas, project_id)
return quota_set
def _cleanup_quotas(self, project_id):
# Try to clean up the resources. If it fails, then
# assume that everything was already deleted, so
# it is OK to continue.
try:
self.admin_client.reset_quotas(project_id)
except lib_exc.NotFound:
pass
def _create_network(self, project_id):
network = self.create_network(client=self.admin_client,
tenant_id=project_id)
self.addCleanup(self.admin_client.delete_network,
network['id'])
return network
def _create_port(self, **kwargs):
port = self.admin_client.create_port(**kwargs)['port']
self.addCleanup(self.admin_client.delete_port,
port['id'])
return port
class QuotasTest(QuotasTestBase):
"""Test the Neutron API of Quotas.
Tests the following operations in the Neutron API using the REST client for
Neutron:
list quotas for tenants who have non-default quota values
show quotas for a specified tenant
show detail quotas for a specified tenant
update quotas for a specified tenant
reset quotas to default values for a specified tenant
v2.0 of the API is assumed.
It is also assumed that the per-tenant quota extension API is configured
in /etc/neutron/neutron.conf as follows:
quota_driver = neutron.db.driver.DbQuotaDriver
"""
@decorators.attr(type='gate')
@decorators.idempotent_id('2390f766-836d-40ef-9aeb-e810d78207fb')
def test_quotas(self):
tenant_id = self.create_project()['id']
new_quotas = {'network': 0, 'security_group': 0}
# Change quotas for tenant
quota_set = self._setup_quotas(tenant_id, **new_quotas)
for key, value in new_quotas.items():
self.assertEqual(value, quota_set[key])
# Confirm our tenant is listed among tenants with non default quotas
non_default_quotas = self.admin_client.list_quotas()
found = False
for qs in non_default_quotas['quotas']:
if qs['tenant_id'] == tenant_id:
self.assertEqual(tenant_id, qs['project_id'])
found = True
self.assertTrue(found)
# Confirm from API quotas were changed as requested for tenant
quota_set = self.admin_client.show_quotas(tenant_id)
quota_set = quota_set['quota']
for key, value in new_quotas.items():
self.assertEqual(value, quota_set[key])
# Reset quotas to default and confirm
self.admin_client.reset_quotas(tenant_id)
non_default_quotas = self.admin_client.list_quotas()
for q in non_default_quotas['quotas']:
self.assertNotEqual(tenant_id, q['tenant_id'])
@decorators.idempotent_id('e974b5ba-090a-452c-a578-f9710151d9fc')
@decorators.attr(type='gate')
@utils.requires_ext(extension="quota_details", service="network")
def test_detail_quotas(self):
tenant_id = self.create_project()['id']
new_quotas = {'network': {'used': 1, 'limit': 2, 'reserved': 0},
'port': {'used': 1, 'limit': 2, 'reserved': 0}}
# update quota limit for tenant
new_quota = {'network': new_quotas['network']['limit'], 'port':
new_quotas['port']['limit']}
quota_set = self._setup_quotas(tenant_id, **new_quota)
# create test resources
network = self._create_network(tenant_id)
post_body = {"network_id": network['id'],
"tenant_id": tenant_id}
self._create_port(**post_body)
# confirm from extended API quotas were changed
# as requested for tenant
quota_set = self.admin_client.show_details_quota(tenant_id)
quota_set = quota_set['quota']
for key, value in six.iteritems(new_quotas):
self.assertEqual(new_quotas[key]['limit'],
quota_set[key]['limit'])
self.assertEqual(new_quotas[key]['reserved'],
quota_set[key]['reserved'])
self.assertEqual(new_quotas[key]['used'],
quota_set[key]['used'])
# validate 'default' action for old extension
quota_limit = self.admin_client.show_quotas(tenant_id)['quota']
for key, value in six.iteritems(new_quotas):
self.assertEqual(new_quotas[key]['limit'], quota_limit[key])

175
neutron/tests/tempest/api/admin/test_quotas_negative.py
Unescape View File

@ -1,175 +0,0 @@
# 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.common import utils
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron.tests.tempest.api.admin import test_quotas
from neutron.tests.tempest import config
CONF = config.CONF
class QuotasAdminNegativeTestJSON(test_quotas.QuotasTestBase):
@decorators.attr(type='negative')
@decorators.idempotent_id('952f9b24-9156-4bdc-90f3-682a3d4302f0')
def test_create_network_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
new_quotas = {'network': 1}
self._setup_quotas(tenant_id, **new_quotas)
net_args = {'tenant_id': tenant_id}
net = self.admin_client.create_network(**net_args)['network']
self.addCleanup(self.admin_client.delete_network, net['id'])
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_network, **net_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('0b7f99e3-9f77-45ce-9a89-b39a184de618')
def test_create_subnet_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
new_quotas = {'subnet': 1}
self._setup_quotas(tenant_id, **new_quotas)
net_args = {'tenant_id': tenant_id}
net = self.admin_client.create_network(**net_args)['network']
self.addCleanup(self.admin_client.delete_network, net['id'])
subnet_args = {'tenant_id': tenant_id,
'network_id': net['id'],
'cidr': '10.0.0.0/24',
'ip_version': '4'}
subnet = self.admin_client.create_subnet(**subnet_args)['subnet']
self.addCleanup(self.admin_client.delete_subnet, subnet['id'])
subnet_args['cidr'] = '10.1.0.0/24'
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_subnet, **subnet_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('fe20d9f9-346c-4a20-bbfa-d9ca390f4dc6')
def test_create_port_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
new_quotas = {'port': 1}
self._setup_quotas(tenant_id, **new_quotas)
net_args = {'tenant_id': tenant_id}
net = self.admin_client.create_network(**net_args)['network']
self.addCleanup(self.admin_client.delete_network, net['id'])
subnet_args = {'tenant_id': tenant_id,
'network_id': net['id'],
'enable_dhcp': False,
'cidr': '10.0.0.0/24',
'ip_version': '4'}
subnet = self.admin_client.create_subnet(**subnet_args)['subnet']
self.addCleanup(self.admin_client.delete_subnet, subnet['id'])
port_args = {'tenant_id': tenant_id,
'network_id': net['id']}
port = self.admin_client.create_port(**port_args)['port']
self.addCleanup(self.admin_client.delete_port, port['id'])
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_port, **port_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('bb1e9c3c-7e6f-41f1-b579-63dbc655ecb7')
@utils.requires_ext(extension="router", service="network")
def test_create_router_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
new_quotas = {'router': 1}
self._setup_quotas(tenant_id, **new_quotas)
name = data_utils.rand_name('test_router')
router_args = {'tenant_id': tenant_id}
router = self.admin_client.create_router(
name, True, **router_args)['router']
self.addCleanup(self.admin_client.delete_router, router['id'])
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_router,
name, True, **router_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('5c924ff7-b7a9-474f-92a3-dbe0f976ec13')
@utils.requires_ext(extension="security-group", service="network")
def test_create_security_group_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
sg_args = {'tenant_id': tenant_id}
# avoid a number that is made by default
sg_list = self.admin_client.list_security_groups(
tenant_id=tenant_id)['security_groups']
num = len(sg_list) + 1
new_quotas = {'security_group': num}
self._setup_quotas(tenant_id, **new_quotas)
sg = self.admin_client.create_security_group(
**sg_args)['security_group']
self.addCleanup(self.admin_client.delete_security_group, sg['id'])
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_security_group, **sg_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('b7143480-6118-4ed4-be38-1b6f15f30d05')
@utils.requires_ext(extension="security-group", service="network")
def test_create_security_group_rule_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
sg_args = {'tenant_id': tenant_id}
sg = self.admin_client.create_security_group(
**sg_args)['security_group']
self.addCleanup(self.admin_client.delete_security_group, sg['id'])
# avoid a number that is made by default
sg_rule_list = self.admin_client.list_security_group_rules(
tenant_id=tenant_id)['security_group_rules']
num = len(sg_rule_list) + 1
new_quotas = {'security_group_rule': num}
self._setup_quotas(tenant_id, **new_quotas)
sg_rule_args = {'tenant_id': tenant_id,
'security_group_id': sg['id'],
'direction': 'ingress'}
sg_rule = self.admin_client.create_security_group_rule(
**sg_rule_args)['security_group_rule']
self.addCleanup(
self.admin_client.delete_security_group_rule, sg_rule['id'])
sg_rule_args['direction'] = 'egress'
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_security_group_rule,
**sg_rule_args)
@decorators.attr(type='negative')
@decorators.idempotent_id('d00fe5bb-9db8-4e1a-9c31-490f52897e6f')
@utils.requires_ext(extension="router", service="network")
def test_create_floatingip_when_quotas_is_full(self):
tenant_id = self.create_project()['id']
new_quotas = {'floatingip': 1}
self._setup_quotas(tenant_id, **new_quotas)
ext_net_id = CONF.network.public_network_id
fip_args = {'tenant_id': tenant_id,
'floating_network_id': ext_net_id}
fip = self.admin_client.create_floatingip(**fip_args)['floatingip']
self.addCleanup(self.admin_client.delete_floatingip, fip['id'])
self.assertRaises(lib_exc.Conflict,
self.admin_client.create_floatingip, **fip_args)

101
neutron/tests/tempest/api/admin/test_routers_dvr.py
Unescape View File

@ -1,101 +0,0 @@
# Copyright 2015 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.
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from neutron.tests.tempest.api import base_routers as base
class RoutersTestDVR(base.BaseRouterTest):
required_extensions = ['router', 'dvr']
@classmethod
def resource_setup(cls):
# The check above will pass if api_extensions=all, which does
# not mean DVR extension itself is present.
# Instead, we have to check whether DVR is actually present by using
# admin credentials to create router with distributed=True attribute
# and checking for BadRequest exception and that the resulting router
# has a distributed attribute.
super(RoutersTestDVR, cls).resource_setup()
name = data_utils.rand_name('pretest-check')
router = cls.admin_client.create_router(name)
if 'distributed' not in router['router']:
msg = "'distributed' attribute not found. DVR Possibly not enabled"
raise cls.skipException(msg)
cls.admin_client.delete_router(router['router']['id'])
@decorators.idempotent_id('08a2a0a8-f1e4-4b34-8e30-e522e836c44e')
def test_distributed_router_creation(self):
"""
Test uses administrative credentials to creates a
DVR (Distributed Virtual Routing) router using the
distributed=True.
Acceptance
The router is created and the "distributed" attribute is
set to True
"""
name = data_utils.rand_name('router')
router = self.admin_client.create_router(name, distributed=True)
self.addCleanup(self.admin_client.delete_router,
router['router']['id'])
self.assertTrue(router['router']['distributed'])
@decorators.idempotent_id('8a0a72b4-7290-4677-afeb-b4ffe37bc352')
def test_centralized_router_creation(self):
"""
Test uses administrative credentials to creates a
CVR (Centralized Virtual Routing) router using the
distributed=False.
Acceptance
The router is created and the "distributed" attribute is
set to False, thus making it a "Centralized Virtual Router"
as opposed to a "Distributed Virtual Router"
"""
name = data_utils.rand_name('router')
router = self.admin_client.create_router(name, distributed=False)
self.addCleanup(self.admin_client.delete_router,
router['router']['id'])
self.assertFalse(router['router']['distributed'])
@decorators.idempotent_id('acd43596-c1fb-439d-ada8-31ad48ae3c2e')
def test_centralized_router_update_to_dvr(self):
"""
Test uses administrative credentials to creates a
CVR (Centralized Virtual Routing) router using the
distributed=False.Then it will "update" the router
distributed attribute to True
Acceptance
The router is created and the "distributed" attribute is
set to False. Once the router is updated, the distributed
attribute will be set to True
"""
name = data_utils.rand_name('router')
# router needs to be in admin state down in order to be upgraded to DVR
router = self.admin_client.create_router(name, distributed=False,
ha=False,
admin_state_up=False)
self.addCleanup(self.admin_client.delete_router,
router['router']['id'])
self.assertFalse(router['router']['distributed'])
self.assertFalse(router['router']['ha'])
router = self.admin_client.update_router(router['router']['id'],
distributed=True)
self.assertTrue(router['router']['distributed'])

104
neutron/tests/tempest/api/admin/test_routers_flavors.py
Unescape View File

@ -1,104 +0,0 @@
#
# 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 neutron_lib.plugins import constants
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
import testtools
from neutron.tests.tempest.api import base_routers as base
class RoutersFlavorTestCase(base.BaseRouterTest):
required_extensions = ['router', 'flavors', 'l3-flavors']
@classmethod
def resource_setup(cls):
super(RoutersFlavorTestCase, cls).resource_setup()
cls.service_profiles = []
cls.flavor_service_profiles = []
# make a flavor based on legacy router for regular tenant to use
driver = ('neutron.services.l3_router.service_providers.'
'single_node.SingleNodeDriver')
try:
sp = cls.admin_client.create_service_profile(driver=driver)
except lib_exc.NotFound as e:
if e.resp_body['type'] == 'ServiceProfileDriverNotFound':
raise cls.skipException("%s is not available" % driver)
raise
cls.service_profiles.append(sp['service_profile'])
cls.flavor = cls.create_flavor(
name='special_flavor',
description='econonomy class',
service_type=constants.L3)
cls.admin_client.create_flavor_service_profile(
cls.flavor['id'], sp['service_profile']['id'])
cls.flavor_service_profiles.append((cls.flavor['id'],
sp['service_profile']['id']))
# make another with a different driver
driver = ('neutron.services.l3_router.service_providers.'
'dvr.DvrDriver')
try:
sp = cls.admin_client.create_service_profile(driver=driver)
except lib_exc.NotFound as e:
if e.resp_body['type'] == 'ServiceProfileDriverNotFound':
raise cls.skipException("%s is not available" % driver)
raise
cls.service_profiles.append(sp['service_profile'])
cls.prem_flavor = cls.create_flavor(
name='better_special_flavor',
description='econonomy comfort',
service_type=constants.L3)
cls.admin_client.create_flavor_service_profile(
cls.prem_flavor['id'], sp['service_profile']['id'])
cls.flavor_service_profiles.append((cls.prem_flavor['id'],
sp['service_profile']['id']))
@classmethod
def resource_cleanup(cls):
for flavor_id, service_profile_id in cls.flavor_service_profiles:
cls.admin_client.delete_flavor_service_profile(flavor_id,
service_profile_id)
for service_profile in cls.service_profiles:
cls.admin_client.delete_service_profile(
service_profile['id'])
super(RoutersFlavorTestCase, cls).resource_cleanup()
@decorators.idempotent_id('a4d01977-e968-4983-b4d9-824ea6c33f4b')
def test_create_router_with_flavor(self):
# ensure regular client can see flavor
flavors = self.client.list_flavors(id=self.flavor['id'])
flavor = flavors['flavors'][0]
self.assertEqual('special_flavor', flavor['name'])
flavors = self.client.list_flavors(id=self.prem_flavor['id'])
prem_flavor = flavors['flavors'][0]
self.assertEqual('better_special_flavor', prem_flavor['name'])
# ensure client can create router with both flavors
router = self.create_router('name', flavor_id=flavor['id'])
self.assertEqual(flavor['id'], router['flavor_id'])
router = self.create_router('name', flavor_id=prem_flavor['id'])
self.assertEqual(prem_flavor['id'], router['flavor_id'])
@decorators.idempotent_id('30e73858-a0fc-409c-a2e0-e9cd2826f6a2')
def test_delete_router_flavor_in_use(self):
self.create_router('name', flavor_id=self.flavor['id'])
with testtools.ExpectedException(lib_exc.Conflict):
self.admin_client.delete_flavor(self.flavor['id'])
@decorators.idempotent_id('83939cf7-5070-41bc-9a3e-cd9f22df2186')
def test_badrequest_on_requesting_flags_and_flavor(self):
with testtools.ExpectedException(lib_exc.BadRequest):
self.admin_client.create_router(
'name', flavor_id=self.flavor['id'], distributed=True)

92
neutron/tests/tempest/api/admin/test_routers_ha.py
Unescape View File

@ -1,92 +0,0 @@
# 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.lib.common.utils import data_utils
from tempest.lib import decorators
from neutron.tests.tempest.api import base_routers as base
class RoutersTestHA(base.BaseRouterTest):
required_extensions = ['router', 'l3-ha']
@classmethod
def resource_setup(cls):
# The check above will pass if api_extensions=all, which does
# not mean "l3-ha" extension itself is present.
# Instead, we have to check whether "ha" is actually present by using
# admin credentials to create router with ha=True attribute
# and checking for BadRequest exception and that the resulting router
# has a high availability attribute.
super(RoutersTestHA, cls).resource_setup()
name = data_utils.rand_name('pretest-check')
router = cls.admin_client.create_router(name)
if 'ha' not in router['router']:
cls.admin_client.delete_router(router['router']['id'])
msg = "'ha' attribute not found. HA Possibly not enabled"
raise cls.skipException(msg)
@decorators.idempotent_id('8abc177d-14f1-4018-9f01-589b299cbee1')
def test_ha_router_creation(self):
"""
Test uses administrative credentials to create a
HA (High Availability) router using the ha=True.
Acceptance
The router is created and the "ha" attribute is set to True
"""
name = data_utils.rand_name('router')
router = self.admin_client.create_router(name, ha=True)
self.addCleanup(self.admin_client.delete_router,
router['router']['id'])
self.assertTrue(router['router']['ha'])
@decorators.idempotent_id('97b5f7ef-2192-4fa3-901e-979cd5c1097a')
def test_legacy_router_creation(self):
"""
Test uses administrative credentials to create a
SF (Single Failure) router using the ha=False.
Acceptance
The router is created and the "ha" attribute is
set to False, thus making it a "Single Failure Router"
as opposed to a "High Availability Router