Browse Source

make tempest plugin independent of neutron-tempest-plugin

tempest-tox-plugin-sanity-check fails, and one of the many reasons is
that networking-l2gw-tempest-plugin depends on neutron-tempest-plugin.
This patch removes this dependency by adding clients to handle l2gw and
l2gw-connection REST API calls, and make the tests use these new clients.
NOTE: this patch doesn't handle the issues with the scenario tests.

Change-Id: I3d7569d4a65750b50fc5ff0eee6fb8d94d7c323a
changes/94/667594/3
elajkat 1 year ago
parent
commit
2e265b5e1e
9 changed files with 378 additions and 121 deletions
  1. +0
    -0
      networking_l2gw_tempest_plugin/services/__init__.py
  2. +0
    -0
      networking_l2gw_tempest_plugin/services/l2gw/__init__.py
  3. +59
    -0
      networking_l2gw_tempest_plugin/services/l2gw/l2_gateway_connections_client.py
  4. +72
    -0
      networking_l2gw_tempest_plugin/services/l2gw/l2_gateways_client.py
  5. +56
    -0
      networking_l2gw_tempest_plugin/tests/api/base_l2gw.py
  6. +92
    -0
      networking_l2gw_tempest_plugin/tests/api/clients.py
  7. +39
    -46
      networking_l2gw_tempest_plugin/tests/api/test_l2gw_extensions.py
  8. +59
    -74
      networking_l2gw_tempest_plugin/tests/api/test_l2gw_negative.py
  9. +1
    -1
      networking_l2gw_tempest_plugin/tests/scenario/test_l2gateways.py

+ 0
- 0
networking_l2gw_tempest_plugin/services/__init__.py View File


+ 0
- 0
networking_l2gw_tempest_plugin/services/l2gw/__init__.py View File


+ 59
- 0
networking_l2gw_tempest_plugin/services/l2gw/l2_gateway_connections_client.py View File

@@ -0,0 +1,59 @@
# Copyright (c) 2019 Ericsson
#
# 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.services.network import base


class L2GatewayConnectionsClient(base.BaseNetworkClient):

def create_l2_gateway_connection(self, **kwargs):
"""Creates an l2gw connection.

:param kwargs: dict of l2gw in the form
{'l2_gateway_id': '1234', 'network_id': '4321',
'segmentation_id': '0'}
:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateway-connections'
post_data = {'l2_gateway_connection': kwargs}
return self.create_resource(uri, post_data)

def show_l2_gateway_connection(self, l2gw_connection_id, **fields):
"""Show details of an l2gw connection.

:param l2gw_id: The uuid of the l2gw connection.
:param fields: a dict with key fields and a list of fields name

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateway-connections/%s' % l2gw_connection_id
return self.show_resource(uri, **fields)

def delete_l2_gateway_connection(self, l2gw_id):
"""Deletes and l2gw connection.

:param l2gw_id: The uuid of the l2gw connection.

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateway-connections/%s' % l2gw_id
return self.delete_resource(uri)

def list_l2_gateway_connections(self, **filters):
"""Lists l2gw connections.

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateway-connections'
return self.list_resources(uri, **filters)

+ 72
- 0
networking_l2gw_tempest_plugin/services/l2gw/l2_gateways_client.py View File

@@ -0,0 +1,72 @@
# Copyright (c) 2019 Ericsson
#
# 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.services.network import base


class L2GatewaysClient(base.BaseNetworkClient):

def create_l2_gateway(self, **kwargs):
"""Creates an l2gw.

:param kwargs: dict of l2gw in the form
{'name': 'foo', 'devices': [{'device_name': 'foo',
'interfaces': [{'name': 'bar', 'segmentation_id': 4}]}]
}
:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateways'
post_data = {'l2_gateway': kwargs}
return self.create_resource(uri, post_data)

def update_l2_gateway(self, l2gw_id, **kwargs):
"""Updates an l2gw.

:param l2gw_id: The uuid of the l2gw
:param kwargs: dict of l2gw to be updated.

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateways/%s' % l2gw_id
post_data = {'l2_gateway': kwargs}
return self.update_resource(uri, post_data)

