arif/zaza-openstack-tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #17 from openstack-charmers/feature/neutron-api

Browse Source 
Add new Neutron API Tests
This commit is contained in:
David Ames
2019-06-26 10:15:12 -07:00
committed by GitHub
parent 18f5f87255 83e4ca7247
commit bf93335017
2 changed files with 290 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
zaza/openstack/charm_tests/neutron/tests.py Normal file
View File

@@ -0,0 +1,274 @@
#!/usr/bin/env python3
# Copyright 2018 Canonical Ltd.
#
# 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.
"""Encapsulating `neutron-openvswitch` testing."""
import logging
import unittest
import zaza
import zaza.openstack.charm_tests.glance.setup as glance_setup
import zaza.openstack.charm_tests.nova.utils as nova_utils
import zaza.openstack.charm_tests.test_utils as test_utils
import zaza.openstack.configure.guest as guest
import zaza.openstack.utilities.openstack as openstack_utils
class NeutronApiTest(test_utils.OpenStackBaseTest):
"""Test basic Neutron API Charm functionality."""
def test_900_restart_on_config_change(self):
"""Checking restart happens on config change.
Change disk format and assert then change propagates to the correct
file and that services are restarted as a result
"""
# Expected default and alternate values
current_value = zaza.model.get_application_config(
'neutron-api')['debug']['value']
new_value = str(not bool(current_value)).title()
current_value = str(current_value).title()
set_default = {'debug': current_value}
set_alternate = {'debug': new_value}
default_entry = {'DEFAULT': {'debug': [current_value]}}
alternate_entry = {'DEFAULT': {'debug': [new_value]}}
# Config file affected by juju set config change
conf_file = '/etc/neutron/neutron.conf'
# Make config change, check for service restarts
logging.info(
'Setting verbose on neutron-api {}'.format(set_alternate))
bionic_stein = openstack_utils.get_os_release('bionic_stein')
if openstack_utils.get_os_release() >= bionic_stein:
pgrep_full = True
else:
pgrep_full = False
self.restart_on_changed(
conf_file,
set_default,
set_alternate,
default_entry,
alternate_entry,
['neutron-server'],
pgrep_full=pgrep_full)
def test_901_pause_resume(self):
"""Run pause and resume tests.
Pause service and check services are stopped then resume and check
they are started
"""
bionic_stein = openstack_utils.get_os_release('bionic_stein')
if openstack_utils.get_os_release() >= bionic_stein:
pgrep_full = True
else:
pgrep_full = False
with self.pause_resume(
["neutron-server", "apache2", "haproxy"],
pgrep_full=pgrep_full):
logging.info("Testing pause resume")
class SecurityTest(test_utils.OpenStackBaseTest):
"""Neutron APIsecurity tests tests."""
def test_security_checklist(self):
"""Verify expected state with security-checklist."""
# Changes fixing the below expected failures will be made following
# this initial work to get validation in. There will be bugs targeted
# to each one and resolved independently where possible.
expected_failures = [
'validate-enables-tls',
'validate-uses-tls-for-keystone',
]
expected_passes = [
'validate-file-ownership',
'validate-file-permissions',
'validate-uses-keystone',
]
for unit in zaza.model.get_units('neutron-api',
model_name=self.model_name):
logging.info('Running `security-checklist` action'
' on unit {}'.format(unit.entity_id))
test_utils.audit_assertions(
zaza.model.run_action(
unit.entity_id,
'security-checklist',
model_name=self.model_name,
action_params={}),
expected_passes,
expected_failures,
expected_to_pass=False)
class NeutronNetworkingTest(unittest.TestCase):
"""Ensure that openstack instances have valid networking."""
RESOURCE_PREFIX = 'zaza-neutrontests'
@classmethod
def setUpClass(cls):
"""Run class setup for running Neutron API Networking tests."""
cls.keystone_session = (
openstack_utils.get_overcloud_keystone_session())
cls.nova_client = (
openstack_utils.get_nova_session_client(cls.keystone_session))
@classmethod
def tearDown(cls):
"""Remove test resources."""
logging.info('Running teardown')
for server in cls.nova_client.servers.list():
if server.name.startswith(cls.RESOURCE_PREFIX):
openstack_utils.delete_resource(
cls.nova_client.servers,
server.id,
msg="server")
def test_instances_have_networking(self):
"""Validate North/South and East/West networking."""
guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-1'.format(self.RESOURCE_PREFIX))
guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-2'.format(self.RESOURCE_PREFIX))
instance_1 = self.nova_client.servers.find(
name='{}-ins-1'.format(self.RESOURCE_PREFIX))
instance_2 = self.nova_client.servers.find(
name='{}-ins-2'.format(self.RESOURCE_PREFIX))
def verify(stdin, stdout, stderr):
"""Validate that the SSH command exited 0."""
self.assertEqual(stdout.channel.recv_exit_status(), 0)
# Verify network from 1 to 2
self.validate_instance_can_reach_other(instance_1, instance_2, verify)
# Verify network from 2 to 1
self.validate_instance_can_reach_other(instance_2, instance_1, verify)
# Validate tenant to external network routing
self.validate_instance_can_reach_router(instance_1, verify)
self.validate_instance_can_reach_router(instance_2, verify)
def validate_instance_can_reach_other(self,
instance_1,
instance_2,
verify):
"""
Validate that an instance can reach a fixed and floating of another.
:param instance_1: The instance to check networking from
:type instance_1: nova_client.Server
:param instance_2: The instance to check networking from
:type instance_2: nova_client.Server
"""
floating_1 = floating_ips_from_instance(instance_1)[0]
floating_2 = floating_ips_from_instance(instance_2)[0]
address_2 = fixed_ips_from_instance(instance_2)[0]
username = guest.boot_tests['bionic']['username']
password = guest.boot_tests['bionic'].get('password')
privkey = openstack_utils.get_private_key(nova_utils.KEYPAIR_NAME)
openstack_utils.ssh_command(
username, floating_1, 'instance-1',
'ping -c 1 {}'.format(address_2),
password=password, privkey=privkey, verify=verify)
openstack_utils.ssh_command(
username, floating_1, 'instance-1',
'ping -c 1 {}'.format(floating_2),
password=password, privkey=privkey, verify=verify)
def validate_instance_can_reach_router(self, instance, verify):
"""
Validate that an instance can reach it's primary gateway.
We make the assumption that the router's IP is 192.168.0.1
as that's the network that is setup in
neutron.setup.basic_overcloud_network which is used in all
Zaza Neutron validations.
:param instance: The instance to check networking from
:type instance: nova_client.Server
"""
address = floating_ips_from_instance(instance)[0]
username = guest.boot_tests['bionic']['username']
password = guest.boot_tests['bionic'].get('password')
privkey = openstack_utils.get_private_key(nova_utils.KEYPAIR_NAME)
openstack_utils.ssh_command(
username, address, 'instance', 'ping -c 1 192.168.0.1',
password=password, privkey=privkey, verify=verify)
pass
def floating_ips_from_instance(instance):
"""
Retrieve floating IPs from an instance.
:param instance: The instance to fetch floating IPs from
:type instance: nova_client.Server
:returns: A list of floating IPs for the specified server
:rtype: list[str]
"""
return ips_from_instance(instance, 'floating')
def fixed_ips_from_instance(instance):
"""
Retrieve fixed IPs from an instance.
:param instance: The instance to fetch fixed IPs from
:type instance: nova_client.Server
:returns: A list of fixed IPs for the specified server
:rtype: list[str]
"""
return ips_from_instance(instance, 'fixed')
def ips_from_instance(instance, ip_type):
"""
Retrieve IPs of a certain type from an instance.
:param instance: The instance to fetch IPs from
:type instance: nova_client.Server
:param ip_type: the type of IP to fetch, floating or fixed
:type ip_type: str
:returns: A list of IPs for the specified server
:rtype: list[str]
"""
if ip_type not in ['floating', 'fixed']:
raise RuntimeError(
"Only 'floating' and 'fixed' are valid IP types to search for"
)
return list([
ip['addr'] for ip in instance.addresses['private']
if ip['OS-EXT-IPS:type'] == ip_type])

