Clean up API tests for neutron-dynamic-routing

This patch adds the devstack plugin for neutron-dynamic-routing,
allowing the API tests here to be run in check jobs and under the
tempest framework. It also cleans up the API tests so they execute
properly after being spun out of the main repository.

Change-Id: I0ed82282037fa6c571dd73c8d9900d8e5d0c93c7
Implements: blueprint bgp-spinout
This commit is contained in:
Ryan Tidwell 2016-06-28 21:41:23 -07:00
parent 1d7155cc0c
commit a636af4e30
9 changed files with 260 additions and 65 deletions

View File

@ -3,6 +3,8 @@ test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \ OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-60} \ OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-60} \
OS_LOG_CAPTURE=1 \ OS_LOG_CAPTURE=1 \
${PYTHON:-python} -m subunit.run discover -t ./ . $LISTOPT $IDOPTION ${PYTHON:-python} -m subunit.run discover -t ./ \
${OS_TEST_PATH:-./neutron_dynamic_routing/tests/unit} \
$LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE test_id_option=--load-list $IDFILE
test_list_option=--list test_list_option=--list

View File

@ -16,14 +16,36 @@ import netaddr
from tempest import config from tempest import config
from tempest.lib import exceptions as lib_exc from tempest.lib import exceptions as lib_exc
from tempest import test from tempest import test
import testtools
from neutron.tests.api import base from neutron.tests.tempest.api import base
from neutron.tests.tempest.common import tempest_fixtures as fixtures from neutron.tests.tempest.common import tempest_fixtures as fixtures
from neutron_dynamic_routing.tests.tempest import bgp_client
CONF = config.CONF 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): class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
default_bgp_speaker_args = {'local_as': '1234', default_bgp_speaker_args = {'local_as': '1234',
@ -36,6 +58,24 @@ class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
'peer_ip': '192.168.1.1', 'peer_ip': '192.168.1.1',
'auth_type': 'md5', 'password': 'my-secret'} '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 @classmethod
def resource_setup(cls): def resource_setup(cls):
super(BgpSpeakerTestJSONBase, cls).resource_setup() super(BgpSpeakerTestJSONBase, cls).resource_setup()
@ -47,6 +87,8 @@ class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
cls.admin_floatingips = [] cls.admin_floatingips = []
cls.admin_routers = [] cls.admin_routers = []
cls.ext_net_id = CONF.network.public_network_id cls.ext_net_id = CONF.network.public_network_id
cls._setup_bgp_admin_client()
cls._setup_bgp_non_admin_client()
@classmethod @classmethod
def resource_cleanup(cls): def resource_cleanup(cls):
@ -64,27 +106,27 @@ class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
def create_bgp_speaker(self, auto_delete=True, **args): def create_bgp_speaker(self, auto_delete=True, **args):
data = {'bgp_speaker': args} data = {'bgp_speaker': args}
bgp_speaker = self.admin_client.create_bgp_speaker(data) bgp_speaker = self.bgp_adm_client.create_bgp_speaker(data)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['bgp_speaker']['id']
if auto_delete: if auto_delete:
self.addCleanup(self.delete_bgp_speaker, bgp_speaker_id) self.addCleanup(self.delete_bgp_speaker, bgp_speaker_id)
return bgp_speaker return bgp_speaker['bgp_speaker']
def create_bgp_peer(self, **args): def create_bgp_peer(self, **args):
bgp_peer = self.admin_client.create_bgp_peer({'bgp_peer': args}) bgp_peer = self.bgp_adm_client.create_bgp_peer({'bgp_peer': args})
bgp_peer_id = bgp_peer['bgp-peer']['id'] bgp_peer_id = bgp_peer['bgp_peer']['id']
self.addCleanup(self.delete_bgp_peer, bgp_peer_id) self.addCleanup(self.delete_bgp_peer, bgp_peer_id)
return bgp_peer return bgp_peer['bgp_peer']
def update_bgp_speaker(self, id, **args): def update_bgp_speaker(self, id, **args):
data = {'bgp_speaker': args} data = {'bgp_speaker': args}
return self.admin_client.update_bgp_speaker(id, data) return self.bgp_adm_client.update_bgp_speaker(id, data)
def delete_bgp_speaker(self, id): def delete_bgp_speaker(self, id):
return self.admin_client.delete_bgp_speaker(id) return self.bgp_adm_client.delete_bgp_speaker(id)
def get_bgp_speaker(self, id): def get_bgp_speaker(self, id):
return self.admin_client.get_bgp_speaker(id) return self.bgp_adm_client.get_bgp_speaker(id)['bgp_speaker']
def create_bgp_speaker_and_peer(self): def create_bgp_speaker_and_peer(self):
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
@ -92,15 +134,15 @@ class BgpSpeakerTestJSONBase(base.BaseAdminNetworkTest):
return (bgp_speaker, bgp_peer) return (bgp_speaker, bgp_peer)
def delete_bgp_peer(self, id): def delete_bgp_peer(self, id):
return self.admin_client.delete_bgp_peer(id) return self.bgp_adm_client.delete_bgp_peer(id)
def add_bgp_peer(self, bgp_speaker_id, bgp_peer_id): def add_bgp_peer(self, bgp_speaker_id, bgp_peer_id):
return self.admin_client.add_bgp_peer_with_id(bgp_speaker_id, return self.bgp_adm_client.add_bgp_peer_with_id(bgp_speaker_id,
bgp_peer_id) bgp_peer_id)
def remove_bgp_peer(self, bgp_speaker_id, bgp_peer_id): def remove_bgp_peer(self, bgp_speaker_id, bgp_peer_id):
return self.admin_client.remove_bgp_peer_with_id(bgp_speaker_id, return self.bgp_adm_client.remove_bgp_peer_with_id(bgp_speaker_id,
bgp_peer_id) bgp_peer_id)
def delete_address_scope(self, id): def delete_address_scope(self, id):
return self.admin_client.delete_address_scope(id) return self.admin_client.delete_address_scope(id)
@ -123,7 +165,7 @@ class BgpSpeakerTestJSON(BgpSpeakerTestJSONBase):
def test_delete_bgp_speaker(self): def test_delete_bgp_speaker(self):
bgp_speaker = self.create_bgp_speaker(auto_delete=False, bgp_speaker = self.create_bgp_speaker(auto_delete=False,
**self.default_bgp_speaker_args) **self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.delete_bgp_speaker(bgp_speaker_id) self.delete_bgp_speaker(bgp_speaker_id)
self.assertRaises(lib_exc.NotFound, self.assertRaises(lib_exc.NotFound,
self.get_bgp_speaker, self.get_bgp_speaker,
@ -136,82 +178,78 @@ class BgpSpeakerTestJSON(BgpSpeakerTestJSONBase):
@test.idempotent_id('6ade0319-1ee2-493c-ac4b-5eb230ff3a77') @test.idempotent_id('6ade0319-1ee2-493c-ac4b-5eb230ff3a77')
def test_add_bgp_peer(self): def test_add_bgp_peer(self):
bgp_speaker, bgp_peer = self.create_bgp_speaker_and_peer() bgp_speaker, bgp_peer = self.create_bgp_speaker_and_peer()
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
bgp_peer_id = bgp_peer['bgp-peer']['id'] bgp_peer_id = bgp_peer['id']
self.add_bgp_peer(bgp_speaker_id, bgp_peer_id) self.add_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['bgp-speaker']['peers'] bgp_peers_list = bgp_speaker['peers']
self.assertEqual(1, len(bgp_peers_list)) self.assertEqual(1, len(bgp_peers_list))
self.assertTrue(bgp_peer_id in bgp_peers_list) self.assertTrue(bgp_peer_id in bgp_peers_list)
@test.idempotent_id('f9737708-1d79-440b-8350-779f97d882ee') @test.idempotent_id('f9737708-1d79-440b-8350-779f97d882ee')
def test_remove_bgp_peer(self): def test_remove_bgp_peer(self):
bgp_peer = self.create_bgp_peer(**self.default_bgp_peer_args) bgp_peer = self.create_bgp_peer(**self.default_bgp_peer_args)
bgp_peer_id = bgp_peer['bgp-peer']['id'] bgp_peer_id = bgp_peer['id']
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.add_bgp_peer(bgp_speaker_id, bgp_peer_id) self.add_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['bgp-speaker']['peers'] bgp_peers_list = bgp_speaker['peers']
self.assertTrue(bgp_peer_id in bgp_peers_list) self.assertTrue(bgp_peer_id in bgp_peers_list)
bgp_speaker = self.remove_bgp_peer(bgp_speaker_id, bgp_peer_id) bgp_speaker = self.remove_bgp_peer(bgp_speaker_id, bgp_peer_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
bgp_peers_list = bgp_speaker['bgp-speaker']['peers'] bgp_peers_list = bgp_speaker['peers']
self.assertTrue(not bgp_peers_list) self.assertTrue(not bgp_peers_list)
@testtools.skip('bug/1553374')
@test.idempotent_id('23c8eb37-d10d-4f43-b2e7-6542cb6a4405') @test.idempotent_id('23c8eb37-d10d-4f43-b2e7-6542cb6a4405')
def test_add_gateway_network(self): def test_add_gateway_network(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding')) self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id) self.ext_net_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
network_list = bgp_speaker['bgp-speaker']['networks'] network_list = bgp_speaker['networks']
self.assertEqual(1, len(network_list)) self.assertEqual(1, len(network_list))
self.assertTrue(self.ext_net_id in network_list) self.assertTrue(self.ext_net_id in network_list)
@testtools.skip('bug/1553374')
@test.idempotent_id('6cfc7137-0d99-4a3d-826c-9d1a3a1767b0') @test.idempotent_id('6cfc7137-0d99-4a3d-826c-9d1a3a1767b0')
def test_remove_gateway_network(self): def test_remove_gateway_network(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding')) self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id) self.ext_net_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
networks = bgp_speaker['bgp-speaker']['networks'] networks = bgp_speaker['networks']
self.assertTrue(self.ext_net_id in networks) self.assertTrue(self.ext_net_id in networks)
self.admin_client.remove_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.remove_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id) self.ext_net_id)
bgp_speaker = self.admin_client.get_bgp_speaker(bgp_speaker_id) bgp_speaker = self.get_bgp_speaker(bgp_speaker_id)
network_list = bgp_speaker['bgp-speaker']['networks'] network_list = bgp_speaker['networks']
self.assertTrue(not network_list) self.assertTrue(not network_list)
@testtools.skip('bug/1553374')
@test.idempotent_id('5bef22ad-5e70-4f7b-937a-dc1944642996') @test.idempotent_id('5bef22ad-5e70-4f7b-937a-dc1944642996')
def test_get_advertised_routes_null_address_scope(self): def test_get_advertised_routes_null_address_scope(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding')) self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id) self.ext_net_id)
routes = self.admin_client.get_bgp_advertised_routes(bgp_speaker_id) routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(0, len(routes['advertised_routes'])) self.assertEqual(0, len(routes['advertised_routes']))
@testtools.skip('bug/1553374')
@test.idempotent_id('cae9cdb1-ad65-423c-9604-d4cd0073616e') @test.idempotent_id('cae9cdb1-ad65-423c-9604-d4cd0073616e')
def test_get_advertised_routes_floating_ips(self): def test_get_advertised_routes_floating_ips(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding')) self.useFixture(fixtures.LockFixture('gateway_network_binding'))
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
self.ext_net_id) self.ext_net_id)
tenant_net = self.create_network() tenant_net = self.create_network()
tenant_subnet = self.create_subnet(tenant_net) tenant_subnet = self.create_subnet(tenant_net)
ext_gw_info = {'network_id': self.ext_net_id} ext_gw_info = {'network_id': self.ext_net_id}
@ -231,12 +269,11 @@ class BgpSpeakerTestJSON(BgpSpeakerTestJSONBase):
self.admin_floatingips.append(floatingip) self.admin_floatingips.append(floatingip)
self.client.update_floatingip(floatingip['id'], self.client.update_floatingip(floatingip['id'],
port_id=tenant_port['id']) port_id=tenant_port['id'])
routes = self.admin_client.get_bgp_advertised_routes(bgp_speaker_id) routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes'])) self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(floatingip['floating_ip_address'] + '/32', self.assertEqual(floatingip['floating_ip_address'] + '/32',
routes['advertised_routes'][0]['destination']) routes['advertised_routes'][0]['destination'])
@testtools.skip('bug/1553374')
@test.idempotent_id('c9ad566e-fe8f-4559-8303-bbad9062a30c') @test.idempotent_id('c9ad566e-fe8f-4559-8303-bbad9062a30c')
def test_get_advertised_routes_tenant_networks(self): def test_get_advertised_routes_tenant_networks(self):
self.useFixture(fixtures.LockFixture('gateway_network_binding')) self.useFixture(fixtures.LockFixture('gateway_network_binding'))
@ -276,10 +313,10 @@ class BgpSpeakerTestJSON(BgpSpeakerTestJSONBase):
self.admin_routerports.append({'router_id': router['id'], self.admin_routerports.append({'router_id': router['id'],
'subnet_id': tenant_subnet['id']}) 'subnet_id': tenant_subnet['id']})
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
ext_net['id']) ext_net['id'])
routes = self.admin_client.get_bgp_advertised_routes(bgp_speaker_id) routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes'])) self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(tenant_subnet['cidr'], self.assertEqual(tenant_subnet['cidr'],
routes['advertised_routes'][0]['destination']) routes['advertised_routes'][0]['destination'])