def show_l2_gateway(self, l2gw_id, **fields):
"""Show details of an l2gw.

:param l2gw_id: The uuid of the l2gw.
:param fields: a dict with key fields and a list of fields name

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateways/%s' % l2gw_id
return self.show_resource(uri, **fields)

def delete_l2_gateway(self, l2gw_id):
"""Deletes and l2gw.

:param l2gw_id: The uuid of the l2gw.

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateways/%s' % l2gw_id
return self.delete_resource(uri)

def list_l2_gateways(self, **filters):
"""Lists l2gws.

:return: tempest.lib.common.rest_client.ResponseBody
"""
uri = '/l2-gateways'
return self.list_resources(uri, **filters)

+ 56
- 0
networking_l2gw_tempest_plugin/tests/api/base_l2gw.py View File

@@ -13,6 +13,14 @@
# License for the specific language governing permissions and limitations
# under the License.

from tempest.api.network import base
from tempest.common import utils
from tempest import config

from networking_l2gw_tempest_plugin.tests.api import clients

CONF = config.CONF

SEGMENTATION_ID_DELIMITER = "#"
INTERFACE_SEG_ID_DELIMITER = "|"
DEVICE_INTERFACE_DELIMITER = "::"
@@ -86,3 +94,51 @@ def form_dict_devices(devices):
devices1.setdefault(device_name, []).append(int_seg)
int_seg = []
return devices1


class BaseL2GatewayTest(base.BaseNetworkTest):
force_tenant_isolation = False
credentials = ['primary']

@classmethod
def setup_clients(cls):
super(BaseL2GatewayTest, cls).setup_clients()
cls.l2gws_client = cls.os.l2gw_client
cls.l2gw_conns_client = cls.os.l2gw_connection_client

@classmethod
def resource_setup(cls):
super(BaseL2GatewayTest, cls).resource_setup()
# At least one switch detail should be provided to run the tests
if (len(CONF.l2gw.l2gw_switch) <= 0):
msg = ('At least one switch detail must be defined.')
raise cls.skipException(msg)
if not utils.is_extension_enabled('l2-gateway', 'network'):
msg = "L2Gateway Extension not enabled."
raise cls.skipException(msg)

@classmethod
def get_client_manager(cls, credential_type=None, roles=None,
force_new=None):
manager = super(BaseL2GatewayTest, cls).get_client_manager(
credential_type=credential_type,
roles=roles,
force_new=force_new
)
# Neutron uses a different clients manager than the one in the Tempest
# save the original in case mixed tests need it
if credential_type == 'primary':
cls.os_tempest = manager
return clients.Manager(manager.credentials)


class BaseAdminL2GatewayTest(BaseL2GatewayTest):

credentials = ['primary', 'admin']

@classmethod
def setup_clients(cls):
super(BaseAdminL2GatewayTest, cls).setup_clients()
cls.admin_networks_client = cls.os_admin.networks_client
cls.admin_l2gws_client = cls.os_admin.l2gw_client
cls.admin_l2gw_conns_client = cls.os_admin.l2gw_connection_client

+ 92
- 0
networking_l2gw_tempest_plugin/tests/api/clients.py View File

@@ -0,0 +1,92 @@
# Copyright (c) 2019 Ericsson
#
# 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 clients
from tempest import config
from tempest.lib.services.identity.v2 import tenants_client
from tempest.lib.services.identity.v3 import projects_client

from networking_l2gw_tempest_plugin.services.l2gw import \
l2_gateway_connections_client
from networking_l2gw_tempest_plugin.services.l2gw import l2_gateways_client

CONF = config.CONF


class Manager(clients.Manager):
"""Top level manager for OpenStack tempest clients"""
default_params = {
'disable_ssl_certificate_validation':
CONF.identity.disable_ssl_certificate_validation,
'ca_certs': CONF.identity.ca_certificates_file,
'trace_requests': CONF.debug.trace_requests,
'proxy_url': CONF.service_clients.proxy_url
}

# NOTE: Tempest uses timeout values of compute API if project specific
# timeout values don't exist.
default_params_with_timeout_values = {
'build_interval': CONF.compute.build_interval,
'build_timeout': CONF.compute.build_timeout
}
default_params_with_timeout_values.update(default_params)

def __init__(self, credentials=None, service=None):
super(Manager, self).__init__(credentials=credentials)

