Migrate neutron-fwaas tests to neutron-tempest-plugin

As discussed in the neutron_c1 meeting [1] the QA team would like to
move the tempest tests for the stadium projects from their repos to
repos specific to being tempest plugins.  This is the first part of a
two stage move, by copying over the tempest tests to the
neutron-tempest-plugin repo [2] rather than spawning new repos to be
separate.

[1] http://eavesdrop.openstack.org/meetings/neutron_ci/2019/neutron_ci.2019-03-12-16.01.log.html#l-94
[2] https://etherpad.openstack.org/p/neutron_stadium_move_to_tempest_plugin_repo

Needed-By: https://review.opendev.org/643668
Depends-On: https://review.opendev.org/660483
Change-Id: I979edd26264ae5f9ceab2da350bc99c40145ec40
This commit is contained in:
Nate Johnston 2019-03-15 15:01:21 -04:00
parent 97a8cd53b7
commit 1d36a20e24
13 changed files with 1929 additions and 0 deletions

View File

@ -745,6 +745,30 @@
networking-bgpvpn: https://git.openstack.org/openstack/networking-bgpvpn
networking-bagpipe: https://git.openstack.org/openstack/networking-bagpipe
- job:
name: neutron-tempest-plugin-fwaas
parent: neutron-tempest-plugin
timeout: 10800
required-projects:
- openstack/devstack-gate
- openstack/neutron-fwaas
- openstack/neutron
- openstack/neutron-tempest-plugin
- openstack/tempest
vars:
tempest_test_regex: ^neutron_tempest_plugin\.fwaas
tox_envlist: all-plugin
devstack_plugins:
neutron-fwaas: https://opendev.org/openstack/neutron-fwaas.git
neutron-tempest-plugin: https://opendev.org/openstack/neutron-tempest-plugin.git
network_api_extensions_common: *api_extensions_master
network_api_extensions_fwaas:
- fwaas_v2
devstack_localrc:
NETWORK_API_EXTENSIONS: "{{ (network_api_extensions_common + network_api_extensions_fwaas) | join(',') }}"
files:
- ^neutron_tempest_plugin/fwaas/.*$
- project-template:
name: neutron-tempest-plugin-jobs
check:
@ -808,6 +832,8 @@
jobs:
- neutron-tempest-plugin-sfc
- neutron-tempest-plugin-bgpvpn-bagpipe
- neutron-tempest-plugin-fwaas
gate:
jobs:
- neutron-tempest-plugin-bgpvpn-bagpipe
- neutron-tempest-plugin-fwaas

View File

View File

@ -0,0 +1,21 @@
# 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.api.network import base
from neutron_tempest_plugin.fwaas.common import fwaas_v2_client
class BaseFWaaSTest(fwaas_v2_client.FWaaSClientMixin, base.BaseNetworkTest):
pass

View File