View File

@ -35,21 +35,21 @@ class BgpSpeakerTestJSONNegative(test_base.BgpSpeakerTestJSONBase):
@test.idempotent_id('6742ec2e-382a-4453-8791-13a19b47cd13') @test.idempotent_id('6742ec2e-382a-4453-8791-13a19b47cd13')
def test_create_bgp_speaker_non_admin(self): def test_create_bgp_speaker_non_admin(self):
self.assertRaises(lib_exc.Forbidden, self.assertRaises(lib_exc.Forbidden,
self.client.create_bgp_speaker, self.bgp_client.create_bgp_speaker,
{'bgp_speaker': self.default_bgp_speaker_args}) {'bgp_speaker': self.default_bgp_speaker_args})
@test.attr(type=['negative', 'smoke']) @test.attr(type=['negative', 'smoke'])
@test.idempotent_id('33f7aaf0-9786-478b-b2d1-a51086a50eb4') @test.idempotent_id('33f7aaf0-9786-478b-b2d1-a51086a50eb4')
def test_create_bgp_peer_non_admin(self): def test_create_bgp_peer_non_admin(self):
self.assertRaises(lib_exc.Forbidden, self.assertRaises(lib_exc.Forbidden,
self.client.create_bgp_peer, self.bgp_client.create_bgp_peer,
{'bgp_peer': self.default_bgp_peer_args}) {'bgp_peer': self.default_bgp_peer_args})
@test.attr(type=['negative', 'smoke']) @test.attr(type=['negative', 'smoke'])
@test.idempotent_id('39435932-0266-4358-899b-0e9b1e53c3e9') @test.idempotent_id('39435932-0266-4358-899b-0e9b1e53c3e9')
def test_update_bgp_speaker_local_asn(self): def test_update_bgp_speaker_local_asn(self):
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.assertRaises(lib_exc.BadRequest, self.update_bgp_speaker, self.assertRaises(lib_exc.BadRequest, self.update_bgp_speaker,
bgp_speaker_id, local_as='4321') bgp_speaker_id, local_as='4321')
@ -109,10 +109,10 @@ class BgpSpeakerTestJSONNegative(test_base.BgpSpeakerTestJSONBase):
self.admin_routerports.append({'router_id': router['id'], self.admin_routerports.append({'router_id': router['id'],
'subnet_id': tenant_subnet2['id']}) 'subnet_id': tenant_subnet2['id']})
bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args) bgp_speaker = self.create_bgp_speaker(**self.default_bgp_speaker_args)
bgp_speaker_id = bgp_speaker['bgp-speaker']['id'] bgp_speaker_id = bgp_speaker['id']
self.admin_client.add_bgp_gateway_network(bgp_speaker_id, self.bgp_adm_client.add_bgp_gateway_network(bgp_speaker_id,
ext_net['id']) ext_net['id'])
routes = self.admin_client.get_bgp_advertised_routes(bgp_speaker_id) routes = self.bgp_adm_client.get_bgp_advertised_routes(bgp_speaker_id)
self.assertEqual(1, len(routes['advertised_routes'])) self.assertEqual(1, len(routes['advertised_routes']))
self.assertEqual(tenant_subnet1['cidr'], self.assertEqual(tenant_subnet1['cidr'],
routes['advertised_routes'][0]['destination']) routes['advertised_routes'][0]['destination'])

View File

@ -0,0 +1,119 @@
# 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)

View File

@ -0,0 +1,34 @@
# 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/api"
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

View File

@ -34,6 +34,8 @@ neutron.db.alembic_migrations =
neutron-dynamic-routing = neutron_dynamic_routing.db.migration:alembic_migrations neutron-dynamic-routing = neutron_dynamic_routing.db.migration:alembic_migrations
oslo.config.opts = oslo.config.opts =
bgp.agent = neutron_dynamic_routing.services.bgp.common.opts:list_bgp_agent_opts bgp.agent = neutron_dynamic_routing.services.bgp.common.opts:list_bgp_agent_opts
tempest.test_plugins =
neutron_tests = neutron_dynamic_routing.tests.tempest.plugin:NeutronDynamicRoutingTempestPlugin
[build_sphinx] [build_sphinx]
all_files = 1 all_files = 1

View File

@ -19,3 +19,4 @@ WebOb>=1.2.3 # MIT
WebTest>=2.0 # MIT WebTest>=2.0 # MIT
oslotest>=1.10.0 # Apache-2.0 oslotest>=1.10.0 # Apache-2.0
reno>=1.8.0 # Apache2 reno>=1.8.0 # Apache2
tempest>=12.1.0 # Apache-2.0