self._set_identity_clients()

self.l2gw_client = l2_gateways_client.L2GatewaysClient(
self.auth_provider,
CONF.network.catalog_type,
CONF.network.region or CONF.identity.region,
endpoint_type=CONF.network.endpoint_type,
build_interval=CONF.network.build_interval,
build_timeout=CONF.network.build_timeout,
**self.default_params
)
self.l2gw_connection_client = \
l2_gateway_connections_client.L2GatewayConnectionsClient(
self.auth_provider,
CONF.network.catalog_type,
CONF.network.region or CONF.identity.region,
endpoint_type=CONF.network.endpoint_type,
build_interval=CONF.network.build_interval,
build_timeout=CONF.network.build_timeout,
**self.default_params
)

params = {
'service': CONF.compute.catalog_type,
'region': CONF.compute.region or CONF.identity.region,
'endpoint_type': CONF.compute.endpoint_type,
'build_interval': CONF.compute.build_interval,
'build_timeout': CONF.compute.build_timeout
}
params.update(self.default_params)

def _set_identity_clients(self):
params = {
'service': CONF.identity.catalog_type,
'region': CONF.identity.region
}
params.update(self.default_params_with_timeout_values)
params_v2_admin = params.copy()
params_v2_admin['endpoint_type'] = CONF.identity.v2_admin_endpoint_type
# Client uses admin endpoint type of Keystone API v2
self.tenants_client = tenants_client.TenantsClient(self.auth_provider,
**params_v2_admin)
# Client uses admin endpoint type of Keystone API v3
self.projects_client = projects_client.ProjectsClient(
self.auth_provider, **params_v2_admin)

+ 39
- 46
networking_l2gw_tempest_plugin/tests/api/test_l2gw_extensions.py View File

@@ -15,8 +15,6 @@

import random

from neutron_tempest_plugin.api import base
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
@@ -26,7 +24,7 @@ from networking_l2gw_tempest_plugin.tests.api import base_l2gw
CONF = config.CONF


class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
class L2GatewayExtensionTestJSON(base_l2gw.BaseAdminL2GatewayTest):
_interface = 'json'

"""
@@ -43,28 +41,18 @@ class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
Delete l2gatewayconnection
"""

@classmethod
def resource_setup(cls):
super(L2GatewayExtensionTestJSON, cls).resource_setup()
# At least one switch detail should be provided to run the tests
if (len(CONF.l2gw.l2gw_switch) <= 0):
msg = ('At least one switch detail must be defined.')
raise cls.skipException(msg)
if not utils.is_extension_enabled('l2-gateway', 'network'):
msg = "L2Gateway Extension not enabled."
raise cls.skipException(msg)

