Browse Source

Remove tempest tests that have moved to neutron-tempest-plugin

This patch removes all tempest tests from neutron-dynamic-routing
repository as all of them are now moved to neutron-tempest-plugin repo.
It also removes all tempest job definitions which are now replaced by
job neutron-tempest-plugin-dynamic-routing defined in
neutron-tempest-plugin repo directly.

Change-Id: I94a443c92fe2317ba2591ae3a227ec8df82ec294
changes/14/695014/1
Nate Johnston 2 years ago
parent
commit
e760a49f67
  1. 95
      .zuul.yaml
  2. 0
      neutron_dynamic_routing/tests/tempest/__init__.py
  3. 0
      neutron_dynamic_routing/tests/tempest/api/__init__.py
  4. 327
      neutron_dynamic_routing/tests/tempest/api/test_bgp_speaker_extensions.py
  5. 125
      neutron_dynamic_routing/tests/tempest/api/test_bgp_speaker_extensions_negative.py
  6. 141
      neutron_dynamic_routing/tests/tempest/bgp_client.py
  7. 34
      neutron_dynamic_routing/tests/tempest/plugin.py
  8. 36
      neutron_dynamic_routing/tests/tempest/scenario/README
  9. 0
      neutron_dynamic_routing/tests/tempest/scenario/__init__.py
  10. 372
      neutron_dynamic_routing/tests/tempest/scenario/base.py
  11. 145
      neutron_dynamic_routing/tests/tempest/scenario/base_test_proto.py
  12. 0
      neutron_dynamic_routing/tests/tempest/scenario/basic/__init__.py
  13. 101
      neutron_dynamic_routing/tests/tempest/scenario/basic/base.py
  14. 130
      neutron_dynamic_routing/tests/tempest/scenario/basic/test_4byte_asn.py
  15. 117
      neutron_dynamic_routing/tests/tempest/scenario/basic/test_basic.py
  16. 0
      neutron_dynamic_routing/tests/tempest/scenario/ipv4/__init__.py
  17. 131
      neutron_dynamic_routing/tests/tempest/scenario/ipv4/test_ipv4.py
  18. 0
      neutron_dynamic_routing/tests/tempest/scenario/ipv6/__init__.py
  19. 132
      neutron_dynamic_routing/tests/tempest/scenario/ipv6/test_ipv6.py

95
.zuul.yaml

@ -14,87 +14,6 @@
- ^releasenotes/.*$
- job:
name: neutron-dynamic-routing-dsvm-tempest-api
parent: legacy-dsvm-base
run: playbooks/neutron-dynamic-routing-dsvm-tempest-api/run.yaml
post-run: playbooks/neutron-dynamic-routing-dsvm-tempest-api/post.yaml
timeout: 7500
required-projects:
- openstack/devstack-gate
- openstack/neutron
- openstack/neutron-dynamic-routing
- openstack/neutron-tempest-plugin
- openstack/tempest
irrelevant-files:
- ^.*\.rst$
- ^doc/.*$
- ^releasenotes/.*$
- job:
name: neutron-dynamic-routing-dsvm-tempest-scenario-basic
parent: legacy-dsvm-base
run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-basic/run.yaml
post-run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-basic/post.yaml
timeout: 7800
required-projects:
- openstack/devstack-gate
- openstack/neutron
- openstack/neutron-dynamic-routing
- openstack/neutron-tempest-plugin
- openstack/tempest
irrelevant-files:
- ^.*\.rst$
- ^doc/.*$
- ^releasenotes/.*$
- job:
name: neutron-dynamic-routing-dsvm-tempest-scenario-ipv4
parent: legacy-dsvm-base
run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-ipv4/run.yaml
post-run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-ipv4/post.yaml
timeout: 7800
required-projects:
- openstack/devstack-gate
- openstack/neutron
- openstack/neutron-dynamic-routing
- openstack/neutron-tempest-plugin
- openstack/tempest
irrelevant-files:
- ^.*\.rst$
- ^doc/.*$
- ^releasenotes/.*$
- job:
name: neutron-dynamic-routing-dsvm-tempest-scenario-ipv6
parent: legacy-dsvm-base
run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-ipv6/run.yaml
post-run: playbooks/neutron-dynamic-routing-dsvm-tempest-scenario-ipv6/post.yaml
timeout: 7800
required-projects:
- openstack/devstack-gate
- openstack/neutron
- openstack/neutron-dynamic-routing
- openstack/neutron-tempest-plugin
- openstack/tempest
irrelevant-files:
- ^.*\.rst$
- ^doc/.*$
- ^releasenotes/.*$
- job:
name: neutron-dynamic-routing-dsvm-tempest-with-os-ken-master-scenario-ipv4
parent: legacy-dsvm-base
run: playbooks/neutron-dynamic-routing-dsvm-tempest-with-os-ken-master-scenario-ipv4/run.yaml
post-run: playbooks/neutron-dynamic-routing-dsvm-tempest-with-os-ken-master-scenario-ipv4/post.yaml
timeout: 7800
required-projects:
- openstack/devstack-gate
- openstack/neutron
- openstack/neutron-dynamic-routing
- openstack/neutron-tempest-plugin
- openstack/tempest
- project:
templates:
- check-requirements
@ -107,21 +26,11 @@
check:
jobs:
- neutron-dynamic-routing-dsvm-functional
- neutron-dynamic-routing-dsvm-tempest-api
- neutron-dynamic-routing-dsvm-tempest-scenario-basic
- neutron-dynamic-routing-dsvm-tempest-scenario-ipv4
- neutron-dynamic-routing-dsvm-tempest-scenario-ipv6
- neutron-tempest-plugin-dynamic-routing
gate:
jobs:
- neutron-dynamic-routing-dsvm-functional
- neutron-dynamic-routing-dsvm-tempest-api
- neutron-dynamic-routing-dsvm-tempest-scenario-basic
- neutron-dynamic-routing-dsvm-tempest-scenario-ipv4
- neutron-dynamic-routing-dsvm-tempest-scenario-ipv6
experimental:
jobs:
- neutron-dynamic-routing-dsvm-tempest-with-os-ken-master-scenario-ipv4
- neutron-tempest-plugin-dynamic-routing
periodic:
jobs:
- neutron-dynamic-routing-dsvm-tempest-with-os-ken-master-scenario-ipv4
- legacy-periodic-neutron-dynamic-routing-dsvm-tempest-with-ryu-master-scenario-ipv4

0
neutron_dynamic_routing/tests/tempest/__init__.py

0
neutron_dynamic_routing/tests/tempest/api/__init__.py