@ -0,0 +1,358 @@
# Copyright 2016
#
# 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
import six
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.fwaas.api import fwaas_v2_base as v2_base
CONF = config.CONF
DEFAULT_FWG = 'default'
class FWaaSv2ExtensionTestJSON(v2_base.BaseFWaaSTest):
"""List of tests
Tests the following operations in the Neutron API using the REST client
for Neutron:
List firewall rules
Create firewall rule
Update firewall rule
Delete firewall rule
Show firewall rule
List firewall policies
Create firewall policy
Update firewall policy
Insert firewall rule to policy
Remove firewall rule from policy
Insert firewall rule after/before rule in policy
Update firewall policy audited attribute
Delete firewall policy
Show firewall policy
List firewall group
Create firewall group
Update firewall group
Delete firewall group
Show firewall group
"""
@classmethod
def resource_setup(cls):
super(FWaaSv2ExtensionTestJSON, cls).resource_setup()
if not utils.is_extension_enabled('fwaas_v2', 'network'):
msg = "FWaaS v2 Extension not enabled."
raise cls.skipException(msg)
def setUp(self):
super(FWaaSv2ExtensionTestJSON, self).setUp()
self.fw_rule_1 = self.create_firewall_rule(action="allow",
protocol="tcp")
self.fw_rule_2 = self.create_firewall_rule(action="deny",
protocol="udp")
self.fw_policy_1 = self.create_firewall_policy(
firewall_rules=[self.fw_rule_1['id']])
self.fw_policy_2 = self.create_firewall_policy(
firewall_rules=[self.fw_rule_2['id']])
def _create_router_interfaces(self):
network_1 = self.create_network()
network_2 = self.create_network()
cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
mask_bits = CONF.network.project_network_mask_bits
subnet_cidr_1 = list(cidr.subnet(mask_bits))[-1]
subnet_cidr_2 = list(cidr.subnet(mask_bits))[-2]
subnet_1 = self.create_subnet(network_1, cidr=subnet_cidr_1,
mask_bits=mask_bits)
subnet_2 = self.create_subnet(network_2, cidr=subnet_cidr_2,
mask_bits=mask_bits)
router = self.create_router(
data_utils.rand_name('router-'),
admin_state_up=True)
self.addCleanup(self._try_delete_router, router)
intf_1 = self.routers_client.add_router_interface(router['id'],
subnet_id=subnet_1['id'])
intf_2 = self.routers_client.add_router_interface(router['id'],
subnet_id=subnet_2['id'])
return intf_1, intf_2
def _try_delete_router(self, router):
# delete router, if it exists
try:
self.delete_router(router)
# if router is not found, this means it was deleted in the test
except lib_exc.NotFound:
pass
def _try_delete_policy(self, policy_id):
# delete policy, if it exists
try:
self.firewall_policies_client.delete_firewall_policy(policy_id)
# if policy is not found, this means it was deleted in the test
except lib_exc.NotFound:
pass
def _try_delete_rule(self, rule_id):
# delete rule, if it exists
try:
self.firewall_rules_client.delete_firewall_rule(rule_id)
# if rule is not found, this means it was deleted in the test
except lib_exc.NotFound:
pass
def _try_delete_firewall_group(self, fwg_id):
# delete firewall group, if it exists
try:
self.firewall_groups_client.delete_firewall_group(fwg_id)
# if firewall group is not found, this means it was deleted in the test
except lib_exc.NotFound:
pass
self.firewall_groups_client.wait_for_resource_deletion(fwg_id)
def _wait_until_ready(self, fwg_id):
target_states = ('ACTIVE', 'CREATED')
def _wait():
firewall_group = self.firewall_groups_client.show_firewall_group(
fwg_id)
firewall_group = firewall_group['firewall_group']
return firewall_group['status'] in target_states
if not test_utils.call_until_true(_wait, CONF.network.build_timeout,
CONF.network.build_interval):
m = ("Timed out waiting for firewall_group %s to reach %s "
"state(s)" %
(fwg_id, target_states))
raise lib_exc.TimeoutException(m)
def _wait_until_deleted(self, fwg_id):
def _wait():
try:
fwg = self.firewall_groups_client.show_firewall_group(fwg_id)
except lib_exc.NotFound:
return True
fwg_status = fwg['firewall_group']['status']
if fwg_status == 'ERROR':
raise lib_exc.DeleteErrorException(resource_id=fwg_id)
if not test_utils.call_until_true(_wait, CONF.network.build_timeout,
CONF.network.build_interval):
m = ("Timed out waiting for firewall_group %s deleted" % fwg_id)
raise lib_exc.TimeoutException(m)
@decorators.idempotent_id('ddccfa87-4af7-48a6-9e50-0bd0ad1348cb')
def test_list_firewall_rules(self):
# List firewall rules
fw_rules = self.firewall_rules_client.list_firewall_rules()
fw_rules = fw_rules['firewall_rules']
self.assertIn((self.fw_rule_1['id'],
self.fw_rule_1['name'],
self.fw_rule_1['action'],
self.fw_rule_1['protocol'],
self.fw_rule_1['ip_version'],
self.fw_rule_1['enabled']),
[(m['id'],
m['name'],
m['action'],
m['protocol'],
m['ip_version'],
m['enabled']) for m in fw_rules])
@decorators.idempotent_id('ffc009fa-cd17-4029-8025-c4b81a7dd923')
def test_create_update_delete_firewall_rule(self):
# Create firewall rule
body = self.firewall_rules_client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
protocol="tcp")
fw_rule_id = body['firewall_rule']['id']
self.addCleanup(self._try_delete_rule, fw_rule_id)
# Update firewall rule
body = self.firewall_rules_client.update_firewall_rule(fw_rule_id,
action="deny")
self.assertEqual("deny", body["firewall_rule"]['action'])
# Delete firewall rule
self.firewall_rules_client.delete_firewall_rule(fw_rule_id)
# Confirm deletion
fw_rules = self.firewall_rules_client.list_firewall_rules()
self.assertNotIn(fw_rule_id,
[m['id'] for m in fw_rules['firewall_rules']])
@decorators.idempotent_id('76b07afc-444e-4bb9-abec-9b8c5f994dcd')
def test_show_firewall_rule(self):
# show a created firewall rule
fw_rule = self.firewall_rules_client.show_firewall_rule(
self.fw_rule_1['id'])
for key, value in six.iteritems(fw_rule['firewall_rule']):
if key != 'firewall_policy_id':
self.assertEqual(self.fw_rule_1[key], value)
# This check is placed because we cannot modify policy during
# Create/Update of Firewall Rule but we can see the association
# of a Firewall Rule with the policies it belongs to.
@decorators.idempotent_id('f6b83902-746f-4e74-9403-2ec9899583a3')
def test_list_firewall_policies(self):
fw_policies = self.firewall_policies_client.list_firewall_policies()
fw_policies = fw_policies['firewall_policies']
self.assertIn((self.fw_policy_1['id'],
self.fw_policy_1['name'],
self.fw_policy_1['firewall_rules']),
[(m['id'],
m['name'],
m['firewall_rules']) for m in fw_policies])
@decorators.idempotent_id('6ef9bd02-7349-4d61-8d1f-80479f64d904')
def test_create_update_delete_firewall_policy(self):
# Create firewall policy
body = self.firewall_policies_client.create_firewall_policy(
name=data_utils.rand_name("fw-policy"))
fw_policy_id = body['firewall_policy']['id']
self.addCleanup(self._try_delete_policy, fw_policy_id)
# Update firewall policy
body = self.firewall_policies_client.update_firewall_policy(
fw_policy_id,
name="updated_policy")
updated_fw_policy = body["firewall_policy"]
self.assertEqual("updated_policy", updated_fw_policy['name'])
# Delete firewall policy
self.firewall_policies_client.delete_firewall_policy(fw_policy_id)
# Confirm deletion
fw_policies = self.firewall_policies_client.list_firewall_policies()
fw_policies = fw_policies['firewall_policies']
self.assertNotIn(fw_policy_id, [m['id'] for m in fw_policies])
@decorators.idempotent_id('164381de-61f4-483f-9a5a-48105b8e70e2')
def test_show_firewall_policy(self):
# show a created firewall policy
fw_policy = self.firewall_policies_client.show_firewall_policy(
self.fw_policy_1['id'])
fw_policy = fw_policy['firewall_policy']
for key, value in six.iteritems(fw_policy):
self.assertEqual(self.fw_policy_1[key], value)
@decorators.idempotent_id('48dfcd75-3924-479d-bb65-b3ed33397663')
def test_create_show_delete_firewall_group(self):
# create router and add interfaces
intf_1, intf_2 = self._create_router_interfaces()
# Create firewall_group
body = self.firewall_groups_client.create_firewall_group(
name=data_utils.rand_name("firewall_group"),
ingress_firewall_policy_id=self.fw_policy_1['id'],
egress_firewall_policy_id=self.fw_policy_2['id'],
ports=[intf_1['port_id'], intf_2['port_id']])
created_firewall_group = body['firewall_group']
fwg_id = created_firewall_group['id']
# Wait for the firewall resource to become ready
self._wait_until_ready(fwg_id)
# show created firewall_group
firewall_group = self.firewall_groups_client.show_firewall_group(
fwg_id)
fwg = firewall_group['firewall_group']
for key, value in six.iteritems(fwg):
if key == 'status':
continue
self.assertEqual(created_firewall_group[key], value)
# list firewall_groups
firewall_groups = self.firewall_groups_client.list_firewall_groups()
fwgs = firewall_groups['firewall_groups']
self.assertIn((created_firewall_group['id'],
created_firewall_group['name'],
created_firewall_group['ingress_firewall_policy_id'],
created_firewall_group['egress_firewall_policy_id']),
[(m['id'],
m['name'],
m['ingress_firewall_policy_id'],
m['egress_firewall_policy_id']) for m in fwgs])
# Disassociate all port with this firewall group
self.firewall_groups_client.update_firewall_group(fwg_id, ports=[])
# Delete firewall_group
self.firewall_groups_client.delete_firewall_group(fwg_id)
# Wait for the firewall group to be deleted
self._wait_until_deleted(fwg_id)
# Confirm deletion
firewall_groups = self.firewall_groups_client.list_firewall_groups()
fwgs = firewall_groups['firewall_groups']
self.assertNotIn(fwg_id, [m['id'] for m in fwgs])
@decorators.idempotent_id('e021baab-d4f7-4bad-b382-bde4946e0e0b')
def test_update_firewall_group(self):
# create router and add interfaces
intf_1, intf_2 = self._create_router_interfaces()
# Create firewall_group
body = self.firewall_groups_client.create_firewall_group(
name=data_utils.rand_name("firewall_group"),
ingress_firewall_policy_id=self.fw_policy_1['id'],
egress_firewall_policy_id=self.fw_policy_2['id'],
ports=[intf_1['port_id']])
created_firewall_group = body['firewall_group']
fwg_id = created_firewall_group['id']
self.addCleanup(self._try_delete_firewall_group, fwg_id)
# Wait for the firewall resource to become ready
self._wait_until_ready(fwg_id)
# Update firewall group
body = self.firewall_groups_client.update_firewall_group(
fwg_id,
ports=[intf_2['port_id']])
updated_fwg = body["firewall_group"]
self.assertEqual([intf_2['port_id']], updated_fwg['ports'])
# Delete firewall_group
self.firewall_groups_client.delete_firewall_group(fwg_id)
@decorators.idempotent_id('a1f524d8-5336-4769-aa7b-0830bb4df6c8')
def test_error_on_create_firewall_group_name_default(self):
try:
# Create firewall_group with name == reserved default fwg
self.firewall_groups_client.create_firewall_group(
name=DEFAULT_FWG)
except lib_exc.Conflict:
pass
@decorators.idempotent_id('fd24db16-c8cb-4cb4-ba60-b0cd18a66b83')
def test_default_fwg_created_on_list_firewall_groups(self):
fw_groups = self.firewall_groups_client.list_firewall_groups()
fw_groups = fw_groups['firewall_groups']
self.assertIn(DEFAULT_FWG,
[g['name'] for g in fw_groups])

View File