@decorators.idempotent_id('3ca07946-a3c9-49ac-b058-8be54abecf1f')
def test_create_show_list_update_delete_l2gateway(self):
# Create an L2Gateway
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gateway['id'])
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway,
l2_gateway['id'])
# Show details of L2Gateway
show_body = self.admin_client.show_l2_gateway(l2_gateway['id'])
show_body = self.admin_l2gws_client.show_l2_gateway(l2_gateway['id'])
self.assertEqual(gw_name, show_body['l2_gateway']['name'])
conf_devices = base_l2gw.form_dict_devices(devices)
create_devices = base_l2gw.form_dict_devices(show_body['l2_gateway']
@@ -73,42 +61,45 @@ class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
for k, v in zip(conf_devices.items(), create_devices.items()):
self.assertEqual(k, v)
# List L2Gateways
self.admin_client.list_l2_gateways()
self.admin_l2gws_client.list_l2_gateways()
# Update the name of an L2Gateway and verify the same
updated_name = 'updated ' + gw_name
update_body = self.admin_client.update_l2_gateway(l2_gateway['id'],
name=updated_name)
update_body = self.admin_l2gws_client.update_l2_gateway(
l2_gateway['id'], name=updated_name)
self.assertEqual(update_body['l2_gateway']['name'], updated_name)
show_body = self.admin_client.show_l2_gateway(l2_gateway['id'])
show_body = self.admin_l2gws_client.show_l2_gateway(l2_gateway['id'])
self.assertEqual(show_body['l2_gateway']['name'], updated_name)

@decorators.idempotent_id('3ad5e945-2b42-4ea8-9c03-0bf41d4167f2')
def test_create_show_list_delete_l2gateway_connection(self):
# Create a network
name = data_utils.rand_name('network')
net_body = self.admin_client.create_network(name=name)
net_body = self.admin_networks_client.create_network(name=name)
net_id = net_body['network']['id']
self.addCleanup(self.admin_client.delete_network, net_id)
self.addCleanup(self.admin_networks_client.delete_network, net_id)
# Create an L2Gateway
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
l2_gw_body = self.admin_client.create_l2_gateway(
l2_gw_body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gw_id = l2_gw_body['l2_gateway']['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)
# Create an L2Gateway Connection
l2_gw_conn_body = self.admin_client.create_l2_gateway_connection(
l2_gateway_id=l2_gw_id, network_id=net_id)
l2_gw_conn_body = \
self.admin_l2gw_conns_client.create_l2_gateway_connection(
l2_gateway_id=l2_gw_id, network_id=net_id)
l2_gw_conn_id = l2_gw_conn_body['l2_gateway_connection']['id']
self.addCleanup(self.admin_client.delete_l2_gateway_connection,
l2_gw_conn_id)
self.addCleanup(
self.admin_l2gw_conns_client.delete_l2_gateway_connection,
l2_gw_conn_id)
# Show details of created L2 Gateway connection
show_body = self.admin_client.show_l2_gateway_connection(l2_gw_conn_id)
show_body = self.admin_l2gw_conns_client.show_l2_gateway_connection(
l2_gw_conn_id)
l2_gw_conn = show_body['l2_gateway_connection']
self.assertEqual(net_id, l2_gw_conn['network_id'])
self.assertEqual(l2_gw_id, l2_gw_conn['l2_gateway_id'])
# List L2Gateway Connections
self.admin_client.list_l2_gateway_connections()
self.admin_l2gw_conns_client.list_l2_gateway_connections()

def test_create_l2gw_conn_with_segid_when_l2gw_created_without_segid(self):
# Create an L2Gateway
@@ -117,27 +108,29 @@ class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
if devices[0]['interfaces'][0]['segmentation_id']:
seg_id = devices[0]['interfaces'][0]['segmentation_id'][0]
devices[0]['interfaces'][0].pop('segmentation_id')
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
l2_gw_id = l2_gateway['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)
# Create a network
name = data_utils.rand_name('network')
net_body = self.admin_client.create_network(name=name)
net_body = self.admin_networks_client.create_network(name=name)
net_id = net_body['network']['id']
self.addCleanup(self.admin_client.delete_network, net_id)
self.addCleanup(self.admin_networks_client.delete_network, net_id)
# Create an L2Gateway Connection
l2_gw_conn_body = self.admin_client.create_l2_gateway_connection(
l2_gateway_id=l2_gw_id, network_id=net_id,
segmentation_id=seg_id)
l2_gw_conn_body = \
self.admin_l2gw_conns_client.create_l2_gateway_connection(
l2_gateway_id=l2_gw_id, network_id=net_id,
segmentation_id=seg_id)
l2_gw_conn_id = l2_gw_conn_body['l2_gateway_connection']['id']
l2_gw_seg_id = l2_gw_conn_body['l2_gateway_connection'][
'segmentation_id']
self.addCleanup(self.admin_client.delete_l2_gateway_connection,
l2_gw_conn_id)
self.addCleanup(
self.admin_l2gw_conns_client.delete_l2_gateway_connection,
l2_gw_conn_id)
# Show details of created L2 Gateway connection
show_body = self.admin_client.show_l2_gateway_connection(
show_body = self.admin_l2gw_conns_client.show_l2_gateway_connection(
l2_gw_conn_id)
l2_gw_conn = show_body['l2_gateway_connection']
self.assertEqual(net_id, l2_gw_conn['network_id'])
@@ -171,15 +164,15 @@ class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
"name": interface_name_list[2]}],
"segmentation_id": str(seg_id[2])}]
# Create the multi-device L2gateway
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices_list)
l2_gateway = body['l2_gateway']
l2_gw_id = l2_gateway['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)
# Check the created multi-device L2Gateway
device_list = list(range(3))
interface_list = list(range(3))
show_body = self.admin_client.show_l2_gateway(l2_gw_id)
show_body = self.admin_l2gws_client.show_l2_gateway(l2_gw_id)
self.assertEqual(gw_name, show_body['l2_gateway']['name'])
self.assertEqual(l2_gateway['id'], show_body['l2_gateway']['id'])
for i in range(3):
@@ -197,9 +190,9 @@ class L2GatewayExtensionTestJSON(base.BaseAdminNetworkTest):
"name": "intNameNew"}]}]
interface_name_list = [
interface_name_list[2], interface_name_list[1], "intNameNew"]
body = self.admin_client.update_l2_gateway(
body = self.admin_l2gws_client.update_l2_gateway(
l2_gateway['id'], devices=device_updated)
show_body_updated = self.admin_client.show_l2_gateway(
show_body_updated = self.admin_l2gws_client.show_l2_gateway(
l2_gateway['id'])
# Check updating of multi-device L2gateway
self.assertEqual(gw_name, show_body_updated['l2_gateway']['name'])


+ 59
- 74
networking_l2gw_tempest_plugin/tests/api/test_l2gw_negative.py View File

@@ -16,8 +16,6 @@

from random import randint

from neutron_tempest_plugin.api import base
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
@@ -28,41 +26,30 @@ from networking_l2gw_tempest_plugin.tests.api import base_l2gw
CONF = config.CONF


class L2GatewaysNegativeTestJSON(base.BaseNetworkTest):
class L2GatewaysNegativeTestJSON(base_l2gw.BaseL2GatewayTest):
_interface = 'json'

@classmethod
def resource_setup(cls):
super(L2GatewaysNegativeTestJSON, cls).resource_setup()
# At least one switch detail should be provided to run the tests
if (len(CONF.l2gw.l2gw_switch) <= 0):
msg = ('At least one switch detail must be defined.')
raise cls.skipException(msg)
if not utils.is_extension_enabled('l2-gateway', 'network'):
msg = "L2Gateway Extension not enabled."
raise cls.skipException(msg)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('b301d83d-3af3-4712-86dc-a6824e9b14e5')
def test_create_l2gateway_non_admin_user(self):
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
self.assertRaises(lib_exc.Forbidden,
self.client.create_l2_gateway,
self.l2gws_client.create_l2_gateway,
name=gw_name, devices=devices)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('68451dfe-b3b5-4eb1-b03f-9935d4a2dbe7')
def test_list_l2gateway_non_admin_user(self):
self.assertRaises(lib_exc.Forbidden,
self.client.list_l2_gateways)
self.l2gws_client.list_l2_gateways)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('f8589452-7aff-4992-b865-5bb5c41fa755')
def test_update_l2gateway_non_admin_user(self):
non_exist_id = data_utils.rand_name('l2gw')
self.assertRaises(lib_exc.Forbidden,
self.client.update_l2_gateway,
self.l2gws_client.update_l2_gateway,
non_exist_id, name="updated_name")

@decorators.attr(type=['negative', 'smoke'])
@@ -70,7 +57,7 @@ class L2GatewaysNegativeTestJSON(base.BaseNetworkTest):
def test_delete_l2gateway_non_admin_user(self):
non_exist_id = data_utils.rand_name('l2gw')
self.assertRaises(lib_exc.Forbidden,
self.client.delete_l2_gateway,
self.l2gws_client.delete_l2_gateway,
non_exist_id)

@decorators.attr(type=['negative', 'smoke'])
@@ -78,61 +65,53 @@ class L2GatewaysNegativeTestJSON(base.BaseNetworkTest):
def test_create_l2gateway_connection_non_admin_user(self):
non_exist_id = data_utils.rand_name('network')
self.assertRaises(lib_exc.Forbidden,
self.client.create_l2_gateway_connection,
self.l2gw_conns_client.create_l2_gateway_connection,
network_id=non_exist_id, l2_gateway_id=non_exist_id)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('a56a0180-7d98-414c-9a44-fe47a30fe436')
def test_list_l2gateway_connection_non_admin_user(self):
self.assertRaises(lib_exc.Forbidden,
self.client.list_l2_gateway_connections)
self.l2gw_conns_client.list_l2_gateway_connections)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('ce42c68d-5c41-4988-8912-233e3fe5c8fd')
def test_delete_l2gateway_connection_non_admin_user(self):
non_exist_id = data_utils.rand_name('l2gwconnection')
self.assertRaises(lib_exc.Forbidden,
self.client.delete_l2_gateway_connection,
self.l2gw_conns_client.delete_l2_gateway_connection,
non_exist_id)