327
neutron_dynamic_routing/tests/tempest/api/test_bgp_speaker_extensions.py

@ -1,327 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 netaddr
from tempest.common import utils
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.api import base
from neutron_tempest_plugin.common import tempest_fixtures as fixtures
from neutron_dynamic_routing.tests.tempest import bgp_client
CONF = config.CONF
def _setup_client_args(auth_provider):
"""Set up ServiceClient arguments using config settings. """
service = CONF.network.catalog_type or 'network'
region = CONF.network.region or 'regionOne'
endpoint_type = CONF.network.endpoint_type
build_interval = CONF.network.build_interval
build_timeout = CONF.network.build_timeout
# The disable_ssl appears in identity
disable_ssl_certificate_validation = (
CONF.identity.disable_ssl_certificate_validation)
ca_certs = None
# Trace in debug section
trace_requests = CONF.debug.trace_requests
return [auth_provider, service, region, endpoint_type,
build_interval, build_timeout,
disable_ssl_certificate_validation, ca_certs,
trace_requests]
class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
default_bgp_speaker_args = {'local_as': '1234',
'ip_version': 4,
'name': 'my-bgp-speaker',
'advertise_floating_ip_host_routes': True,
'advertise_tenant_networks': True}
default_bgp_peer_args = {'remote_as': '4321',
'name': 'my-bgp-peer',
'peer_ip': '192.168.1.1',
'auth_type': 'md5', 'password': 'my-secret'}
def setUp(self):
self.addCleanup(self.resource_cleanup)
super(BgpSpeakerTestJSONBase, self).setUp()
@classmethod
def _setup_bgp_admin_client(cls):
mgr = cls.get_client_manager(credential_type='admin')
auth_provider = mgr.auth_provider
client_args = _setup_client_args(auth_provider)
cls.bgp_adm_client = bgp_client.BgpSpeakerClientJSON(*client_args)
@classmethod
def _setup_bgp_non_admin_client(cls):
mgr = cls.get_client_manager()
auth_provider = mgr.auth_provider
client_args = _setup_client_args(auth_provider)
cls.bgp_client = bgp_client.BgpSpeakerClientJSON(*client_args)
@classmethod
def resource_setup(cls):
super(BgpSpeakerTestJSONBase, cls).resource_setup()
if not utils.is_extension_enabled('bgp', 'network'):
msg = "BGP Speaker extension is not enabled."
raise cls.skipException(msg)
cls.admin_routerports = []
cls.admin_floatingips = []
cls.admin_routers = []
cls.ext_net_id = CONF.network.public_network_id
cls._setup_bgp_admin_client()
cls._setup_bgp_non_admin_client()
@classmethod
def resource_cleanup(cls):
for floatingip in cls.admin_floatingips:
cls._try_delete_resource(cls.admin_client.delete_floatingip,
floatingip['id'])
for routerport in cls.admin_routerports:
cls._try_delete_resource(
cls.admin_client.remove_router_interface_with_subnet_id,
routerport['router_id'], routerport['subnet_id'])
for router in cls.admin_routers:
cls._try_delete_resource(cls.admin_client.delete_router,
router['id'])
super(BgpSpeakerTestJSONBase, cls).resource_cleanup()
def create_bgp_speaker(self, auto_delete=True, **args):
data = {'bgp_speaker': args}
bgp_speaker = self.bgp_adm_client.create_bgp_speaker(data)
bgp_speaker_id = bgp_speaker['bgp_speaker']['id']
if auto_delete:
self.addCleanup(self.delete_bgp_speaker, bgp_speaker_id)
return bgp_speaker['bgp_speaker']
def create_bgp_peer(self, **args):
bgp_peer = self.bgp_adm_client.create_bgp_peer({'bgp_peer': args})
bgp_peer_id = bgp_peer['bgp_peer']['id']
self.addCleanup(self.delete_bgp_peer, bgp_peer_id)
return bgp_peer['bgp_peer']
def update_bgp_speaker(self, id, **args):
data = {'bgp_speaker': args}
return self.bgp_adm_client.update_bgp_speaker(id, data)
def delete_bgp_speaker(self, id):
return self.bgp_adm_client.delete_bgp_speaker(id)
def get_bgp_speaker(self, id):
return self.bgp_adm_client.get_bgp_speaker(id)['bgp_speaker']
def create_bgp_speaker_and_peer(self):
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_peer = self.create_bgp_peer(**self.default_bgp_peer_args)
return (bgp_speaker, bgp_peer)
def delete_bgp_peer(self, id):
return self.bgp_adm_client.delete_bgp_peer(id)
def add_bgp_peer(self, bgp_speaker_id, bgp_peer_id):
return self.bgp_adm_client.add_bgp_peer_with_id(bgp_speaker_id,
bgp_peer_id)
def remove_bgp_peer(self, bgp_speaker_id, bgp_peer_id):
return self.bgp_adm_client.remove_bgp_peer_with_id(bgp_speaker_id,
bgp_peer_id)
def delete_address_scope(self, id):
return self.admin_client.delete_address_scope(id)
class BgpSpeakerTestJSON(BgpSpeakerTestJSONBase):
"""
Tests the following operations in the Neutron API using the REST client for
Neutron:
Create bgp-speaker
Delete bgp-speaker
Create bgp-peer
Update bgp-peer
Delete bgp-peer
"""
@decorators.idempotent_id('df259771-7104-4ffa-b77f-bd183600d7f9')
def test_delete_bgp_speaker(self):
bgp_speaker = self.create_bgp_speaker(auto_delete=False,
**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.delete_bgp_speaker(bgp_speaker_id)
self.assertRaises(lib_exc.NotFound,
self.get_bgp_speaker,
bgp_speaker_id)
@decorators.idempotent_id('81d9dc45-19f8-4c6e-88b8-401d965cd1b0')
def test_create_bgp_peer(self):
self.create_bgp_peer(**self.default_bgp_peer_args)
@decorators.idempotent_id('6ade0319-1ee2-493c-ac4b-5eb230ff3a77')
def test_add_bgp_peer(self):
bgp_speaker, bgp_peer = self.create_bgp_speaker_and_peer()
bgp_speaker_id = bgp_speaker['id']
bgp_peer_id = bgp_peer['id']
self.add_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['peers']
self.assertEqual(1, len(bgp_peers_list))
self.assertTrue(bgp_peer_id in bgp_peers_list)
@decorators.idempotent_id('f9737708-1d79-440b-8350-779f97d882ee')
def test_remove_bgp_peer(self):
bgp_peer = self.create_bgp_peer(**self.default_bgp_peer_args)
bgp_peer_id = bgp_peer['id']
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.add_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['peers']
self.assertTrue(bgp_peer_id in bgp_peers_list)
bgp_speaker = self.remove_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['peers']
self.assertTrue(not bgp_peers_list)
@decorators.idempotent_id('23c8eb37-d10d-4f43-b2e7-6542cb6a4405')
def test_add_gateway_network(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
network_list = bgp_speaker['networks']
self.assertEqual(1, len(network_list))
self.assertTrue(self.ext_net_id in network_list)
@decorators.idempotent_id('6cfc7137-0d99-4a3d-826c-9d1a3a1767b0')
def test_remove_gateway_network(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
networks = bgp_speaker['networks']
self.assertTrue(self.ext_net_id in networks)
self.bgp_adm_client.remove_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id)
bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
network_list = bgp_speaker['networks']
self.assertTrue(not network_list)
@decorators.idempotent_id('5bef22ad-5e70-4f7b-937a-dc1944642996')
def test_get_advertised_routes_null_address_scope(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id)
routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(0, len(routes['advertised_routes']))
@decorators.idempotent_id('cae9cdb1-ad65-423c-9604-d4cd0073616e')
def test_get_advertised_routes_floating_ips(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id)
tenant_net = self.create_network()
tenant_subnet = self.create_subnet(tenant_net)
ext_gw_info = {'network_id': self.ext_net_id}
router = self.admin_client.create_router(
'my-router',
external_gateway_info=ext_gw_info,
admin_state_up=True,
distributed=False)
self.admin_routers.append(router['router'])
self.admin_client.add_router_interface_with_subnet_id(
router['router']['id'],
tenant_subnet['id'])
self.admin_routerports.append({'router_id': router['router']['id'],
'subnet_id': tenant_subnet['id']})
tenant_port = self.create_port(tenant_net)
floatingip = self.create_floatingip(self.ext_net_id)
self.admin_floatingips.append(floatingip)
self.client.update_floatingip(floatingip['id'],
port_id=tenant_port['id'])
routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(floatingip['floating_ip_address'] + '/32',
routes['advertised_routes'][0]['destination'])
@decorators.idempotent_id('c9ad566e-fe8f-4559-8303-bbad9062a30c')
def test_get_advertised_routes_tenant_networks(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding'))
addr_scope = self.create_address_scope('my-scope', ip_version=4)
ext_net = self.create_shared_network(**{'router:external': True})
tenant_net = self.create_network()
ext_subnetpool = self.create_subnetpool(
'test-pool-ext',
is_admin=True,
default_prefixlen=24,
address_scope_id=addr_scope['id'],
prefixes=['8.0.0.0/8'])
tenant_subnetpool = self.create_subnetpool(
'tenant-test-pool',
default_prefixlen=25,
address_scope_id=addr_scope['id'],
prefixes=['10.10.0.0/16'])
self.create_subnet({'id': ext_net['id']},
cidr=netaddr.IPNetwork('8.0.0.0/24'),
ip_version=4,
client=self.admin_client,
subnetpool_id=ext_subnetpool['id'])
tenant_subnet = self.create_subnet(
{'id': tenant_net['id']},
cidr=netaddr.IPNetwork('10.10.0.0/24'),
ip_version=4,
subnetpool_id=tenant_subnetpool['id'])
ext_gw_info = {'network_id': ext_net['id']}
router = self.admin_client.create_router(
'my-router',
external_gateway_info=ext_gw_info,
distributed=False)['router']
self.admin_routers.append(router)
self.admin_client.add_router_interface_with_subnet_id(
router['id'],
tenant_subnet['id'])
self.admin_routerports.append({'router_id': router['id'],
'subnet_id': tenant_subnet['id']})
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
ext_net['id'])
routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(tenant_subnet['cidr'],
routes['advertised_routes'][0]['destination'])
fixed_ip = router['external_gateway_info']['external_fixed_ips'][0]
self.assertEqual(fixed_ip['ip_address'],
routes['advertised_routes'][0]['next_hop'])

125
neutron_dynamic_routing/tests/tempest/api/test_bgp_speaker_extensions_negative.py

@ -1,125 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 netaddr
from tempest.common import utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron_dynamic_routing.tests.tempest.api import test_bgp_speaker_extensions as test_base # noqa
class BgpSpeakerTestJSONNegative(test_base.BgpSpeakerTestJSONBase):
"""Negative test cases asserting proper behavior of BGP API extension"""
@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('75e9ee2f-6efd-4320-bff7-ae24741c8b06')
def test_create_bgp_speaker_illegal_local_asn(self):
wrong_asn = 65537
if utils.is_extension_enabled('bgp_4byte_asn', 'network'):
wrong_asn = 4294967296
self.assertRaises(lib_exc.BadRequest,
self.create_bgp_speaker,
local_as=wrong_asn)
@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('6742ec2e-382a-4453-8791-13a19b47cd13')
def test_create_bgp_speaker_non_admin(self):
self.assertRaises(lib_exc.Forbidden,
self.bgp_client.create_bgp_speaker,
{'bgp_speaker': self.default_bgp_speaker_args})
@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('33f7aaf0-9786-478b-b2d1-a51086a50eb4')
def test_create_bgp_peer_non_admin(self):
self.assertRaises(lib_exc.Forbidden,
self.bgp_client.create_bgp_peer,
{'bgp_peer': self.default_bgp_peer_args})
@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('39435932-0266-4358-899b-0e9b1e53c3e9')
def test_update_bgp_speaker_local_asn(self):
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.assertRaises(lib_exc.BadRequest, self.update_bgp_speaker,
bgp_speaker_id, local_as='4321')
@decorators.idempotent_id('9cc33701-51e5-421f-a5d5-fd7b330e550f')
def test_get_advertised_routes_tenant_networks(self):
addr_scope1 = self.create_address_scope('my-scope1', ip_version=4)
addr_scope2 = self.create_address_scope('my-scope2', ip_version=4)
ext_net = self.create_shared_network(**{'router:external': True})
tenant_net1 = self.create_network()
tenant_net2 = self.create_network()
ext_subnetpool = self.create_subnetpool(
'test-pool-ext',
is_admin=True,
default_prefixlen=24,
address_scope_id=addr_scope1['id'],
prefixes=['8.0.0.0/8'])
tenant_subnetpool1 = self.create_subnetpool(
'tenant-test-pool',
default_prefixlen=25,
address_scope_id=addr_scope1['id'],
prefixes=['10.10.0.0/16'])
tenant_subnetpool2 = self.create_subnetpool(
'tenant-test-pool',
default_prefixlen=25,
address_scope_id=addr_scope2['id'],
prefixes=['11.10.0.0/16'])
self.create_subnet({'id': ext_net['id']},
cidr=netaddr.IPNetwork('8.0.0.0/24'),
ip_version=4,
client=self.admin_client,
subnetpool_id=ext_subnetpool['id'])
tenant_subnet1 = self.create_subnet(
{'id': tenant_net1['id']},
cidr=netaddr.IPNetwork('10.10.0.0/24'),
ip_version=4,
subnetpool_id=tenant_subnetpool1['id'])
tenant_subnet2 = self.create_subnet(
{'id': tenant_net2['id']},
cidr=netaddr.IPNetwork('11.10.0.0/24'),
ip_version=4,
subnetpool_id=tenant_subnetpool2['id'])
ext_gw_info = {'network_id': ext_net['id']}
router = self.admin_client.create_router(
'my-router',
distributed=False,
external_gateway_info=ext_gw_info)['router']
self.admin_routers.append(router)
self.admin_client.add_router_interface_with_subnet_id(
router['id'],
tenant_subnet1['id'])
self.admin_routerports.append({'router_id': router['id'],
'subnet_id': tenant_subnet1['id']})
self.admin_client.add_router_interface_with_subnet_id(
router['id'],
tenant_subnet2['id'])
self.admin_routerports.append({'router_id': router['id'],
'subnet_id': tenant_subnet2['id']})
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
ext_net['id'])
routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(tenant_subnet1['cidr'],
routes['advertised_routes'][0]['destination'])
fixed_ip = router['external_gateway_info']['external_fixed_ips'][0]
self.assertEqual(fixed_ip['ip_address'],
routes['advertised_routes'][0]['next_hop'])

141
neutron_dynamic_routing/tests/tempest/bgp_client.py

@ -1,141 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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_serialization import jsonutils
from tempest.lib.common import rest_client
class BgpSpeakerClientJSON(rest_client.RestClient):
def create_bgp_speaker(self, post_data):
post_body = jsonutils.dumps(post_data)
resp, body = self.post('v2.0/bgp-speakers', post_body)
body = jsonutils.loads(body)
self.expected_success(201, resp.status)
return rest_client.ResponseBody(resp, body)
def get_bgp_speaker(self, id):
uri = 'v2.0/bgp-speakers/{0}'.format(id)
resp, body = self.get(uri)
body = jsonutils.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def get_bgp_speakers(self):
uri = self.get_uri("bgp-speakers")
resp, body = self.get(uri)
body = jsonutils.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBodyList(resp, body)
def update_bgp_speaker(self, id, put_data):
uri = 'v2.0/bgp-speakers/{0}'.format(id)
update_body = {'bgp_speaker': put_data}
update_body = jsonutils.dumps(update_body)
resp, body = self.put(uri, update_body)
body = jsonutils.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_bgp_speaker(self, id):
uri = 'v2.0/bgp-speakers/{0}'.format(id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)
def create_bgp_peer(self, post_data):
post_body = jsonutils.dumps(post_data)
resp, body = self.post('v2.0/bgp-peers', post_body)
body = jsonutils.loads(body)
self.expected_success(201, resp.status)
return rest_client.ResponseBody(resp, body)
def get_bgp_peer(self, id):
uri = 'v2.0/bgp-peers/{0}'.format(id)
resp, body = self.get(uri)
body = jsonutils.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_bgp_peer(self, id):
uri = 'v2.0/bgp-peers/{0}'.format(id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)
def add_bgp_peer_with_id(self, bgp_speaker_id, bgp_peer_id):
uri = 'v2.0/bgp-speakers/%s/add_bgp_peer' % bgp_speaker_id
update_body = {"bgp_peer_id": bgp_peer_id}
update_body = jsonutils.dumps(update_body)
resp, body = self.put(uri, update_body)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def remove_bgp_peer_with_id(self, bgp_speaker_id, bgp_peer_id):
uri = 'v2.0/bgp-speakers/%s/remove_bgp_peer' % bgp_speaker_id
update_body = {"bgp_peer_id": bgp_peer_id}
update_body = jsonutils.dumps(update_body)
resp, body = self.put(uri, update_body)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def add_bgp_gateway_network(self, bgp_speaker_id, network_id):
uri = 'v2.0/bgp-speakers/%s/add_gateway_network' % bgp_speaker_id
update_body = {"network_id": network_id}
update_body = jsonutils.dumps(update_body)
resp, body = self.put(uri, update_body)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def remove_bgp_gateway_network(self, bgp_speaker_id, network_id):
uri = 'v2.0/bgp-speakers/%s/remove_gateway_network' % bgp_speaker_id
update_body = {"network_id": network_id}
update_body = jsonutils.dumps(update_body)
resp, body = self.put(uri, update_body)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def get_bgp_advertised_routes(self, bgp_speaker_id):
base_uri = 'v2.0/bgp-speakers/%s/get_advertised_routes'
uri = base_uri % bgp_speaker_id
resp, body = self.get(uri)
body = jsonutils.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def list_dragents_for_bgp_speaker(self, bgp_speaker_id):
uri = 'v2.0/bgp-speakers/%s/bgp-dragents' % bgp_speaker_id
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def add_bgp_speaker_to_dragent(self, agent_id, bgp_speaker_id):
uri = 'v2.0/agents/%s/bgp-drinstances' % agent_id
update_body = {"bgp_speaker_id": bgp_speaker_id}
update_body = jsonutils.dumps(update_body)
resp, body = self.post(uri, update_body)
self.expected_success(201, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def remove_bgp_speaker_from_dragent(self, agent_id, bgp_speaker_id):
uri = 'v2.0/agents/%s/bgp-drinstances/%s' % (agent_id, bgp_speaker_id)
resp, body = self.delete(uri)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)

34
neutron_dynamic_routing/tests/tempest/plugin.py

@ -1,34 +0,0 @@
# Copyright 2016 Hewlett Packard Enterprise Development Company LP
#
# 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 os
from tempest.test_discover import plugins
import neutron_dynamic_routing
class NeutronDynamicRoutingTempestPlugin(plugins.TempestPlugin):
def load_tests(self):
base_path = os.path.split(os.path.dirname(
os.path.abspath(neutron_dynamic_routing.__file__)))[0]
test_dir = "neutron_dynamic_routing/tests/tempest"
full_test_dir = os.path.join(base_path, test_dir)
return full_test_dir, base_path
def register_opts(self, conf):
pass
def get_opt_lists(self):
pass

36
neutron_dynamic_routing/tests/tempest/scenario/README

@ -1,36 +0,0 @@
scenario tests use the following environment.
diagram:
----------------------+--------------- tenant
| network
+--------+
| router |
+--------+
|
-----+----------------+--------------- provider
| | network
+---------+ |
| dragent | |
+---------+ |
| |
| +--------------+
| |
+--------+
| docker |
| bridge |
+--------+
|
+-----------+------------+-------
| |
+---------+ +---------+
docker | quagga1 | | quagga2 | ...
container +---------+ +---------+
docker container environment is provided by test tool of os-ken.
It has the following functions:
- build and remove a container image.
- run, stop and remove a container.
- some operations to quagga container.
- get some information from quagga container.

0
neutron_dynamic_routing/tests/tempest/scenario/__init__.py

372
neutron_dynamic_routing/tests/tempest/scenario/base.py

@ -1,372 +0,0 @@
# Copyright (C) 2016 VA Linux Systems Japan K.K.
# Copyright (C) 2016 Fumihiko Kakuma <kakuma at valinux co jp>
# 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 collections
import threading
import time
import netaddr
from tempest.common import utils
from tempest import config
from neutron_tempest_plugin.api import base
from os_ken.tests.integrated.common import docker_base as ctn_base
from neutron_dynamic_routing.tests.tempest import bgp_client
CONF = config.CONF
Scope = collections.namedtuple('Scope', 'name')
Pool = collections.namedtuple('Pool', 'name, prefixlen, prefixes')
Net = collections.namedtuple('Net', 'name, net, mask, cidr, router')
SubNet = collections.namedtuple('SubNet', 'name, cidr, mask')
Router = collections.namedtuple('Router', 'name, gw')
AS = collections.namedtuple('AS', 'asn, router_id, adv_net')
CHECKTIME = 180
CHECKTIME_INFO = 60
CHECKTIME_INT = 1
BRIDGE_TYPE = ctn_base.BRIDGE_TYPE_DOCKER
def _setup_client_args(auth_provider):
"""Set up ServiceClient arguments using config settings. """
service = CONF.network.catalog_type or 'network'
region = CONF.network.region or 'regionOne'
endpoint_type = CONF.network.endpoint_type
build_interval = CONF.network.build_interval
build_timeout = CONF.network.build_timeout
# The disable_ssl appears in identity
disable_ssl_certificate_validation = (
CONF.identity.disable_ssl_certificate_validation)
ca_certs = None
# Trace in debug section
trace_requests = CONF.debug.trace_requests
return [auth_provider, service, region, endpoint_type,
build_interval, build_timeout,
disable_ssl_certificate_validation, ca_certs,
trace_requests]
class BgpSpeakerScenarioTestJSONBase(base.BaseAdminNetworkTest):
def setUp(self):
self.addCleanup(self.net_resource_cleanup)
super(BgpSpeakerScenarioTestJSONBase, self).setUp()
@classmethod
def _setup_bgp_non_admin_client(cls):
mgr = cls.get_client_manager()
auth_provider = mgr.auth_provider
client_args = _setup_client_args(auth_provider)
cls.bgp_client = bgp_client.BgpSpeakerClientJSON(*client_args)
@classmethod
def _setup_bgp_admin_client(cls):
mgr = cls.get_client_manager(credential_type='admin')
auth_provider = mgr.auth_provider
client_args = _setup_client_args(auth_provider)
cls.bgp_adm_client = bgp_client.BgpSpeakerClientJSON(*client_args)
@classmethod
def resource_setup(cls):
super(BgpSpeakerScenarioTestJSONBase, cls).resource_setup()
if not utils.is_extension_enabled('bgp', 'network'):
msg = "BGP Speaker extension is not enabled."
raise cls.skipException(msg)
cls.images = []
cls.containers = []
cls.r_ass = []
cls.r_as_ip = []
cls.bridges = []
cls.admin_routerports = []
cls.admin_floatingips = []
cls.admin_routers = []
cls.admin_router_ip = []
cls.resource_setup_container()
cls._setup_bgp_admin_client()
cls._setup_bgp_non_admin_client()
cls.lock = threading.Lock()
@classmethod
def resource_cleanup(cls):
for ctn in cls.containers:
try:
ctn.stop()
except ctn_base.CommandError:
pass
ctn.remove()
for br in cls.bridges:
br.delete()
super(BgpSpeakerScenarioTestJSONBase, cls).resource_cleanup()
@classmethod
def get_subnet(self, start='10.10.1.0', end='10.10.255.0', step=256):
subnet_gen = netaddr.iter_iprange(start, end, step=step)
i = 1
while True:
with self.lock:
try:
yield (i, str(next(subnet_gen)))
except StopIteration:
subnet_gen = netaddr.iter_iprange(start, end, step=step)
yield (i, str(next(subnet_gen)))
i += 1
def net_resource_cleanup(self):
for floatingip in self.admin_floatingips:
self._try_delete_resource(self.admin_client.delete_floatingip,
floatingip['id'])
for routerport in self.admin_routerports:
self._try_delete_resource(
self.admin_client.remove_router_interface_with_subnet_id,
routerport['router_id'], routerport['subnet_id'])
for router in self.admin_routers:
self._try_delete_resource(self.admin_client.delete_router,
router['id'])
def create_bgp_speaker(self, auto_delete=True, **args):
data = {'bgp_speaker': args}
bgp_speaker = self.bgp_adm_client.create_bgp_speaker(data)
bgp_speaker_id = bgp_speaker['bgp_speaker']['id']
if auto_delete:
self.addCleanup(self.bgp_adm_client.delete_bgp_speaker,
bgp_speaker_id)
return bgp_speaker['bgp_speaker']
def delete_bgp_speaker(self, id):
return self.bgp_adm_client.delete_bgp_speaker(id)
def create_bgp_peer(self, auto_delete=True, **args):
bgp_peer = self.bgp_adm_client.create_bgp_peer({'bgp_peer': args})
bgp_peer_id = bgp_peer['bgp_peer']['id']
if auto_delete:
self.addCleanup(self.bgp_adm_client.delete_bgp_peer,
bgp_peer_id)
return bgp_peer['bgp_peer']
def delete_bgp_peer(self, id):
return self.bgp_adm_client.delete_bgp_peer(id)
def get_dragent_id(self):
agents = self.admin_client.list_agents(
agent_type="BGP dynamic routing agent")
self.assertTrue(agents['agents'][0]['alive'])
return agents['agents'][0]['id']
def add_bgp_speaker_to_dragent(self, agent_id, speaker_id):
self.bgp_adm_client.add_bgp_speaker_to_dragent(agent_id, speaker_id)
# tnets[[net1, subnet1, router1], [net2, subnet2, router2], ...]
def create_bgp_network(self, ip_version, scope,
exnet, expool, exsubnet,
tpool, tnets):
addr_scope = self.create_address_scope(scope.name,
ip_version=ip_version)
# external network
ext_net = self.create_shared_network(**{'router:external': True})
ext_net_id = ext_net['id']
ext_subnetpool = self.create_subnetpool(
expool.name,
is_admin=True,
default_prefixlen=expool.prefixlen,
address_scope_id=addr_scope['id'],
prefixes=expool.prefixes)
self.create_subnet(
{'id': ext_net_id},
cidr=netaddr.IPNetwork(exsubnet.cidr),
mask_bits=exsubnet.mask,
ip_version=ip_version,
client=self.admin_client,
subnetpool_id=ext_subnetpool['id'],
reserve_cidr=False)
# tenant network
tenant_subnetpool = self.create_subnetpool(
tpool.name,
default_prefixlen=tpool.prefixlen,
address_scope_id=addr_scope['id'],
prefixes=tpool.prefixes)
for tnet, tsubnet, router in tnets:
tenant_net = self.create_network()
tenant_subnet = self.create_subnet(
{'id': tenant_net['id']},
cidr=netaddr.IPNetwork(tsubnet.cidr),
mask_bits=tsubnet.mask,
ip_version=ip_version,
subnetpool_id=tenant_subnetpool['id'],
reserve_cidr=False)
# router
ext_gw_info = {'network_id': ext_net_id}
router_cr = self.admin_client.create_router(
router.name,
external_gateway_info=ext_gw_info)['router']
self.admin_routers.append(router_cr)
self.admin_client.add_router_interface_with_subnet_id(
router_cr['id'],
tenant_subnet['id'])
self.admin_routerports.append({'router_id': router_cr['id'],
'subnet_id': tenant_subnet['id']})
router = self.admin_client.show_router(router_cr['id'])['router']
fixed_ips = router['external_gateway_info']['external_fixed_ips']
self.admin_router_ip.append(fixed_ips[0]['ip_address'])
return ext_net_id
def create_and_add_peers_to_speaker(self, ext_net_id,
speaker_info, peer_infos,
auto_delete=True):
speaker = self.create_bgp_speaker(auto_delete=auto_delete,
**speaker_info)
speaker_id = speaker['id']
self.bgp_adm_client.add_bgp_gateway_network(speaker_id,
ext_net_id)
peer_ids = []
for peer_args in peer_infos:
peer = self.create_bgp_peer(auto_delete=auto_delete,
**peer_args)
peer_id = peer['id']
peer_ids.append(peer_id)
self.bgp_adm_client.add_bgp_peer_with_id(speaker_id,
peer_id)
return (speaker_id, peer_ids)
def get_remote_as_state(self, l_as, r_as,
expected_state,
init_state=ctn_base.BGP_FSM_IDLE,
checktime=CHECKTIME,
checktime_int=CHECKTIME_INT):
neighbor_state = init_state
for i in range(0, checktime):
neighbor_state = r_as.get_neighbor_state(l_as)
if neighbor_state == expected_state:
break
time.sleep(checktime_int)
return neighbor_state
def check_remote_as_state(self, l_as, r_as,
expected_state,
init_state=ctn_base.BGP_FSM_IDLE,
checktime=CHECKTIME,
checktime_int=CHECKTIME_INT):
neighbor_state = self.get_remote_as_state(l_as, r_as,
expected_state,
init_state=init_state,
checktime=checktime,
checktime_int=checktime_int)
self.assertEqual(neighbor_state, expected_state)
def get_remote_as_of_state_ok(self, l_as, r_ass,
expected_state,
init_state=ctn_base.BGP_FSM_IDLE,
checktime=CHECKTIME,
checktime_int=CHECKTIME_INT):
neighbor_state = init_state
ras_list = []
ras_max = len(r_ass)
for r_as in r_ass:
ras_list.append({'as': r_as, 'check': False})
ok_ras = []
for i in range(0, checktime):
for ras in ras_list:
if ras['check']:
continue
neighbor_state = ras['as'].get_neighbor_state(l_as)
if neighbor_state == expected_state:
ras['check'] = True
ok_ras.append(ras['as'])
if len(ok_ras) >= ras_max:
break
time.sleep(checktime_int)
return ok_ras
def check_multi_remote_as_state(self, l_as, r_ass,
expected_state,
init_state=ctn_base.BGP_FSM_IDLE,
checktime=CHECKTIME,
checktime_int=CHECKTIME_INT):
ok_ras = self.get_remote_as_of_state_ok(
l_as, r_ass,
expected_state,
init_state=init_state,
checktime=checktime,
checktime_int=checktime_int)
self.assertEqual(len(ok_ras), len(r_ass))
def get_remote_as_rib(self, r_as, prefix, rf, key, expected_item,
checktime=CHECKTIME_INFO,
checktime_int=CHECKTIME_INT):
item = None
for i in range(0, checktime):
rib = r_as.get_global_rib(prefix=prefix, rf=rf)
if rib and key in rib[0]:
if expected_item == rib[0][key]:
item = rib[0][key]
break
time.sleep(checktime_int)
return item
def check_remote_as_rib(self, r_as, prefix, rf, key, expected_item,
checktime=CHECKTIME_INFO,
checktime_int=CHECKTIME_INT):
item = self.get_remote_as_rib(r_as=r_as, prefix=prefix, rf=rf,
key=key, expected_item=expected_item,
checktime=checktime,
checktime_int=checktime_int)
self.assertEqual(expected_item, item)
def get_remote_as_of_rib_ok(self, r_ass, prefix, rf, key, expected_item,
checktime=CHECKTIME_INFO,
checktime_int=CHECKTIME_INT):
ras_list = []
ras_max = len(r_ass)
for r_as in r_ass:
ras_list.append({'as': r_as, 'check': False})
ok_ras = []
for i in range(0, checktime):
for ras in ras_list:
if ras['check']:
continue
rib = r_as.get_global_rib(prefix=prefix, rf=rf)
if rib and key in rib[0]:
if expected_item == rib[0][key]:
ras['check'] = True
ok_ras.append(ras['as'])
if len(ok_ras) >= ras_max:
break
time.sleep(checktime_int)
return ok_ras
def check_multi_remote_as_rib(self, r_ass, prefix, rf, key, expected_item,
checktime=CHECKTIME_INFO,
checktime_int=CHECKTIME_INT):
ok_ras = self.get_remote_as_of_rib_ok(
r_ass=r_ass, prefix=prefix, rf=rf,
key=key, expected_item=expected_item,
checktime=checktime,
checktime_int=checktime_int)
self.assertEqual(len(ok_ras), len(r_ass))
def get_next_hop(self, speaker_id, dest_addr):
routes = self.bgp_adm_client.get_bgp_advertised_routes(speaker_id)
next_hop = ''
for route in routes['advertised_routes']:
if route['destination'] == dest_addr:
next_hop = route['next_hop']
break
return next_hop

145
neutron_dynamic_routing/tests/tempest/scenario/base_test_proto.py

@ -1,145 +0,0 @@
# Copyright (C) 2016 VA Linux Systems Japan K.K.
# Copyright (C) 2016 Fumihiko Kakuma <kakuma at valinux co jp>
# 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 import config
from neutron_dynamic_routing.tests.tempest.scenario import base
from os_ken.tests.integrated.common import docker_base as ctn_base
CONF = config.CONF
class BgpSpeakerProtoTestBase(base.BgpSpeakerScenarioTestJSONBase):
def _test_check_neighbor_established(self, ip_version):
self.bgp_peer_args[0]['peer_ip'] = self.r_as_ip[0].split('/')[0]
num, subnet = next(self.tnet_gen)
mask = '/' + str(self.TPool.prefixlen)
TNet = base.Net(name='', net=subnet, mask=self.TPool.prefixlen,
cidr=subnet + mask, router=None)
TSubNet = base.SubNet(name='', cidr=TNet.cidr, mask=TNet.mask)
MyRouter = base.Router(name='my-router' + str(num), gw='')
ext_net_id = self.create_bgp_network(
ip_version, self.MyScope,
self.PNet, self.PPool, self.PSubNet,
self.TPool, [[TNet, TSubNet, MyRouter]])
speaker_id, peers_ids = self.create_and_add_peers_to_speaker(
ext_net_id,
self.bgp_speaker_args,
[self.bgp_peer_args[0]])
self.check_remote_as_state(self.dr, self.r_ass[0],
ctn_base.BGP_FSM_ESTABLISHED)
def _test_check_advertised_tenant_network(self, ip_version):
self.bgp_peer_args[0]['peer_ip'] = self.r_as_ip[0].split('/')[0]
num, subnet = next(self.tnet_gen)
mask = '/' + str(self.TPool.prefixlen)
TNet = base.Net(name='', net=subnet, mask=self.TPool.prefixlen,
cidr=subnet + mask, router=None)
TSubNet = base.SubNet(name='', cidr=TNet.cidr, mask=TNet.mask)
MyRouter = base.Router(name='my-router' + str(num), gw='')
ext_net_id = self.create_bgp_network(
ip_version, self.MyScope,
self.PNet, self.PPool, self.PSubNet,
self.TPool, [[TNet, TSubNet, MyRouter]])
speaker_id, peers_ids = self.create_and_add_peers_to_speaker(
ext_net_id,
self.bgp_speaker_args,
[self.bgp_peer_args[0]])
self.check_remote_as_state(self.dr, self.r_ass[0],
ctn_base.BGP_FSM_ESTABLISHED)
rf = 'ipv' + str(ip_version)
self.check_remote_as_rib(self.r_ass[0], TNet.cidr, rf,
'nexthop',
self.get_next_hop(speaker_id, TNet.cidr))
def _test_check_advertised_multiple_tenant_network(self, ip_version):
self.bgp_peer_args[0]['peer_ip'] = self.r_as_ip[0].split('/')[0]
tnets = []
tnets_cidr = []
for i in range(0, 3):
num, subnet = next(self.tnet_gen)
mask = '/' + str(self.TPool.prefixlen)
TNet = base.Net(name='', net=subnet, mask=self.TPool.prefixlen,
cidr=subnet + mask, router=None)
TSubNet = base.SubNet(name='', cidr=TNet.cidr, mask=TNet.mask)
MyRouter = base.Router(name='my-router' + str(num), gw='')
tnets.append([TNet, TSubNet, MyRouter])
tnets_cidr.append(TNet.cidr)
ext_net_id = self.create_bgp_network(
ip_version, self.MyScope,
self.PNet, self.PPool, self.PSubNet,
self.TPool, tnets)
speaker_id, peers_ids = self.create_and_add_peers_to_speaker(
ext_net_id,
self.bgp_speaker_args,
[self.bgp_peer_args[0]])
self.check_remote_as_state(self.dr, self.r_ass[0],
ctn_base.BGP_FSM_ESTABLISHED)
rf = 'ipv' + str(ip_version)
for cidr in tnets_cidr:
self.check_remote_as_rib(self.r_ass[0], cidr, rf,
'nexthop',
self.get_next_hop(speaker_id, cidr))
def _test_check_neighbor_established_with_multiple_peers(
self, ip_version):
for (bgp_peer_args, r_as_ip) in zip(self.bgp_peer_args,
self.r_as_ip):
bgp_peer_args['peer_ip'] = r_as_ip.split('/')[0]
num, subnet = next(self.tnet_gen)
mask = '/' + str(self.TPool.prefixlen)
TNet = base.Net(name='', net=subnet, mask=self.TPool.prefixlen,
cidr=subnet + mask, router=None)
TSubNet = base.SubNet(name='', cidr=TNet.cidr, mask=TNet.mask)
MyRouter = base.Router(name='my-router' + str(num), gw='')
ext_net_id = self.create_bgp_network(
ip_version, self.MyScope,
self.PNet, self.PPool, self.PSubNet,
self.TPool, [[TNet, TSubNet, MyRouter]])
speaker_id, peers_ids = self.create_and_add_peers_to_speaker(
ext_net_id,
self.bgp_speaker_args,
self.bgp_peer_args)
self.check_multi_remote_as_state(self.dr, self.r_ass,
ctn_base.BGP_FSM_ESTABLISHED)
def _test_check_advertised_tenant_network_with_multiple_peers(
self, ip_version):
for (bgp_peer_args, r_as_ip) in zip(self.bgp_peer_args,
self.r_as_ip):
bgp_peer_args['peer_ip'] = r_as_ip.split('/')[0]
num, subnet = next(self.tnet_gen)
mask = '/' + str(self.TPool.prefixlen)
TNet = base.Net(name='', net=subnet, mask=self.TPool.prefixlen,
cidr=subnet + mask, router=None)
TSubNet = base.SubNet(name='', cidr=TNet.cidr, mask=TNet.mask)
MyRouter = base.Router(name='my-router' + str(num), gw='')
ext_net_id = self.create_bgp_network(
ip_version, self.MyScope,
self.PNet, self.PPool, self.PSubNet,
self.TPool, [[TNet, TSubNet, MyRouter]])
speaker_id, peers_ids = self.create_and_add_peers_to_speaker(
ext_net_id,
self.bgp_speaker_args,
self.bgp_peer_args)
self.check_multi_remote_as_state(self.dr, self.r_ass,
ctn_base.BGP_FSM_ESTABLISHED)
rf = 'ipv' + str(ip_version)
next_hop = self.get_next_hop(speaker_id, TNet.cidr)
self.check_multi_remote_as_rib(self.r_ass, TNet.cidr, rf,
'nexthop', next_hop)

0
neutron_dynamic_routing/tests/tempest/scenario/basic/__init__.py

101
neutron_dynamic_routing/tests/tempest/scenario/basic/base.py

@ -1,101 +0,0 @@
# Copyright (C) 2016 VA Linux Systems Japan K.K.
# Copyright (C) 2016 Fumihiko Kakuma <kakuma at valinux co jp>
# 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 neutron_dynamic_routing.tests.tempest.scenario import base
from os_ken.tests.integrated.common import docker_base as ctn_base
from os_ken.tests.integrated.common import quagga
class BgpSpeakerBasicTestJSONBase(base.BgpSpeakerScenarioTestJSONBase):
RAS_MAX = 3
public_gw = '192.168.10.1'
MyScope = base.Scope(name='my-scope')
PNet = base.Net(name='', net='172.24.6.0', mask=24,
cidr='172.24.6.0/24', router=None)
PPool = base.Pool(name='test-pool-ext', prefixlen=PNet.mask,
prefixes=[PNet.net + '/8'])
PSubNet = base.SubNet(name='', cidr=PNet.cidr, mask=PNet.mask)
TPool = base.Pool(name='tenant-test-pool', prefixlen=28,
prefixes=['10.10.0.0/16'])
L_AS = base.AS(asn='64512', router_id='192.168.0.2', adv_net='')
ras_l = [
base.AS(asn='64522', router_id='192.168.0.12',
adv_net='192.168.162.0/24'),
base.AS(asn='64523', router_id='192.168.0.13',
adv_net='192.168.163.0/24'),
base.AS(asn='64524', router_id='192.168.0.14',
adv_net='192.168.164.0/24')
]
bgp_speaker_args = {
'local_as': L_AS.asn,
'ip_version': 4,
'name': 'my-bgp-speaker1',
'advertise_floating_ip_host_routes': True,
'advertise_tenant_networks': True
}
bgp_peer_args = [
{'remote_as': ras_l[0].asn,
'name': 'my-bgp-peer1',
'peer_ip': None,
'auth_type': 'none'},
{'remote_as': ras_l[1].asn,
'name': 'my-bgp-peer2',
'peer_ip': None,
'auth_type': 'none'},
{'remote_as': ras_l[2].asn,
'name': 'my-bgp-peer3',
'peer_ip': None,
'auth_type': 'none'}
]
def setUp(self):
super(BgpSpeakerBasicTestJSONBase, self).setUp()
@classmethod
def resource_setup_container(cls):
cls.brdc = ctn_base.Bridge(name='br-docker-basic',
subnet='192.168.10.0/24',
start_ip='192.168.10.128',
end_ip='192.168.10.254',
self_ip=True,
fixed_ip=cls.public_gw + '/24',
br_type=base.BRIDGE_TYPE)
cls.bridges.append(cls.brdc)
# This is dummy container object for a dr service.
# This keeps data which passes to a quagga container.
cls.dr = ctn_base.BGPContainer(name='dr', asn=int(cls.L_AS.asn),
router_id=cls.L_AS.router_id)
cls.dr.set_addr_info(bridge='br-docker-basic', ipv4=cls.public_gw)
# quagga container
cls.dockerimg = ctn_base.DockerImage()
cls.q_img = cls.dockerimg.create_quagga(check_exist=True)
cls.images.append(cls.q_img)
for i in range(cls.RAS_MAX):
qg = quagga.QuaggaBGPContainer(name='q' + str(i + 1),
asn=int(cls.ras_l[i].asn),
router_id=cls.ras_l[i].router_id,
ctn_image_name=cls.q_img)
cls.containers.append(qg)
cls.r_ass.append(qg)
qg.add_route(cls.ras_l[i].adv_net)
qg.run(wait=True)
cls.r_as_ip.append(cls.brdc.addif(qg))
qg.add_peer(cls.dr, bridge=cls.brdc.name,
peer_info={'passive': True})
cls.tnet_gen = cls.get_subnet(start='10.10.1.0', end='10.10.255.0',
step=256)

130
neutron_dynamic_routing/tests/tempest/scenario/basic/test_4byte_asn.py

@ -1,130 +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 import config
from tempest.lib import decorators
from neutron_dynamic_routing.tests.tempest.scenario import base
from neutron_dynamic_routing.tests.tempest.scenario import base_test_proto as test_base # noqa
from os_ken.tests.integrated.common import docker_base as ctn_base
from os_ken.tests.integrated.common import quagga
CONF = config.CONF
class BgpSpeaker4byteASNTest(test_base.BgpSpeakerProtoTestBase):
RAS_MAX = 3
ip_version = 4
public_gw = '192.168.10.1'
MyScope = base.Scope(name='my-scope')
PNet = base.Net(name='', net='172.24.6.0', mask=24,
cidr='172.24.6.0/24', router=None)
PPool = base.Pool(name='test-pool-ext', prefixlen=PNet.mask,
prefixes=[PNet.net + '/8'])
PSubNet = base.SubNet(name='', cidr=PNet.cidr, mask=PNet.mask)
TPool = base.Pool(name='tenant-test-pool', prefixlen=28,
prefixes=['10.10.0.0/16'])
L_AS = base.AS(asn='4200000000', router_id='192.168.0.2', adv_net='')
ras_l = [
base.AS(asn='4210000000', router_id='192.168.0.12',
adv_net='192.168.162.0/24'),
base.AS(asn='64522', router_id='192.168.0.13',
adv_net='192.168.163.0/24'),
base.AS(asn='4230000000', router_id='192.168.0.14',
adv_net='192.168.164.0/24')
]
bgp_speaker_args = {
'local_as': L_AS.asn,
'ip_version': ip_version,
'name': 'my-bgp-speaker1',
'advertise_floating_ip_host_routes': True,
'advertise_tenant_networks': True
}