@ -0,0 +1,162 @@
# Copyright (c) 2015 Midokura SARL
# 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 time
from neutron_lib import constants as nl_constants
from tempest import config
from tempest import exceptions
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.fwaas.services import v2_client
CONF = config.CONF
class FWaaSClientMixin(object):
@classmethod
def resource_setup(cls):
super(FWaaSClientMixin, cls).resource_setup()
manager = cls.os_primary
cls.firewall_groups_client = v2_client.FirewallGroupsClient(
manager.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,
**manager.default_params)
cls.firewall_policies_client = v2_client.FirewallPoliciesClient(
manager.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,
**manager.default_params)
cls.firewall_rules_client = v2_client.FirewallRulesClient(
manager.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,
**manager.default_params)
def create_firewall_rule(self, **kwargs):
body = self.firewall_rules_client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
**kwargs)
fw_rule = body['firewall_rule']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.firewall_rules_client.delete_firewall_rule,
fw_rule['id'])
return fw_rule
def create_firewall_policy(self, **kwargs):
body = self.firewall_policies_client.create_firewall_policy(
name=data_utils.rand_name("fw-policy"),
**kwargs)
fw_policy = body['firewall_policy']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.firewall_policies_client.delete_firewall_policy,
fw_policy['id'])
return fw_policy
def create_firewall_group(self, **kwargs):
body = self.firewall_groups_client.create_firewall_group(
name=data_utils.rand_name("fwg"),
**kwargs)
fwg = body['firewall_group']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.delete_firewall_group_and_wait,
fwg['id'])
return fwg
def delete_firewall_group_and_wait(self, firewall_group_id):
self.firewall_groups_client.delete_firewall_group(firewall_group_id)
self._wait_firewall_group_while(firewall_group_id,
[nl_constants.PENDING_DELETE],
not_found_ok=True)
def insert_firewall_rule_in_policy_and_wait(self,
firewall_group_id,
firewall_policy_id,
firewall_rule_id, **kwargs):
self.firewall_policies_client.insert_firewall_rule_in_policy(
firewall_policy_id=firewall_policy_id,
firewall_rule_id=firewall_rule_id,
**kwargs)
self.addCleanup(
self._call_and_ignore_exceptions,
(lib_exc.NotFound, lib_exc.BadRequest),
self.remove_firewall_rule_from_policy_and_wait,
firewall_group_id=firewall_group_id,
firewall_policy_id=firewall_policy_id,
firewall_rule_id=firewall_rule_id)
self._wait_firewall_group_ready(firewall_group_id)
def remove_firewall_rule_from_policy_and_wait(self,
firewall_group_id,
firewall_policy_id,
firewall_rule_id):
self.firewall_policies_client.remove_firewall_rule_from_policy(
firewall_policy_id=firewall_policy_id,
firewall_rule_id=firewall_rule_id)
self._wait_firewall_group_ready(firewall_group_id)
@staticmethod
def _call_and_ignore_exceptions(exc_list, func, *args, **kwargs):
"""Call the given function and pass if a given exception is raised."""
try:
return func(*args, **kwargs)
except exc_list:
pass
def _wait_firewall_group_ready(self, firewall_group_id):
self._wait_firewall_group_while(firewall_group_id,
[nl_constants.PENDING_CREATE,
nl_constants.PENDING_UPDATE])
def _wait_firewall_group_while(self, firewall_group_id, statuses,
not_found_ok=False):
start = int(time.time())
if not_found_ok:
expected_exceptions = (lib_exc.NotFound)
else:
expected_exceptions = ()
while True:
try:
fwg = self.firewall_groups_client.show_firewall_group(
firewall_group_id)
except expected_exceptions:
break
status = fwg['firewall_group']['status']
if status not in statuses:
break
if (int(time.time()) - start >=
self.firewall_groups_client.build_timeout):
msg = ("Firewall Group %(firewall_group)s failed to reach "
"non PENDING status (current %(status)s)") % {
"firewall_group": firewall_group_id,
"status": status,
}
raise exceptions.TimeoutException(msg)
time.sleep(1)

View File

@ -0,0 +1,69 @@
# Copyright (c) 2015 Midokura SARL
# 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 tempest.lib.common import ssh
from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.fwaas.common import fwaas_v2_client
from neutron_tempest_plugin.fwaas.scenario import fwaas_v2_manager as manager
CONF = config.CONF
class FWaaSScenarioTestBase(object):
def check_connectivity(self, ip_address, username=None, private_key=None,
should_connect=True,
check_icmp=True, check_ssh=True,
check_reverse_icmp_ip=None,
should_reverse_connect=True):
if should_connect:
msg = "Timed out waiting for %s to become reachable" % ip_address
else:
msg = "ip address %s is reachable" % ip_address
if check_icmp:
ok = self.ping_ip_address(ip_address,
should_succeed=should_connect)
self.assertTrue(ok, msg=msg)
if check_ssh:
connect_timeout = CONF.validation.connect_timeout
kwargs = {}
if not should_connect:
# Use a shorter timeout for negative case
kwargs['timeout'] = 1
try:
client = ssh.Client(ip_address, username, pkey=private_key,
channel_timeout=connect_timeout,
**kwargs)
client.test_connection_auth()
self.assertTrue(should_connect, "Unexpectedly reachable")
if check_reverse_icmp_ip:
cmd = 'ping -c1 -w1 %s' % check_reverse_icmp_ip
try:
client.exec_command(cmd)
self.assertTrue(should_reverse_connect,
"Unexpectedly reachable (reverse)")
except lib_exc.SSHExecCommandFailed:
if should_reverse_connect:
raise
except lib_exc.SSHTimeout:
if should_connect:
raise
class FWaaSScenarioTest_V2(fwaas_v2_client.FWaaSClientMixin,
FWaaSScenarioTestBase,
manager.NetworkScenarioTest):
pass

View File