class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
class L2GatewaysNegativeAdminTestJSON(base_l2gw.BaseAdminL2GatewayTest):
_interface = 'json'

@classmethod
def resource_setup(cls):
super(L2GatewaysNegativeAdminTestJSON, cls).resource_setup()
# At least one switch detail should be provided to run the tests
if (len(CONF.l2gw.l2gw_switch) <= 0):
msg = ('At least one switch detail must be defined.')
raise cls.skipException(msg)
if not utils.is_extension_enabled('l2-gateway', 'network'):
msg = "L2Gateway Extension not enabled."
raise cls.skipException(msg)

@decorators.attr(type=['negative', 'smoke'])
@decorators.idempotent_id('42067b44-3aff-4428-8305-d0496bd38179')
def test_delete_l2gw_associated_l2gw_connection(self):
# Create a network
name = data_utils.rand_name('network')
net_body = self.admin_client.create_network(name=name)
net_body = self.admin_networks_client.create_network(name=name)
net_id = net_body['network']['id']
self.addCleanup(self.admin_client.delete_network, net_id)
self.addCleanup(self.admin_networks_client.delete_network, net_id)
# Create an L2Gateway
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gateway['id'])
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway,
l2_gateway['id'])
# Create an L2Gateway Connection
l2_gw_conn_body = self.admin_client.create_l2_gateway_connection(
l2_gateway_id=l2_gateway['id'], network_id=net_id)
l2_gw_conn_body = \
self.admin_l2gw_conns_client.create_l2_gateway_connection(
l2_gateway_id=l2_gateway['id'], network_id=net_id)
l2_gw_conn_id = l2_gw_conn_body['l2_gateway_connection']['id']
self.addCleanup(self.admin_client.delete_l2_gateway_connection,
l2_gw_conn_id)
self.addCleanup(
self.admin_l2gw_conns_client.delete_l2_gateway_connection,
l2_gw_conn_id)
self.assertRaises(lib_exc.Conflict,
self.admin_client.delete_l2_gateway,
self.admin_l2gws_client.delete_l2_gateway,
l2_gateway['id'])