22
zaza/openstack/charm_tests/test_utils.py
View File

@@ -168,7 +168,7 @@ class OpenStackBaseTest(unittest.TestCase):
model.block_until_all_units_idle()
def restart_on_changed(self, config_file, default_config, alternate_config,
default_entry, alternate_entry, services):
default_entry, alternate_entry, services, pgrep_full=False):
"""Run restart on change tests.
Test that changing config results in config file being updates and
@@ -189,6 +189,9 @@ class OpenStackBaseTest(unittest.TestCase):
:param services: Services expected to be restarted when config_file is
changed.
:type services: list
:param pgrep_full: Should pgrep be used rather than pidof to identify
a service.
:type pgrep_full: bool
"""
# lead_unit is only useed to grab a timestamp, the assumption being
# that all the units times are in sync.
@@ -215,7 +218,8 @@ class OpenStackBaseTest(unittest.TestCase):
self.application_name,
mtime,
services,
model_name=self.model_name)
model_name=self.model_name,
pgrep_full=pgrep_full)
logging.debug(
'Waiting for updates to propagate to '.format(config_file))
@@ -226,7 +230,7 @@ class OpenStackBaseTest(unittest.TestCase):
model_name=self.model_name)
@contextlib.contextmanager
def pause_resume(self, services):
def pause_resume(self, services, pgrep_full=False):
"""Run Pause and resume tests.
Pause and then resume a unit checking that services are in the
@@ -235,12 +239,16 @@ class OpenStackBaseTest(unittest.TestCase):
:param services: Services expected to be restarted when config_file is
changed.
:type services: list
:param pgrep_full: Should pgrep be used rather than pidof to identify
a service.
:type pgrep_full: bool
"""
model.block_until_service_status(
self.lead_unit,
services,
'running',
model_name=self.model_name)
model_name=self.model_name,
pgrep_full=pgrep_full)
model.block_until_unit_wl_status(
self.lead_unit,
'active',
@@ -258,7 +266,8 @@ class OpenStackBaseTest(unittest.TestCase):
self.lead_unit,
services,
'stopped',
model_name=self.model_name)
model_name=self.model_name,
pgrep_full=pgrep_full)
yield
model.run_action(
self.lead_unit,
@@ -273,4 +282,5 @@ class OpenStackBaseTest(unittest.TestCase):
self.lead_unit,
services,
'running',
model_name=self.model_name)
model_name=self.model_name,
pgrep_full=pgrep_full)