@ -0,0 +1,867 @@
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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 subprocess
import netaddr
from oslo_log import log
from oslo_utils import netutils
from tempest.common import compute
from tempest.common.utils.linux import remote_client
from tempest.common.utils import net_utils
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import exceptions as lib_exc
import tempest.test
CONF = config.CONF
LOG = log.getLogger(__name__)
class ScenarioTest(tempest.test.BaseTestCase):
"""Base class for scenario tests. Uses tempest own clients. """
credentials = ['primary']
@classmethod
def setup_clients(cls):
super(ScenarioTest, cls).setup_clients()
# Clients (in alphabetical order)
cls.keypairs_client = cls.os_primary.keypairs_client
cls.servers_client = cls.os_primary.servers_client
# Neutron network client
cls.networks_client = cls.os_primary.networks_client
cls.ports_client = cls.os_primary.ports_client
cls.routers_client = cls.os_primary.routers_client
cls.subnets_client = cls.os_primary.subnets_client
cls.floating_ips_client = cls.os_primary.floating_ips_client
cls.security_groups_client = cls.os_primary.security_groups_client
cls.security_group_rules_client = (
cls.os_primary.security_group_rules_client)
# Test functions library
#
# The create_[resource] functions only return body and discard the
# resp part which is not used in scenario tests
def _create_port(self, network_id, client=None, namestart='port-quotatest',
**kwargs):
if not client:
client = self.ports_client
name = data_utils.rand_name(namestart)
result = client.create_port(
name=name,
network_id=network_id,
**kwargs)
self.assertIsNotNone(result, 'Unable to allocate port')
port = result['port']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
client.delete_port, port['id'])
return port
def create_keypair(self, client=None):
if not client:
client = self.keypairs_client
name = data_utils.rand_name(self.__class__.__name__)
# We don't need to create a keypair by pubkey in scenario
body = client.create_keypair(name=name)
self.addCleanup(client.delete_keypair, name)
return body['keypair']
def create_server(self, name=None, image_id=None, flavor=None,
validatable=False, wait_until='ACTIVE',
clients=None, **kwargs):
"""Wrapper utility that returns a test server.
This wrapper utility calls the common create test server and
returns a test server. The purpose of this wrapper is to minimize
the impact on the code of the tests already using this
function.
"""
# NOTE(jlanoux): As a first step, ssh checks in the scenario
# tests need to be run regardless of the run_validation and
# validatable parameters and thus until the ssh validation job
# becomes voting in CI. The test resources management and IP
# association are taken care of in the scenario tests.
# Therefore, the validatable parameter is set to false in all
# those tests. In this way create_server just return a standard
# server and the scenario tests always perform ssh checks.
# Needed for the cross_tenant_traffic test:
if clients is None:
clients = self.os_primary
if name is None:
name = data_utils.rand_name(self.__class__.__name__ + "-server")
vnic_type = CONF.network.port_vnic_type
# If vnic_type is configured create port for
# every network
if vnic_type:
ports = []
create_port_body = {'binding:vnic_type': vnic_type,
'namestart': 'port-smoke'}
if kwargs:
# Convert security group names to security group ids
# to pass to create_port
if 'security_groups' in kwargs:
security_groups = \
clients.security_groups_client.list_security_groups(
).get('security_groups')
sec_dict = dict([(s['name'], s['id'])
for s in security_groups])
sec_groups_names = [s['name'] for s in kwargs.pop(
'security_groups')]
security_groups_ids = [sec_dict[s]
for s in sec_groups_names]
if security_groups_ids:
create_port_body[
'security_groups'] = security_groups_ids
networks = kwargs.pop('networks', [])
else:
networks = []
# If there are no networks passed to us we look up
# for the project's private networks and create a port.
# The same behaviour as we would expect when passing
# the call to the clients with no networks
if not networks:
networks = clients.networks_client.list_networks(
**{'router:external': False, 'fields': 'id'})['networks']
# It's net['uuid'] if networks come from kwargs
# and net['id'] if they come from
# clients.networks_client.list_networks
for net in networks:
net_id = net.get('uuid', net.get('id'))
if 'port' not in net:
port = self._create_port(network_id=net_id,
client=clients.ports_client,
**create_port_body)
ports.append({'port': port['id']})
else:
ports.append({'port': net['port']})
if ports:
kwargs['networks'] = ports
self.ports = ports
tenant_network = self.get_tenant_network()
body, servers = compute.create_test_server(
clients,
tenant_network=tenant_network,
wait_until=wait_until,
name=name, flavor=flavor,
image_id=image_id, **kwargs)
self.addCleanup(waiters.wait_for_server_termination,
clients.servers_client, body['id'])
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
clients.servers_client.delete_server, body['id'])
server = clients.servers_client.show_server(body['id'])['server']
return server
def get_remote_client(self, ip_address, username=None, private_key=None):
"""Get a SSH client to a remote server
@param ip_address the server floating or fixed IP address to use
for ssh validation
@param username name of the Linux account on the remote server
@param private_key the SSH private key to use
@return a RemoteClient object
"""
if username is None:
username = CONF.validation.image_ssh_user
# Set this with 'keypair' or others to log in with keypair or
# username/password.
if CONF.validation.auth_method == 'keypair':
password = None
if private_key is None:
private_key = self.keypair['private_key']
else:
password = CONF.validation.image_ssh_password
private_key = None
linux_client = remote_client.RemoteClient(ip_address, username,
pkey=private_key,
password=password)
try:
linux_client.validate_authentication()
except Exception as e:
message = ('Initializing SSH connection to %(ip)s failed. '
'Error: %(error)s' % {'ip': ip_address,
'error': e})
caller = test_utils.find_test_caller()
if caller:
message = '(%s) %s' % (caller, message)
LOG.exception(message)
self._log_console_output()
raise
return linux_client
def _log_console_output(self, servers=None):
if not CONF.compute_feature_enabled.console_output:
LOG.debug('Console output not supported, cannot log')
return
if not servers:
servers = self.servers_client.list_servers()
servers = servers['servers']
for server in servers:
try:
console_output = self.servers_client.get_console_output(
server['id'])['output']
LOG.debug('Console output for %s\nbody=\n%s',
server['id'], console_output)
except lib_exc.NotFound:
LOG.debug("Server %s disappeared(deleted) while looking "
"for the console log", server['id'])
def _log_net_info(self, exc):
# network debug is called as part of ssh init
if not isinstance(exc, lib_exc.SSHTimeout):
LOG.debug('Network information on a devstack host')
def ping_ip_address(self, ip_address, should_succeed=True,
ping_timeout=None, mtu=None):
timeout = ping_timeout or CONF.validation.ping_timeout
cmd = ['ping', '-c1', '-w1']
if mtu:
cmd += [
# don't fragment
'-M', 'do',
# ping receives just the size of ICMP payload
'-s', str(net_utils.get_ping_payload_size(mtu, 4))
]
cmd.append(ip_address)
def ping():
proc = subprocess.Popen(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
proc.communicate()
return (proc.returncode == 0) == should_succeed
caller = test_utils.find_test_caller()
LOG.debug('%(caller)s begins to ping %(ip)s in %(timeout)s sec and the'
' expected result is %(should_succeed)s', {
'caller': caller, 'ip': ip_address, 'timeout': timeout,
'should_succeed':
'reachable' if should_succeed else 'unreachable'
})
result = test_utils.call_until_true(ping, timeout, 1)
LOG.debug('%(caller)s finishes ping %(ip)s in %(timeout)s sec and the '
'ping result is %(result)s', {
'caller': caller, 'ip': ip_address, 'timeout': timeout,
'result': 'expected' if result else 'unexpected'
})
return result
def check_vm_connectivity(self, ip_address,
username=None,
private_key=None,
should_connect=True,
mtu=None):
"""Check server connectivity
:param ip_address: server to test against
:param username: server's ssh username
:param private_key: server's ssh private key to be used
:param should_connect: True/False indicates positive/negative test
positive - attempt ping and ssh
negative - attempt ping and fail if succeed
:param mtu: network MTU to use for connectivity validation
:raises: AssertError if the result of the connectivity check does
not match the value of the should_connect param
"""
if should_connect:
msg = "Timed out waiting for %s to become reachable" % ip_address
else:
msg = "ip address %s is reachable" % ip_address
self.assertTrue(self.ping_ip_address(ip_address,
should_succeed=should_connect,
mtu=mtu),
msg=msg)
if should_connect:
# no need to check ssh for negative connectivity
self.get_remote_client(ip_address, username, private_key)
def check_public_network_connectivity(self, ip_address, username,
private_key, should_connect=True,
msg=None, servers=None, mtu=None):
# The target login is assumed to have been configured for
# key-based authentication by cloud-init.
LOG.debug('checking network connections to IP %s with user: %s',
ip_address, username)
try:
self.check_vm_connectivity(ip_address,
username,
private_key,
should_connect=should_connect,
mtu=mtu)
except Exception:
ex_msg = 'Public network connectivity check failed'
if msg:
ex_msg += ": " + msg
LOG.exception(ex_msg)
self._log_console_output(servers)
raise
class NetworkScenarioTest(ScenarioTest):
"""Base class for network scenario tests.
This class provide helpers for network scenario tests, using the neutron
API. Helpers from ancestor which use the nova network API are overridden
with the neutron API.
This Class also enforces using Neutron instead of novanetwork.
Subclassed tests will be skipped if Neutron is not enabled
"""
credentials = ['primary', 'admin']
@classmethod
def skip_checks(cls):
super(NetworkScenarioTest, cls).skip_checks()
if not CONF.service_available.neutron:
raise cls.skipException('Neutron not available')
def _create_network(self, networks_client=None,
tenant_id=None,
namestart='network-smoke-',
port_security_enabled=True):
if not networks_client:
networks_client = self.networks_client
if not tenant_id:
tenant_id = networks_client.tenant_id
name = data_utils.rand_name(namestart)
network_kwargs = dict(name=name, tenant_id=tenant_id)
# Neutron disables port security by default so we have to check the
# config before trying to create the network with port_security_enabled
if CONF.network_feature_enabled.port_security:
network_kwargs['port_security_enabled'] = port_security_enabled
result = networks_client.create_network(**network_kwargs)
network = result['network']
self.assertEqual(network['name'], name)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
networks_client.delete_network,
network['id'])
return network
def _create_subnet(self, network, subnets_client=None,
routers_client=None, namestart='subnet-smoke',
**kwargs):
"""Create a subnet for the given network
within the cidr block configured for tenant networks.
"""
if not subnets_client:
subnets_client = self.subnets_client
if not routers_client:
routers_client = self.routers_client
def cidr_in_use(cidr, tenant_id):
"""Check cidr existence
:returns: True if subnet with cidr already exist in tenant
False else
"""
cidr_in_use = self.os_admin.subnets_client.list_subnets(
tenant_id=tenant_id, cidr=cidr)['subnets']
return len(cidr_in_use) != 0
ip_version = kwargs.pop('ip_version', 4)
if ip_version == 6:
tenant_cidr = netaddr.IPNetwork(
CONF.network.project_network_v6_cidr)
num_bits = CONF.network.project_network_v6_mask_bits
else:
tenant_cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
num_bits = CONF.network.project_network_mask_bits
result = None
str_cidr = None
# Repeatedly attempt subnet creation with sequential cidr
# blocks until an unallocated block is found.
for subnet_cidr in tenant_cidr.subnet(num_bits):
str_cidr = str(subnet_cidr)
if cidr_in_use(str_cidr, tenant_id=network['tenant_id']):
continue
subnet = dict(
name=data_utils.rand_name(namestart),
network_id=network['id'],
tenant_id=network['tenant_id'],
cidr=str_cidr,
ip_version=ip_version,
**kwargs
)
try:
result = subnets_client.create_subnet(**subnet)
break
except lib_exc.Conflict as e:
is_overlapping_cidr = 'overlaps with another subnet' in str(e)
if not is_overlapping_cidr:
raise
self.assertIsNotNone(result, 'Unable to allocate tenant network')
subnet = result['subnet']
self.assertEqual(subnet['cidr'], str_cidr)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
subnets_client.delete_subnet, subnet['id'])
return subnet
def _get_server_port_id_and_ip4(self, server, ip_addr=None):
ports = self.os_admin.ports_client.list_ports(
device_id=server['id'], fixed_ip=ip_addr)['ports']
# A port can have more than one IP address in some cases.
# If the network is dual-stack (IPv4 + IPv6), this port is associated
# with 2 subnets
p_status = ['ACTIVE']
# NOTE(vsaienko) With Ironic, instances live on separate hardware
# servers. Neutron does not bind ports for Ironic instances, as a
# result the port remains in the DOWN state.
# TODO(vsaienko) remove once bug: #1599836 is resolved.
if getattr(CONF.service_available, 'ironic', False):
p_status.append('DOWN')
port_map = [(p["id"], fxip["ip_address"])
for p in ports
for fxip in p["fixed_ips"]
if (netutils.is_valid_ipv4(fxip["ip_address"]) and
p['status'] in p_status)]
inactive = [p for p in ports if p['status'] != 'ACTIVE']
if inactive:
LOG.warning("Instance has ports that are not ACTIVE: %s", inactive)
self.assertNotEqual(0, len(port_map),
"No IPv4 addresses found in: %s" % ports)
self.assertEqual(len(port_map), 1,
"Found multiple IPv4 addresses: %s. "
"Unable to determine which port to target."
% port_map)
return port_map[0]
def _get_network_by_name(self, network_name):
net = self.os_admin.networks_client.list_networks(
name=network_name)['networks']
self.assertNotEqual(len(net), 0,
"Unable to get network by name: %s" % network_name)
return net[0]
def create_floating_ip(self, thing, external_network_id=None,
port_id=None, client=None):
"""Create a floating IP and associates to a resource/port on Neutron"""
if not external_network_id:
external_network_id = CONF.network.public_network_id
if not client:
client = self.floating_ips_client
if not port_id:
port_id, ip4 = self._get_server_port_id_and_ip4(thing)
else:
ip4 = None
result = client.create_floatingip(
floating_network_id=external_network_id,
port_id=port_id,
tenant_id=thing['tenant_id'],
fixed_ip_address=ip4
)
floating_ip = result['floatingip']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
client.delete_floatingip,
floating_ip['id'])
return floating_ip
def _associate_floating_ip(self, floating_ip, server):
port_id, _ = self._get_server_port_id_and_ip4(server)
kwargs = dict(port_id=port_id)
floating_ip = self.floating_ips_client.update_floatingip(
floating_ip['id'], **kwargs)['floatingip']
self.assertEqual(port_id, floating_ip['port_id'])
return floating_ip
def _disassociate_floating_ip(self, floating_ip):
""":param floating_ip: floating_ips_client.create_floatingip"""
kwargs = dict(port_id=None)
floating_ip = self.floating_ips_client.update_floatingip(
floating_ip['id'], **kwargs)['floatingip']
self.assertIsNone(floating_ip['port_id'])
return floating_ip
def check_floating_ip_status(self, floating_ip, status):
"""Verifies floatingip reaches the given status
:param dict floating_ip: floating IP dict to check status
:param status: target status
:raises: AssertionError if status doesn't match
"""
floatingip_id = floating_ip['id']
def refresh():
result = (self.floating_ips_client.
show_floatingip(floatingip_id)['floatingip'])
return status == result['status']
test_utils.call_until_true(refresh,
CONF.network.build_timeout,
CONF.network.build_interval)
floating_ip = self.floating_ips_client.show_floatingip(
floatingip_id)['floatingip']
self.assertEqual(status, floating_ip['status'],
message="FloatingIP: {fp} is at status: {cst}. "
"failed to reach status: {st}"
.format(fp=floating_ip, cst=floating_ip['status'],
st=status))
LOG.info("FloatingIP: {fp} is at status: {st}"
.format(fp=floating_ip, st=status))
def _check_tenant_network_connectivity(self, server,
username,
private_key,
should_connect=True,
servers_for_debug=None):
if not CONF.network.project_networks_reachable:
msg = 'Tenant networks not configured to be reachable.'
LOG.info(msg)
return
# The target login is assumed to have been configured for
# key-based authentication by cloud-init.
try:
for net_name, ip_addresses in server['addresses'].items():
for ip_address in ip_addresses:
self.check_vm_connectivity(ip_address['addr'],
username,
private_key,
should_connect=should_connect)
except Exception as e:
LOG.exception('Tenant network connectivity check failed')
self._log_console_output(servers_for_debug)
self._log_net_info(e)
raise
def _check_remote_connectivity(self, source, dest, should_succeed=True,
nic=None):
"""check ping server via source ssh connection
:param source: RemoteClient: an ssh connection from which to ping
:param dest: and IP to ping against
:param should_succeed: boolean should ping succeed or not
:param nic: specific network interface to ping from
:returns: boolean -- should_succeed == ping
:returns: ping is false if ping failed
"""
def ping_remote():
try:
source.ping_host(dest, nic=nic)
except lib_exc.SSHExecCommandFailed:
LOG.warning('Failed to ping IP: %s via a ssh connection '
'from: %s.', dest, source.ssh_client.host)
return not should_succeed
return should_succeed
return test_utils.call_until_true(ping_remote,
CONF.validation.ping_timeout,
1)
def _create_security_group(self, security_group_rules_client=None,
tenant_id=None,
namestart='secgroup-smoke',
security_groups_client=None):
if security_group_rules_client is None:
security_group_rules_client = self.security_group_rules_client
if security_groups_client is None:
security_groups_client = self.security_groups_client
if tenant_id is None:
tenant_id = security_groups_client.tenant_id
secgroup = self._create_empty_security_group(
namestart=namestart, client=security_groups_client,
tenant_id=tenant_id)
# Add rules to the security group
rules = self._create_loginable_secgroup_rule(
security_group_rules_client=security_group_rules_client,
secgroup=secgroup,
security_groups_client=security_groups_client)
for rule in rules:
self.assertEqual(tenant_id, rule['tenant_id'])
self.assertEqual(secgroup['id'], rule['security_group_id'])
return secgroup
def _create_empty_security_group(self, client=None, tenant_id=None,
namestart='secgroup-smoke'):
"""Create a security group without rules.
Default rules will be created:
- IPv4 egress to any
- IPv6 egress to any
:param tenant_id: secgroup will be created in this tenant
:returns: the created security group
"""
if client is None:
client = self.security_groups_client
if not tenant_id:
tenant_id = client.tenant_id
sg_name = data_utils.rand_name(namestart)
sg_desc = sg_name + " description"
sg_dict = dict(name=sg_name,
description=sg_desc)
sg_dict['tenant_id'] = tenant_id
result = client.create_security_group(**sg_dict)
secgroup = result['security_group']
self.assertEqual(secgroup['name'], sg_name)
self.assertEqual(tenant_id, secgroup['tenant_id'])
self.assertEqual(secgroup['description'], sg_desc)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
client.delete_security_group, secgroup['id'])
return secgroup
def _default_security_group(self, client=None, tenant_id=None):
"""Get default secgroup for given tenant_id.
:returns: default secgroup for given tenant
"""
if client is None:
client = self.security_groups_client
if not tenant_id:
tenant_id = client.tenant_id
sgs = [
sg for sg in list(client.list_security_groups().values())[0]
if sg['tenant_id'] == tenant_id and sg['name'] == 'default'
]
msg = "No default security group for tenant %s." % (tenant_id)
self.assertGreater(len(sgs), 0, msg)
return sgs[0]
def _create_security_group_rule(self, secgroup=None,
sec_group_rules_client=None,
tenant_id=None,
security_groups_client=None, **kwargs):
"""Create a rule from a dictionary of rule parameters.
Create a rule in a secgroup. if secgroup not defined will search for
default secgroup in tenant_id.
:param secgroup: the security group.
:param tenant_id: if secgroup not passed -- the tenant in which to
search for default secgroup
:param kwargs: a dictionary containing rule parameters:
for example, to allow incoming ssh:
rule = {
direction: 'ingress'
protocol:'tcp',
port_range_min: 22,
port_range_max: 22
}
"""
if sec_group_rules_client is None:
sec_group_rules_client = self.security_group_rules_client
if security_groups_client is None:
security_groups_client = self.security_groups_client
if not tenant_id:
tenant_id = security_groups_client.tenant_id
if secgroup is None:
secgroup = self._default_security_group(
client=security_groups_client, tenant_id=tenant_id)
ruleset = dict(security_group_id=secgroup['id'],
tenant_id=secgroup['tenant_id'])
ruleset.update(kwargs)
sg_rule = sec_group_rules_client.create_security_group_rule(**ruleset)
sg_rule = sg_rule['security_group_rule']
self.assertEqual(secgroup['tenant_id'], sg_rule['tenant_id'])
self.assertEqual(secgroup['id'], sg_rule['security_group_id'])
return sg_rule
def _create_loginable_secgroup_rule(self, security_group_rules_client=None,
secgroup=None,
security_groups_client=None):
"""Create loginable security group rule
This function will create:
1. egress and ingress tcp port 22 allow rule in order to allow ssh
access for ipv4.
2. egress and ingress ipv6 icmp allow rule, in order to allow icmpv6.
3. egress and ingress ipv4 icmp allow rule, in order to allow icmpv4.
"""
if security_group_rules_client is None:
security_group_rules_client = self.security_group_rules_client
if security_groups_client is None:
security_groups_client = self.security_groups_client
rules = []
rulesets = [
dict(
# ssh
protocol='tcp',
port_range_min=22,
port_range_max=22,
),
dict(
# ping
protocol='icmp',
),
dict(
# ipv6-icmp for ping6
protocol='icmp',
ethertype='IPv6',
)
]
sec_group_rules_client = security_group_rules_client
for ruleset in rulesets:
for r_direction in ['ingress', 'egress']:
ruleset['direction'] = r_direction
try:
sg_rule = self._create_security_group_rule(
sec_group_rules_client=sec_group_rules_client,
secgroup=secgroup,
security_groups_client=security_groups_client,
**ruleset)
except lib_exc.Conflict as ex:
# if rule already exist - skip rule and continue
msg = 'Security group rule already exists'
if msg not in ex._error_string:
raise ex
else:
self.assertEqual(r_direction, sg_rule['direction'])
rules.append(sg_rule)
return rules
def _get_router(self, client=None, tenant_id=None):
"""Retrieve a router for the given tenant id.
If a public router has been configured, it will be returned.
If a public router has not been configured, but a public
network has, a tenant router will be created and returned that
routes traffic to the public network.
"""
if not client:
client = self.routers_client
if not tenant_id:
tenant_id = client.tenant_id
router_id = CONF.network.public_router_id
network_id = CONF.network.public_network_id
if router_id:
body = client.show_router(router_id)
return body['router']
elif network_id:
router = self._create_router(client, tenant_id)
kwargs = {'external_gateway_info': dict(network_id=network_id)}
router = client.update_router(router['id'], **kwargs)['router']
return router
else:
raise Exception("Neither of 'public_router_id' or "
"'public_network_id' has been defined.")
def _create_router(self, client=None, tenant_id=None,
namestart='router-smoke'):
if not client:
client = self.routers_client
if not tenant_id:
tenant_id = client.tenant_id
name = data_utils.rand_name(namestart)
result = client.create_router(name=name,
admin_state_up=True,
tenant_id=tenant_id)
router = result['router']
self.assertEqual(router['name'], name)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
client.delete_router,
router['id'])
return router
def _update_router_admin_state(self, router, admin_state_up):
kwargs = dict(admin_state_up=admin_state_up)
router = self.routers_client.update_router(
router['id'], **kwargs)['router']
self.assertEqual(admin_state_up, router['admin_state_up'])
def create_networks(self, networks_client=None,
routers_client=None, subnets_client=None,
tenant_id=None, dns_nameservers=None,
port_security_enabled=True):
"""Create a network with a subnet connected to a router.
The baremetal driver is a special case since all nodes are
on the same shared network.
:param tenant_id: id of tenant to create resources in.
:param dns_nameservers: list of dns servers to send to subnet.
:returns: network, subnet, router
"""
if CONF.network.shared_physical_network:
# NOTE(Shrews): This exception is for environments where tenant
# credential isolation is available, but network separation is
# not (the current baremetal case). Likely can be removed when
# test account mgmt is reworked:
# https://blueprints.launchpad.net/tempest/+spec/test-accounts
if not CONF.compute.fixed_network_name:
m = 'fixed_network_name must be specified in config'
raise lib_exc.InvalidConfiguration(m)
network = self._get_network_by_name(
CONF.compute.fixed_network_name)
router = None
subnet = None
else:
network = self._create_network(
networks_client=networks_client,
tenant_id=tenant_id,
port_security_enabled=port_security_enabled)
router = self._get_router(client=routers_client,
tenant_id=tenant_id)
subnet_kwargs = dict(network=network,
subnets_client=subnets_client,
routers_client=routers_client)
# use explicit check because empty list is a valid option
if dns_nameservers is not None:
subnet_kwargs['dns_nameservers'] = dns_nameservers
subnet = self._create_subnet(**subnet_kwargs)
if not routers_client:
routers_client = self.routers_client
router_id = router['id']
routers_client.add_router_interface(router_id,
subnet_id=subnet['id'])
# save a cleanup job to remove this association between
# router and subnet
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
routers_client.remove_router_interface, router_id,
subnet_id=subnet['id'])
return network, subnet, router