@decorators.attr(type=['negative', 'smoke'])
@@ -146,7 +125,7 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
device = [{"device_name": dev_name, "interfaces":
[{"name": interface_name, "segmentation_id": seg_id_str}]}]
self.assertRaises(lib_exc.BadRequest,
self.admin_client.create_l2_gateway,
self.admin_l2gws_client.create_l2_gateway,
name=gw_name, devices=device
)

@@ -158,23 +137,25 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
interface_name = data_utils.rand_name('interface')
devices = [{"device_name": dev_name, "interfaces":
[{"name": interface_name}]}]
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
l2_gw_id = l2_gateway['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)

# Create a network
name = data_utils.rand_name('network')
net_body = self.admin_client.create_network(name=name)
net_body = self.admin_networks_client.create_network(name=name)
net_id = net_body['network']['id']
self.addCleanup(self.admin_client.delete_network, net_id)
self.addCleanup(self.admin_networks_client.delete_network, net_id)
for i in ['-1', '4095', '4096']:
seg_id = [i]
self.assertRaises(lib_exc.BadRequest,
self.admin_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id,
segmentation_id=seg_id)
self.assertRaises(
lib_exc.BadRequest,
self.admin_l2gw_conns_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id,
segmentation_id=seg_id
)

@decorators.attr(type=['negative', 'smoke'])
def test_create_l2gw_with_invalid_segmentation_id(self):
@@ -187,7 +168,7 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
device = [{"device_name": dev_name, "interfaces":
[{"name": interface_name, "segmentation_id": seg_id}]}]
self.assertRaises(lib_exc.BadRequest,
self.admin_client.create_l2_gateway,
self.admin_l2gws_client.create_l2_gateway,
name=gw_name, devices=device
)

@@ -202,7 +183,7 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
device = [{"device_name": dev_name, "interfaces":
[{"name": interface_name, "segmentation_id": seg_id_str}]}]
self.assertRaises(lib_exc.BadRequest,
self.admin_client.create_l2_gateway,
self.admin_l2gws_client.create_l2_gateway,
name=gw_name, devices=device
)

@@ -210,48 +191,51 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
def test_delete_non_existent_l2gateway(self):
non_exist_id = data_utils.rand_name('l2gw')
self.assertRaises(lib_exc.NotFound,
self.admin_client.delete_l2_gateway,
self.admin_l2gws_client.delete_l2_gateway,
non_exist_id)

@decorators.attr(type=['negative', 'smoke'])
def test_delete_non_existent_l2gateway_connection(self):
non_exist_id = data_utils.rand_name('l2gwConnection')
self.assertRaises(lib_exc.NotFound,
self.admin_client.delete_l2_gateway_connection,
non_exist_id)
self.assertRaises(
lib_exc.NotFound,
self.admin_l2gw_conns_client.delete_l2_gateway_connection,
non_exist_id)

@decorators.attr(type=['negative', 'smoke'])
def test_create_l2gw_connection_with_invalid_network_name(self):
# Create an L2Gateway
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
l2_gw_id = l2_gateway['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)

# Create a network
net_id = "network"
self.assertRaises(lib_exc.NotFound,
self.admin_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id
)
self.assertRaises(
lib_exc.NotFound,
self.admin_l2gw_conns_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id
)

@decorators.attr(type=['negative', 'smoke'])
def test_update_gateway_with_invalid_device_name(self):
# Create an L2Gateway
gw_name = data_utils.rand_name('l2gw')
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gateway['id'])
self.addCleanup(
self.admin_l2gws_client.delete_l2_gateway, l2_gateway['id'])
device_1 = [{"device_name": ""}]

# Create a connection again for same L2Gateway and Network
self.assertRaises(lib_exc.BadRequest,
self.admin_client.update_l2_gateway,
self.admin_l2gws_client.update_l2_gateway,
gw_name, devices=device_1
)

@@ -262,18 +246,19 @@ class L2GatewaysNegativeAdminTestJSON(base.BaseAdminNetworkTest):
devices = base_l2gw.get_l2gw_body(CONF.l2gw.l2gw_switch)["devices"]
if devices[0]['interfaces'][0]['segmentation_id']:
devices[0]['interfaces'][0].pop('segmentation_id')
body = self.admin_client.create_l2_gateway(
body = self.admin_l2gws_client.create_l2_gateway(
name=gw_name, devices=devices)
l2_gateway = body['l2_gateway']
l2_gw_id = l2_gateway['id']
self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
self.addCleanup(self.admin_l2gws_client.delete_l2_gateway, l2_gw_id)

# Create a network
name = data_utils.rand_name('network')
net_body = self.admin_client.create_network(name=name)
net_body = self.admin_networks_client.create_network(name=name)
net_id = net_body['network']['id']
self.addCleanup(self.admin_client.delete_network, net_id)
self.assertRaises(lib_exc.BadRequest,
self.admin_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id
)
self.addCleanup(self.admin_networks_client.delete_network, net_id)
self.assertRaises(
lib_exc.BadRequest,
self.admin_l2gw_conns_client.create_l2_gateway_connection,
l2_gateway_id=l2_gw_id, network_id=net_id
)

+ 1
- 1
networking_l2gw_tempest_plugin/tests/scenario/test_l2gateways.py View File

@@ -21,6 +21,7 @@ from neutron.tests.api import base
from neutron_tempest_plugin import exceptions
from neutron_tempest_plugin import manager
from oslo_log import log
from tempest import config
from tempest.lib.common import rest_client
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
@@ -28,7 +29,6 @@ from tempest.lib import exceptions as lib_exc

from networking_l2gw_tempest_plugin.tests.api import base_l2gw
from networking_l2gw_tempest_plugin.tests.scenario import ovsdb_connections
from networking_l2gw_tempest_plugin.tests.tempest import config

CONF = config.CONF
LOG = log.getLogger(__name__)


Loading…
Cancel
Save