View File

@ -0,0 +1,303 @@
# Copyright (c) 2016 Juniper Networks
# 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 testscenarios
from oslo_log import log as logging
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from neutron_tempest_plugin.fwaas.scenario import fwaas_v2_base as base
CONF = config.CONF
LOG = logging.getLogger(__name__)
load_tests = testscenarios.load_tests_apply_scenarios
class TestFWaaS_v2(base.FWaaSScenarioTest_V2):
"""Config Requirement in tempest.conf:
- project_network_cidr_bits- specifies the subnet range for each network
- project_network_cidr
- public_network_id
"""
def setUp(self):
LOG.debug("Initializing FWaaSScenarioTest Setup")
super(TestFWaaS_v2, self).setUp()
required_exts = ['fwaas_v2', 'security-group', 'router']
# if self.router_insertion:
# required_exts.append('fwaasrouterinsertion')
for ext in required_exts:
if not utils.is_extension_enabled(ext, 'network'):
msg = "%s Extension not enabled." % ext
raise self.skipException(msg)
LOG.debug("FWaaSScenarioTest Setup done.")
def _create_server(self, network, security_group=None):
keys = self.create_keypair()
kwargs = {}
if security_group is not None:
kwargs['security_groups'] = [{'name': security_group['name']}]
server = self.create_server(
key_name=keys['name'],
networks=[{'uuid': network['id']}],
wait_until='ACTIVE',
**kwargs)
return server, keys
def _check_connectivity_between_internal_networks(
self, floating_ip1, keys1, network2, server2, should_connect=True):
internal_ips = (p['fixed_ips'][0]['ip_address'] for p in
self.os_admin.ports_client.list_ports(
tenant_id=server2['tenant_id'],
network_id=network2['id'])['ports']
if p['device_owner'].startswith('network'))
self._check_server_connectivity(
floating_ip1, keys1, internal_ips, should_connect)
def _check_server_connectivity(self, floating_ip, keys1, address_list,
should_connect=True):
ip_address = floating_ip['floating_ip_address']
private_key = keys1
ssh_source = self.get_remote_client(
ip_address, private_key=private_key)
for remote_ip in address_list:
if should_connect:
msg = ("Timed out waiting for %s to become "
"reachable") % remote_ip
else:
msg = "ip address %s is reachable" % remote_ip
try:
self.assertTrue(self._check_remote_connectivity
(ssh_source, remote_ip, should_connect),
msg)
except Exception:
LOG.exception("Unable to access {dest} via ssh to "
"floating-ip {src}".format(dest=remote_ip,
src=floating_ip))
raise
def _check_remote_connectivity(self, source, dest, should_succeed=True,
nic=None):
"""check ping server via source ssh connection
:param source: RemoteClient: an ssh connection from which to ping
:param dest: and IP to ping against
:param should_succeed: boolean should ping succeed or not
:param nic: specific network interface to ping from
:returns: boolean -- should_succeed == ping
:returns: ping is false if ping failed
"""
def ping_remote():
try:
source.ping_host(dest, nic=nic)
except lib_exc.SSHExecCommandFailed:
LOG.warning('Failed to ping IP: %s via a ssh connection '
'from: %s.', dest, source.ssh_client.host)
return not should_succeed
return should_succeed
return test_utils.call_until_true(ping_remote,
CONF.validation.ping_timeout,
1)
def _add_router_interface(self, router_id, subnet_id):
resp = self.routers_client.add_router_interface(
router_id, subnet_id=subnet_id)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.routers_client.remove_router_interface, router_id,
subnet_id=subnet_id)
return resp
def _create_network_subnet(self):
network = self._create_network()
subnet_kwargs = dict(network=network)
subnet = self._create_subnet(**subnet_kwargs)
return network, subnet
def _create_test_server(self, network, security_group):
pub_network_id = CONF.network.public_network_id
server, keys = self._create_server(
network, security_group=security_group)
private_key = keys['private_key']
server_floating_ip = self.create_floating_ip(server, pub_network_id)
fixed_ip = list(server['addresses'].values())[0][0]['addr']
return server, private_key, fixed_ip, server_floating_ip
def _create_topology(self):
"""Topology diagram:
+--------+ +-------------+
|"server"| | "subnet" |
| VM-1 +-------------+ "network-1" |
+--------+ +----+--------+
|
| router interface port
+----+-----+
| "router" |
+----+-----+
| router interface port
|
|
+--------+ +-------------+
|"server"| | "subnet" |
| VM-2 +-------------+ "network-2" |
+--------+ +----+--------+
"""
LOG.debug('Starting Topology Creation')
resp = {}
# Create Network1 and Subnet1.
network1, subnet1 = self._create_network_subnet()
resp['network1'] = network1
resp['subnet1'] = subnet1
# Create Network2 and Subnet2.
network2, subnet2 = self._create_network_subnet()
resp['network2'] = network2
resp['subnet2'] = subnet2
# Create a router and attach Network1, Network2 and External Networks
# to it.
router = self._create_router(namestart='SCENARIO-TEST-ROUTER')
pub_network_id = CONF.network.public_network_id
kwargs = {'external_gateway_info': dict(network_id=pub_network_id)}
router = self.routers_client.update_router(
router['id'], **kwargs)['router']
router_id = router['id']
resp_add_intf = self._add_router_interface(
router_id, subnet_id=subnet1['id'])
router_portid_1 = resp_add_intf['port_id']
resp_add_intf = self._add_router_interface(
router_id, subnet_id=subnet2['id'])
router_portid_2 = resp_add_intf['port_id']
resp['router'] = router
resp['router_portid_1'] = router_portid_1
resp['router_portid_2'] = router_portid_2
# Create a VM on each of the network and assign it a floating IP.
security_group = self._create_security_group()
server1, private_key1, server_fixed_ip_1, server_floating_ip_1 = (
self._create_test_server(network1, security_group))
server2, private_key2, server_fixed_ip_2, server_floating_ip_2 = (
self._create_test_server(network2, security_group))
resp['server1'] = server1
resp['private_key1'] = private_key1
resp['server_fixed_ip_1'] = server_fixed_ip_1
resp['server_floating_ip_1'] = server_floating_ip_1
resp['server2'] = server2
resp['private_key2'] = private_key2
resp['server_fixed_ip_2'] = server_fixed_ip_2
resp['server_floating_ip_2'] = server_floating_ip_2
return resp
@decorators.idempotent_id('77fdf3ea-82c1-453d-bfec-f7efe335625d')
def test_icmp_reachability_scenarios(self):
topology = self._create_topology()
ssh_login = CONF.validation.image_ssh_user
self.check_vm_connectivity(
ip_address=topology['server_floating_ip_1']['floating_ip_address'],
username=ssh_login,
private_key=topology['private_key1'])
self.check_vm_connectivity(
ip_address=topology['server_floating_ip_2']['floating_ip_address'],
username=ssh_login,
private_key=topology['private_key2'])
# Scenario 1: Add allow ICMP rules between the two VMs.
fw_allow_icmp_rule = self.create_firewall_rule(action="allow",
protocol="icmp")
fw_allow_ssh_rule = self.create_firewall_rule(action="allow",
protocol="tcp",
destination_port=22)
fw_policy = self.create_firewall_policy(
firewall_rules=[fw_allow_icmp_rule['id'], fw_allow_ssh_rule['id']])
fw_group = self.create_firewall_group(
ports=[
topology['router_portid_1'],
topology['router_portid_2']],
ingress_firewall_policy_id=fw_policy['id'],
egress_firewall_policy_id=fw_policy['id'])
self._wait_firewall_group_ready(fw_group['id'])
LOG.debug('fw_allow_icmp_rule: %s\nfw_allow_ssh_rule: %s\n'
'fw_policy: %s\nfw_group: %s\n',
fw_allow_icmp_rule, fw_allow_ssh_rule, fw_policy, fw_group)
# Check the connectivity between VM1 and VM2. It should Pass.
self._check_server_connectivity(
topology['server_floating_ip_1'],
topology['private_key1'],
address_list=[topology['server_fixed_ip_2']],
should_connect=True)
# Scenario 2: Now remove the allow_icmp rule add a deny_icmp rule and
# check that ICMP gets blocked
fw_deny_icmp_rule = self.create_firewall_rule(action="deny",
protocol="icmp")
self.remove_firewall_rule_from_policy_and_wait(
firewall_group_id=fw_group['id'],
firewall_rule_id=fw_allow_icmp_rule['id'],
firewall_policy_id=fw_policy['id'])
self.insert_firewall_rule_in_policy_and_wait(
firewall_group_id=fw_group['id'],
firewall_rule_id=fw_deny_icmp_rule['id'],
firewall_policy_id=fw_policy['id'])
self._check_server_connectivity(
topology['server_floating_ip_1'],
topology['private_key1'],
address_list=[topology['server_fixed_ip_2']],
should_connect=False)
# Scenario 3: Create a rule allowing ICMP only from server_fixed_ip_1
# to server_fixed_ip_2 and check that traffic from opposite direction
# is blocked.
fw_allow_unidirectional_icmp_rule = self.create_firewall_rule(
action="allow", protocol="icmp",
source_ip_address=topology['server_fixed_ip_1'],
destination_ip_address=topology['server_fixed_ip_2'])
self.remove_firewall_rule_from_policy_and_wait(
firewall_group_id=fw_group['id'],
firewall_rule_id=fw_deny_icmp_rule['id'],
firewall_policy_id=fw_policy['id'])
self.insert_firewall_rule_in_policy_and_wait(
firewall_group_id=fw_group['id'],
firewall_rule_id=fw_allow_unidirectional_icmp_rule['id'],
firewall_policy_id=fw_policy['id'])
self._check_server_connectivity(
topology['server_floating_ip_1'],
topology['private_key1'],
address_list=[topology['server_fixed_ip_2']],
should_connect=True)
self._check_server_connectivity(
topology['server_floating_ip_2'],
topology['private_key2'],
address_list=[topology['server_fixed_ip_1']],
should_connect=False)
# Disassociate ports of this firewall group for cleanup resources
self.firewall_groups_client.update_firewall_group(
fw_group['id'], ports=[])

View File

@ -0,0 +1,123 @@
# Copyright (c) 2016
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.network import base
class FirewallGroupsClient(base.BaseNetworkClient):
def create_firewall_group(self, **kwargs):
uri = '/fwaas/firewall_groups'
post_data = {'firewall_group': kwargs}
return self.create_resource(uri, post_data)
def update_firewall_group(self, firewall_group_id, **kwargs):
uri = '/fwaas/firewall_groups/%s' % firewall_group_id
post_data = {'firewall_group': kwargs}
return self.update_resource(uri, post_data)
def show_firewall_group(self, firewall_group_id, **fields):
uri = '/fwaas/firewall_groups/%s' % firewall_group_id
return self.show_resource(uri, **fields)
def delete_firewall_group(self, firewall_group_id):
uri = '/fwaas/firewall_groups/%s' % firewall_group_id
return self.delete_resource(uri)
def list_firewall_groups(self, **filters):
uri = '/fwaas/firewall_groups'
return self.list_resources(uri, **filters)
def is_resource_deleted(self, id):
try:
self.show_firewall_group(id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'firewall_group'
class FirewallRulesClient(base.BaseNetworkClient):
def create_firewall_rule(self, **kwargs):
uri = '/fwaas/firewall_rules'
post_data = {'firewall_rule': kwargs}
return self.create_resource(uri, post_data)
def update_firewall_rule(self, firewall_rule_id, **kwargs):
uri = '/fwaas/firewall_rules/%s' % firewall_rule_id
post_data = {'firewall_rule': kwargs}
return self.update_resource(uri, post_data)
def show_firewall_rule(self, firewall_rule_id, **fields):
uri = '/fwaas/firewall_rules/%s' % firewall_rule_id
return self.show_resource(uri, **fields)
def delete_firewall_rule(self, firewall_rule_id):
uri = '/fwaas/firewall_rules/%s' % firewall_rule_id
return self.delete_resource(uri)
def list_firewall_rules(self, **filters):
uri = '/fwaas/firewall_rules'
return self.list_resources(uri, **filters)
class FirewallPoliciesClient(base.BaseNetworkClient):
def create_firewall_policy(self, **kwargs):
uri = '/fwaas/firewall_policies'
post_data = {'firewall_policy': kwargs}
return self.create_resource(uri, post_data)
def update_firewall_policy(self, firewall_policy_id, **kwargs):
uri = '/fwaas/firewall_policies/%s' % firewall_policy_id
post_data = {'firewall_policy': kwargs}
return self.update_resource(uri, post_data)
def show_firewall_policy(self, firewall_policy_id, **fields):
uri = '/fwaas/firewall_policies/%s' % firewall_policy_id
return self.show_resource(uri, **fields)
def delete_firewall_policy(self, firewall_policy_id):
uri = '/fwaas/firewall_policies/%s' % firewall_policy_id
return self.delete_resource(uri)
def list_firewall_policies(self, **filters):
uri = '/fwaas/firewall_policies'
return self.list_resources(uri, **filters)
def insert_firewall_rule_in_policy(self, firewall_policy_id,
firewall_rule_id, insert_after='',
insert_before=''):
uri = '/fwaas/firewall_policies/%s/insert_rule' % firewall_policy_id
data = {
'firewall_rule_id': firewall_rule_id,
'insert_after': insert_after,
'insert_before': insert_before,
}
return self.update_resource(uri, data)
def remove_firewall_rule_from_policy(self, firewall_policy_id,
firewall_rule_id):
uri = '/fwaas/firewall_policies/%s/remove_rule' % firewall_policy_id
data = {
'firewall_rule_id': firewall_rule_id,
}
return self.update_resource(uri, data)