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

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
camille.rodriguez
2020-07-24 09:15:43 -05:00
parent 692853e242 59d68e49a5
commit e85c6c784c
34 changed files with 905 additions and 823 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
unit_tests/charm_tests/test_utils.py
View File

@@ -30,3 +30,15 @@ class TestOpenStackBaseTest(unittest.TestCase):
MyTestClass.setUpClass('foo', 'bar')
_setUpClass.assert_called_with('foo', 'bar')
class TestUtils(unittest.TestCase):
def test_format_addr(self):
self.assertEquals('1.2.3.4', test_utils.format_addr('1.2.3.4'))
self.assertEquals(
'[2001:db8::42]', test_utils.format_addr('2001:db8::42'))
with self.assertRaises(ValueError):
test_utils.format_addr('999.999.999.999')
with self.assertRaises(ValueError):
test_utils.format_addr('2001:db8::g')

312
unit_tests/utilities/test_zaza_utilities_juju.py
View File

@@ -1,312 +0,0 @@
# 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.
import mock
import unit_tests.utils as ut_utils
from zaza.openstack.utilities import juju as juju_utils
class TestJujuUtils(ut_utils.BaseTestCase):
def setUp(self):
super(TestJujuUtils, self).setUp()
# Juju Status Object and data
self.key = "instance-id"
self.key_data = "machine-uuid"
self.machine = "1"
self.machine_data = {self.key: self.key_data}
self.unit = "app/1"
self.unit_data = {"machine": self.machine}
self.application = "app"
self.application_data = {"units": {self.unit: self.unit_data}}
self.subordinate_application = "subordinate_application"
self.subordinate_application_data = {
"subordinate-to": [self.application]}
self.juju_status = mock.MagicMock()
self.juju_status.name = "juju_status_object"
self.juju_status.applications.get.return_value = self.application_data
self.juju_status.machines.get.return_value = self.machine_data
# Model
self.patch_object(juju_utils, "model")
self.model_name = "model-name"
self.model.get_juju_model.return_value = self.model_name
self.model.get_status.return_value = self.juju_status
self.run_output = {"Code": "0", "Stderr": "", "Stdout": "RESULT"}
self.error_run_output = {"Code": "1", "Stderr": "ERROR", "Stdout": ""}
self.model.run_on_unit.return_value = self.run_output
# Clouds
self.cloud_name = "FakeCloudName"
self.cloud_type = "FakeCloudType"
self.clouds = {
"clouds":
{self.cloud_name:
{"type": self.cloud_type}}}
# Controller
self.patch_object(juju_utils, "controller")
self.controller.get_cloud.return_value = self.cloud_name
def test_get_application_status(self):
self.patch_object(juju_utils, "get_full_juju_status")
self.get_full_juju_status.return_value = self.juju_status
# Full status juju object return
self.assertEqual(
juju_utils.get_application_status(), self.juju_status)
self.get_full_juju_status.assert_called_once()
# Application only dictionary return
self.assertEqual(
juju_utils.get_application_status(application=self.application),
self.application_data)
# Unit no application dictionary return
self.assertEqual(
juju_utils.get_application_status(unit=self.unit),
self.unit_data)
def test_get_cloud_configs(self):
self.patch_object(juju_utils.Path, "home")
self.patch_object(juju_utils.generic_utils, "get_yaml_config")
self.get_yaml_config.return_value = self.clouds
# All the cloud configs
self.assertEqual(juju_utils.get_cloud_configs(), self.clouds)
# With cloud specified
self.assertEqual(juju_utils.get_cloud_configs(self.cloud_name),
self.clouds["clouds"][self.cloud_name])
def test_get_full_juju_status(self):
self.assertEqual(juju_utils.get_full_juju_status(), self.juju_status)
self.model.get_status.assert_called_once_with(model_name=None)
def test_get_machines_for_application(self):
self.patch_object(juju_utils, "get_application_status")
self.get_application_status.return_value = self.application_data
# Machine data
self.assertEqual(
next(juju_utils.get_machines_for_application(self.application)),
self.machine)
self.get_application_status.assert_called_once()
# Subordinate application has no units
def _get_application_status(application, model_name=None):
_apps = {
self.application: self.application_data,
self.subordinate_application:
self.subordinate_application_data}
return _apps[application]
self.get_application_status.side_effect = _get_application_status
self.assertEqual(
next(juju_utils.get_machines_for_application(
self.subordinate_application)),
self.machine)
def test_get_unit_name_from_host_name(self):
unit_mock1 = mock.MagicMock()
unit_mock1.data = {'machine-id': 12}
unit_mock1.entity_id = 'myapp/2'
unit_mock2 = mock.MagicMock()
unit_mock2.data = {'machine-id': 15}
unit_mock2.entity_id = 'myapp/5'
self.model.get_units.return_value = [unit_mock1, unit_mock2]
self.assertEqual(
juju_utils.get_unit_name_from_host_name('juju-model-12', 'myapp'),
'myapp/2')
def test_get_machine_status(self):
self.patch_object(juju_utils, "get_full_juju_status")
self.get_full_juju_status.return_value = self.juju_status
# All machine data
self.assertEqual(
juju_utils.get_machine_status(self.machine),
self.machine_data)
self.get_full_juju_status.assert_called_once()
# Request a specific key
self.assertEqual(
juju_utils.get_machine_status(self.machine, self.key),
self.key_data)
def test_get_machine_uuids_for_application(self):
self.patch_object(juju_utils, "get_machines_for_application")
self.get_machines_for_application.return_value = [self.machine]
self.assertEqual(
next(juju_utils.get_machine_uuids_for_application(
self.application)),
self.machine_data.get("instance-id"))
self.get_machines_for_application.assert_called_once_with(
self.application, model_name=None)
def test_get_provider_type(self):
self.patch_object(juju_utils, "get_cloud_configs")
self.get_cloud_configs.return_value = {"type": self.cloud_type}
self.assertEqual(juju_utils.get_provider_type(),
self.cloud_type)
self.get_cloud_configs.assert_called_once_with(self.cloud_name)
def test_remote_run(self):
_cmd = "do the thing"
# Success
self.assertEqual(juju_utils.remote_run(self.unit, _cmd),
self.run_output["Stdout"])
self.model.run_on_unit.assert_called_once_with(
self.unit, _cmd, timeout=None, model_name=None)
# Non-fatal failure
self.model.run_on_unit.return_value = self.error_run_output
self.assertEqual(
juju_utils.remote_run(
self.unit,
_cmd,
fatal=False,
model_name=None),
self.error_run_output["Stderr"])
# Fatal failure
with self.assertRaises(Exception):
juju_utils.remote_run(self.unit, _cmd, fatal=True)
def test_get_unit_names(self):
self.patch('zaza.model.get_first_unit_name', new_callable=mock.Mock(),
name='_get_first_unit_name')
juju_utils._get_unit_names(['aunit/0', 'otherunit/0'])
self.assertFalse(self._get_first_unit_name.called)
def test_get_unit_names_called_with_application_name(self):
self.patch_object(juju_utils, 'model')
juju_utils._get_unit_names(['aunit', 'otherunit/0'])
self.model.get_first_unit_name.assert_called()
def test_get_relation_from_unit(self):
self.patch_object(juju_utils, '_get_unit_names')
self.patch_object(juju_utils, 'yaml')
self.patch_object(juju_utils, 'model')
self._get_unit_names.return_value = ['aunit/0', 'otherunit/0']
data = {'foo': 'bar'}
self.model.get_relation_id.return_value = 42
self.model.run_on_unit.return_value = {'Code': 0, 'Stdout': str(data)}
juju_utils.get_relation_from_unit('aunit/0', 'otherunit/0',
'arelation')
self.model.run_on_unit.assert_called_with(
'aunit/0',
'relation-get --format=yaml -r "42" - "otherunit/0"',
model_name=None)
self.yaml.safe_load.assert_called_with(str(data))
def test_get_relation_from_unit_fails(self):
self.patch_object(juju_utils, '_get_unit_names')
self.patch_object(juju_utils, 'yaml')
self.patch_object(juju_utils, 'model')
self._get_unit_names.return_value = ['aunit/0', 'otherunit/0']
self.model.get_relation_id.return_value = 42
self.model.run_on_unit.return_value = {'Code': 1, 'Stderr': 'ERROR'}
with self.assertRaises(Exception):
juju_utils.get_relation_from_unit('aunit/0', 'otherunit/0',
'arelation')
self.model.run_on_unit.assert_called_with(
'aunit/0',
'relation-get --format=yaml -r "42" - "otherunit/0"',
model_name=None)
self.assertFalse(self.yaml.safe_load.called)
def test_leader_get(self):
self.patch_object(juju_utils, 'yaml')
self.patch_object(juju_utils, 'model')
data = {'foo': 'bar'}
self.model.run_on_leader.return_value = {
'Code': 0, 'Stdout': str(data)}
juju_utils.leader_get('application')
self.model.run_on_leader.assert_called_with(
'application', 'leader-get --format=yaml ', model_name=None)
self.yaml.safe_load.assert_called_with(str(data))
def test_leader_get_key(self):
self.patch_object(juju_utils, 'yaml')
self.patch_object(juju_utils, 'model')
data = {'foo': 'bar'}
self.model.run_on_leader.return_value = {
'Code': 0, 'Stdout': data['foo']}
juju_utils.leader_get('application', 'foo')
self.model.run_on_leader.assert_called_with(
'application', 'leader-get --format=yaml foo', model_name=None)
self.yaml.safe_load.assert_called_with(data['foo'])
def test_leader_get_fails(self):
self.patch_object(juju_utils, 'yaml')
self.patch_object(juju_utils, 'model')
self.model.run_on_leader.return_value = {
'Code': 1, 'Stderr': 'ERROR'}
with self.assertRaises(Exception):
juju_utils.leader_get('application')
self.model.run_on_leader.assert_called_with(
'application', 'leader-get --format=yaml ',
model_name=None)
self.assertFalse(self.yaml.safe_load.called)
def test_get_machine_series(self):
self.patch(
'zaza.openstack.utilities.juju.get_machine_status',
new_callable=mock.MagicMock(),
name='_get_machine_status'
)
self._get_machine_status.return_value = 'xenial'
expected = 'xenial'
actual = juju_utils.get_machine_series('6')
self._get_machine_status.assert_called_with(
machine='6',
key='series',
model_name=None
)
self.assertEqual(expected, actual)
def test_get_subordinate_units(self):
juju_status = mock.MagicMock()
juju_status.applications = {
'nova-compute': {
'units': {
'nova-compute/0': {
'subordinates': {
'neutron-openvswitch/2': {
'charm': 'cs:neutron-openvswitch-22'}}}}},
'cinder': {
'units': {
'cinder/1': {
'subordinates': {
'cinder-hacluster/0': {
'charm': 'cs:hacluster-42'},
'cinder-ceph/3': {
'charm': 'cs:cinder-ceph-2'}}}}},
}
self.assertEqual(
sorted(juju_utils.get_subordinate_units(
['nova-compute/0', 'cinder/1'],
status=juju_status)),
sorted(['neutron-openvswitch/2', 'cinder-hacluster/0',
'cinder-ceph/3']))
self.assertEqual(
juju_utils.get_subordinate_units(
['nova-compute/0', 'cinder/1'],
charm_name='ceph',
status=juju_status),
['cinder-ceph/3'])

18
unit_tests/utilities/test_zaza_utilities_openstack.py
View File

@@ -581,21 +581,27 @@ class TestOpenStackUtils(ut_utils.BaseTestCase):
nova_mock.keypairs.create.assert_called_once_with(name='mykeys')
def test_get_private_key_file(self):
self.patch_object(openstack_utils.deployment_env, 'get_tmpdir',
return_value='/tmp/zaza-model1')
self.assertEqual(
openstack_utils.get_private_key_file('mykeys'),
'tests/id_rsa_mykeys')
'/tmp/zaza-model1/id_rsa_mykeys')
def test_write_private_key(self):
self.patch_object(openstack_utils.deployment_env, 'get_tmpdir',
return_value='/tmp/zaza-model1')
m = mock.mock_open()
with mock.patch(
'zaza.openstack.utilities.openstack.open', m, create=False
):
openstack_utils.write_private_key('mykeys', 'keycontents')
m.assert_called_once_with('tests/id_rsa_mykeys', 'w')
m.assert_called_once_with('/tmp/zaza-model1/id_rsa_mykeys', 'w')
handle = m()
handle.write.assert_called_once_with('keycontents')
def test_get_private_key(self):
self.patch_object(openstack_utils.deployment_env, 'get_tmpdir',
return_value='/tmp/zaza-model1')
self.patch_object(openstack_utils.os.path, "isfile",
return_value=True)
m = mock.mock_open(read_data='myprivkey')
@@ -607,6 +613,8 @@ class TestOpenStackUtils(ut_utils.BaseTestCase):
'myprivkey')
def test_get_private_key_file_missing(self):
self.patch_object(openstack_utils.deployment_env, 'get_tmpdir',
return_value='/tmp/zaza-model1')
self.patch_object(openstack_utils.os.path, "isfile",
return_value=False)
self.assertIsNone(openstack_utils.get_private_key('mykeys'))
@@ -765,7 +773,7 @@ class TestOpenStackUtils(ut_utils.BaseTestCase):
privkey='myprivkey')
paramiko_mock.connect.assert_called_once_with(
'10.0.0.10',
password='',
password=None,
pkey='akey',
username='bob')
@@ -809,12 +817,12 @@ class TestOpenStackUtils(ut_utils.BaseTestCase):
name='_get_os_version'
)
self.patch(
'zaza.openstack.utilities.juju.get_machines_for_application',
'zaza.utilities.juju.get_machines_for_application',
new_callable=mock.MagicMock(),
name='_get_machines'
)
self.patch(
'zaza.openstack.utilities.juju.get_machine_series',
'zaza.utilities.juju.get_machine_series',
new_callable=mock.MagicMock(),
name='_get_machine_series'
)

2
zaza/openstack/charm_tests/ceilometer/tests.py
View File

@@ -101,7 +101,7 @@ class CeilometerTest(test_utils.OpenStackBaseTest):
def test_400_api_connection(self):
"""Simple api calls to check service is up and responding."""
if self.current_release >= CeilometerTest.XENIAL_PIKE:
if self.current_release >= CeilometerTest.XENIAL_OCATA:
logging.info('Skipping API checks as ceilometer api has been '
'removed')
return

42
zaza/openstack/charm_tests/ceph/fs/tests.py
View File

@@ -14,10 +14,10 @@
"""Encapsulate CephFS testing."""
import logging
from tenacity import Retrying, stop_after_attempt, wait_exponential
import zaza.model as model
import zaza.openstack.charm_tests.glance.setup as glance_setup
import zaza.openstack.charm_tests.neutron.tests as neutron_tests
import zaza.openstack.charm_tests.nova.utils as nova_utils
import zaza.openstack.charm_tests.test_utils as test_utils
@@ -63,27 +63,10 @@ write_files:
conf = model.run_on_leader(
'ceph-mon', 'cat /etc/ceph/ceph.conf')['Stdout']
# Spawn Servers
for attempt in Retrying(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)):
with attempt:
instance_1 = guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-1'.format(self.RESOURCE_PREFIX),
userdata=self.INSTANCE_USERDATA.format(
_indent(conf, 8),
_indent(keyring, 8)))
for attempt in Retrying(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)):
with attempt:
instance_2 = guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name='{}-ins-2'.format(self.RESOURCE_PREFIX),
userdata=self.INSTANCE_USERDATA.format(
_indent(conf, 8),
_indent(keyring, 8)))
instance_1, instance_2 = self.launch_guests(
userdata=self.INSTANCE_USERDATA.format(
_indent(conf, 8),
_indent(keyring, 8)))
# Write a file on instance_1
def verify_setup(stdin, stdout, stderr):
@@ -124,3 +107,18 @@ write_files:
def _indent(text, amount, ch=' '):
padding = amount * ch
return ''.join(padding+line for line in text.splitlines(True))
class CharmOperationTest(test_utils.BaseCharmTest):
"""CephFS Charm operation tests."""
def test_pause_resume(self):
"""Run pause and resume tests.
Pause service and check services are stopped, then resume and check
they are started.
"""
services = ['ceph-mds']
with self.pause_resume(services):
logging.info('Testing pause resume (services="{}")'
.format(services))

16
zaza/openstack/charm_tests/ceph/setup.py
View File

@@ -14,7 +14,23 @@
"""Setup for ceph-osd deployments."""
import logging
import zaza.model
def basic_setup():
"""Run basic setup for ceph-osd."""
pass
def ceph_ready():
"""Wait for ceph to be ready.
Wait for ceph to be ready. This is useful if the target_deploy_status in
the tests.yaml is expecting ceph to be in a blocked state. After ceph
has been unblocked the deploy may need to wait for ceph to be ready.
"""
logging.info("Waiting for ceph units to settle")
zaza.model.wait_for_application_states()
zaza.model.block_until_all_units_idle()
logging.info("Ceph units settled")

6
zaza/openstack/charm_tests/ceph/tests.py
View File

@@ -544,7 +544,7 @@ class CephRGWTest(test_utils.OpenStackBaseTest):
@classmethod
def setUpClass(cls):
"""Run class setup for running ceph low level tests."""
super(CephRGWTest, cls).setUpClass()
super(CephRGWTest, cls).setUpClass(application_name='ceph-radosgw')
@property
def expected_apps(self):
@@ -622,7 +622,9 @@ class CephRGWTest(test_utils.OpenStackBaseTest):
'multisite configuration')
logging.info('Checking Swift REST API')
keystone_session = zaza_openstack.get_overcloud_keystone_session()
region_name = 'RegionOne'
region_name = zaza_model.get_application_config(
self.application_name,
model_name=self.model_name)['region']['value']
swift_client = zaza_openstack.get_swift_session_client(
keystone_session,
region_name,

28
zaza/openstack/charm_tests/cinder/tests.py
View File

@@ -217,12 +217,22 @@ class CinderTests(test_utils.OpenStackBaseTest):
@property
def services(self):
"""Return a list services for the selected OpenStack release."""
services = ['cinder-scheduler', 'cinder-volume']
if (openstack_utils.get_os_release() >=
openstack_utils.get_os_release('xenial_ocata')):
services.append('apache2')
current_value = zaza.model.get_application_config(
self.application_name)['enabled-services']['value']
if current_value == "all":
services = ['cinder-scheduler', 'cinder-volume', 'cinder-api']
else:
services.append('cinder-api')
services = ['cinder-{}'.format(svc)
for svc in ('api', 'scheduler', 'volume')
if svc in current_value]
if ('cinder-api' in services and
(openstack_utils.get_os_release() >=
openstack_utils.get_os_release('xenial_ocata'))):
services.remove('cinder-api')
services.append('apache2')
return services
def test_900_restart_on_config_change(self):
@@ -246,13 +256,7 @@ class CinderTests(test_utils.OpenStackBaseTest):
Pause service and check services are stopped then resume and check
they are started
"""
services = ['cinder-scheduler', 'cinder-volume']
if (openstack_utils.get_os_release() >=
openstack_utils.get_os_release('xenial_ocata')):
services.append('apache2')
else:
services.append('cinder-api')
with self.pause_resume(services):
with self.pause_resume(self.services):
logging.info("Testing pause resume")

26
zaza/openstack/charm_tests/glance/setup.py
View File

@@ -16,6 +16,7 @@
import logging
import zaza.openstack.utilities.openstack as openstack_utils
import zaza.utilities.deployment_env as deployment_env
CIRROS_IMAGE_NAME = "cirros"
CIRROS_ALT_IMAGE_NAME = "cirros_alt"
@@ -31,7 +32,8 @@ def basic_setup():
"""
def add_image(image_url, glance_client=None, image_name=None, tags=[]):
def add_image(image_url, glance_client=None, image_name=None, tags=[],
properties=None):
"""Retrieve image from ``image_url`` and add it to glance.
:param image_url: Retrievable URL with image data
@@ -42,6 +44,8 @@ def add_image(image_url, glance_client=None, image_name=None, tags=[]):
:type image_name: str
:param tags: List of tags to add to image
:type tags: list of str
:param properties: Properties to add to image
:type properties: dict
"""
if not glance_client:
keystone_session = openstack_utils.get_overcloud_keystone_session()
@@ -60,7 +64,8 @@ def add_image(image_url, glance_client=None, image_name=None, tags=[]):
glance_client,
image_url,
image_name,
tags=tags)
tags=tags,
properties=properties)
def add_cirros_image(glance_client=None, image_name=None):
@@ -90,7 +95,8 @@ def add_cirros_alt_image(glance_client=None, image_name=None):
add_cirros_image(glance_client, image_name)
def add_lts_image(glance_client=None, image_name=None, release=None):
def add_lts_image(glance_client=None, image_name=None, release=None,
properties=None):
"""Add an Ubuntu LTS image to the current deployment.
:param glance: Authenticated glanceclient
@@ -99,12 +105,22 @@ def add_lts_image(glance_client=None, image_name=None, release=None):
:type image_name: str
:param release: Name of ubuntu release.
:type release: str
:param properties: Custom image properties
:type properties: dict
"""
deploy_ctxt = deployment_env.get_deployment_context()
image_arch = deploy_ctxt.get('TEST_IMAGE_ARCH', 'amd64')
arch_image_properties = {
'arm64': {'hw_firmware_type': 'uefi'},
'ppc64el': {'architecture': 'ppc64'}}
properties = properties or arch_image_properties.get(image_arch)
logging.info("Image architecture set to {}".format(image_arch))
image_name = image_name or LTS_IMAGE_NAME
release = release or LTS_RELEASE
image_url = openstack_utils.find_ubuntu_image(
release=release,
arch='amd64')
arch=image_arch)
add_image(image_url,
glance_client=glance_client,
image_name=image_name)
image_name=image_name,
properties=properties)

40
zaza/openstack/charm_tests/glance_simplestreams_sync/setup.py Normal file
View File

@@ -0,0 +1,40 @@
#!/usr/bin/env python3
# Copyright 2019 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.
"""Code for configuring glance-simplestreams-sync."""
import logging
import zaza.model as zaza_model
import zaza.openstack.utilities.generic as generic_utils
def sync_images():
"""Run image sync using an action.
Execute an initial image sync using an action to ensure that the
cloud is populated with images at the right point in time during
deployment.
"""
logging.info("Synchronising images using glance-simplestreams-sync")
generic_utils.assertActionRanOK(
zaza_model.run_action_on_leader(
"glance-simplestreams-sync",
"sync-images",
raise_on_failure=True,
action_params={},
)
)

33
zaza/openstack/charm_tests/glance_simplestreams_sync/tests.py
View File

@@ -24,7 +24,7 @@ import zaza.openstack.utilities.openstack as openstack_utils
@tenacity.retry(
retry=tenacity.retry_if_result(lambda images: len(images) < 3),
retry=tenacity.retry_if_result(lambda images: len(images) < 4),
wait=tenacity.wait_fixed(6), # interval between retries
stop=tenacity.stop_after_attempt(100)) # retry times
def retry_image_sync(glance_client):
@@ -42,7 +42,7 @@ def get_product_streams(url):
# There is a race between the images being available in glance and any
# metadata being written. Use tenacity to avoid this race.
client = requests.session()
json_data = client.get(url).text
json_data = client.get(url, verify=openstack_utils.get_cacert()).text
return json.loads(json_data)
@@ -61,7 +61,7 @@ class GlanceSimpleStreamsSyncTest(test_utils.OpenStackBaseTest):
cls.keystone_session)
def test_010_wait_for_image_sync(self):
"""Wait for images to be synced. Expect at least three."""
"""Wait for images to be synced. Expect at least four."""
self.assertTrue(retry_image_sync(self.glance_client))
def test_050_gss_permissions_regression_check_lp1611987(self):
@@ -94,31 +94,34 @@ class GlanceSimpleStreamsSyncTest(test_utils.OpenStackBaseTest):
'com.ubuntu.cloud:server:14.04:amd64',
'com.ubuntu.cloud:server:16.04:amd64',
'com.ubuntu.cloud:server:18.04:amd64',
'com.ubuntu.cloud:server:20.04:amd64',
]
uri = "streams/v1/auto.sync.json"
key = "url"
xenial_pike = openstack_utils.get_os_release('xenial_pike')
if openstack_utils.get_os_release() <= xenial_pike:
key = "publicURL"
catalog = self.keystone_client.service_catalog.get_endpoints()
ps_interface = catalog["product-streams"][0][key]
url = "{}/{}".format(ps_interface, uri)
# There is a race between the images being available in glance and the
# metadata being written for each image. Use tenacity to avoid this
# race and make the test idempotent.
@tenacity.retry(
retry=tenacity.retry_if_exception_type(AssertionError),
retry=tenacity.retry_if_exception_type(
(AssertionError, KeyError)
),
wait=tenacity.wait_fixed(10), reraise=True,
stop=tenacity.stop_after_attempt(10))
def _check_local_product_streams(url, expected_images):
stop=tenacity.stop_after_attempt(25))
def _check_local_product_streams(expected_images):
# Refresh from catalog as URL may change if swift in use.
ps_interface = self.keystone_client.service_catalog.url_for(
service_type='product-streams', interface='publicURL'
)
url = "{}/{}".format(ps_interface, uri)
logging.info('Retrieving product stream information'
' from {}'.format(url))
product_streams = get_product_streams(url)
logging.debug(product_streams)
images = product_streams["products"]
for image in expected_images:
self.assertIn(image, images)
_check_local_product_streams(url, expected_images)
_check_local_product_streams(expected_images)
logging.debug("Local product stream successful")

23
zaza/openstack/charm_tests/keystone/setup.py
View File

@@ -14,8 +14,12 @@
"""Code for setting up keystone."""
import logging
import keystoneauth1
import zaza.charm_lifecycle.utils as lifecycle_utils
import zaza.model
import zaza.openstack.utilities.openstack as openstack_utils
from zaza.openstack.charm_tests.keystone import (
BaseKeystoneTest,
@@ -30,6 +34,25 @@ from zaza.openstack.charm_tests.keystone import (
)
def wait_for_cacert(model_name=None):
"""Wait for keystone to install a cacert.
:param model_name: Name of model to query.
:type model_name: str
"""
logging.info("Waiting for cacert")
zaza.model.block_until_file_has_contents(
'keystone',
openstack_utils.KEYSTONE_REMOTE_CACERT,
'CERTIFICATE',
model_name=model_name)
zaza.model.block_until_all_units_idle(model_name=model_name)
test_config = lifecycle_utils.get_charm_config(fatal=False)
zaza.model.wait_for_application_states(
states=test_config.get('target_deploy_status', {}),
model_name=model_name)
def add_demo_user():
"""Add a demo user to the current deployment."""
def _v2():

8
zaza/openstack/charm_tests/keystone/tests.py
View File

@@ -21,7 +21,7 @@ import keystoneauth1
import zaza.model
import zaza.openstack.utilities.exceptions as zaza_exceptions
import zaza.openstack.utilities.juju as juju_utils
import zaza.utilities.juju as juju_utils
import zaza.openstack.utilities.openstack as openstack_utils
import zaza.charm_lifecycle.utils as lifecycle_utils
import zaza.openstack.charm_tests.test_utils as test_utils
@@ -262,6 +262,7 @@ class AuthenticationAuthorizationTest(BaseKeystoneTest):
openrc['OS_CACERT'] = openstack_utils.KEYSTONE_LOCAL_CACERT
openrc['OS_AUTH_URL'] = (
openrc['OS_AUTH_URL'].replace('http', 'https'))
logging.info('keystone IP {}'.format(ip))
keystone_session = openstack_utils.get_keystone_session(
openrc)
keystone_client = openstack_utils.get_keystone_session_client(
@@ -319,10 +320,7 @@ class AuthenticationAuthorizationTest(BaseKeystoneTest):
'OS_PROJECT_DOMAIN_NAME': DEMO_DOMAIN,
'OS_PROJECT_NAME': DEMO_PROJECT,
}
with self.config_change(
{'preferred-api-version': self.default_api_version},
{'preferred-api-version': self.api_v3},
application_name="keystone"):
with self.v3_keystone_preferred():
for ip in self.keystone_ips:
openrc.update(
{'OS_AUTH_URL': 'http://{}:5000/v3'.format(ip)})

24
zaza/openstack/charm_tests/masakari/tests.py
View File

@@ -38,7 +38,7 @@ class MasakariTest(test_utils.OpenStackBaseTest):
@classmethod
def setUpClass(cls):
"""Run class setup for running tests."""
super(MasakariTest, cls).setUpClass()
super(MasakariTest, cls).setUpClass(application_name="masakari")
cls.current_release = openstack_utils.get_os_release()
cls.keystone_session = openstack_utils.get_overcloud_keystone_session()
cls.model_name = zaza.model.get_juju_model()
@@ -134,6 +134,26 @@ class MasakariTest(test_utils.OpenStackBaseTest):
vm_uuid,
model_name=self.model_name)
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=2, max=60),
reraise=True, stop=tenacity.stop_after_attempt(5),
retry=tenacity.retry_if_exception_type(AssertionError))
def wait_for_guest_ready(self, vm_name):
"""Wait for the guest to be ready.
:param vm_name: Name of guest to check.
:type vm_name: str
"""
guest_ready_attr_checks = [
('OS-EXT-STS:task_state', None),
('status', 'ACTIVE'),
('OS-EXT-STS:power_state', 1),
('OS-EXT-STS:vm_state', 'active')]
guest = self.nova_client.servers.find(name=vm_name)
logging.info('Checking guest {} attributes'.format(vm_name))
for (attr, required_state) in guest_ready_attr_checks:
logging.info('Checking {} is {}'.format(attr, required_state))
assert getattr(guest, attr) == required_state
def test_instance_failover(self):
"""Test masakari managed guest migration."""
# Workaround for Bug #1874719
@@ -168,6 +188,7 @@ class MasakariTest(test_utils.OpenStackBaseTest):
model_name=self.model_name)
openstack_utils.enable_all_nova_services(self.nova_client)
zaza.openstack.configure.masakari.enable_hosts()
self.wait_for_guest_ready(vm_name)
def test_instance_restart_on_fail(self):
"""Test single guest crash and recovery."""
@@ -178,6 +199,7 @@ class MasakariTest(test_utils.OpenStackBaseTest):
self.current_release))
vm_name = 'zaza-test-instance-failover'
vm = self.ensure_guest(vm_name)
self.wait_for_guest_ready(vm_name)
_, unit_name = self.get_guests_compute_info(vm_name)
logging.info('{} is running on {}'.format(vm_name, unit_name))
guest_pid = self.get_guest_qemu_pid(

55
zaza/openstack/charm_tests/mysql/tests.py
View File

@@ -149,7 +149,7 @@ class MySQLCommonTests(MySQLBaseTest):
set_alternate = {"max-connections": "1000"}
# Make config change, check for service restarts
logging.debug("Setting max connections ...")
logging.info("Setting max connections ...")
self.restart_on_changed(
self.conf_file,
set_default,
@@ -198,7 +198,7 @@ class PerconaClusterBaseTest(MySQLBaseTest):
output = zaza.model.run_on_leader(
self.application, cmd)["Stdout"].strip()
value = re.search(r"^.+?\s+(.+)", output).group(1)
logging.debug("%s = %s" % (attr, value))
logging.info("%s = %s" % (attr, value))
return value
def is_pxc_bootstrapped(self):
@@ -236,7 +236,7 @@ class PerconaClusterBaseTest(MySQLBaseTest):
cmd = "ip -br addr"
result = zaza.model.run_on_unit(unit.entity_id, cmd)
output = result.get("Stdout").strip()
logging.debug(output)
logging.info(output)
if self.vip in output:
logging.info("vip ({}) running in {}".format(
self.vip,
@@ -333,12 +333,12 @@ class PerconaClusterColdStartTest(PerconaClusterBaseTest):
juju_utils.get_machine_uuids_for_application(self.application))
# Stop Nodes
# Avoid hitting an update-status hook
logging.debug("Wait till model is idle ...")
logging.info("Wait till model is idle ...")
zaza.model.block_until_all_units_idle()
logging.info("Stopping instances: {}".format(_machines))
for uuid in _machines:
self.nova_client.servers.stop(uuid)
logging.debug("Wait till all machines are shutoff ...")
logging.info("Wait till all machines are shutoff ...")
for uuid in _machines:
openstack_utils.resource_reaches_status(self.nova_client.servers,
uuid,
@@ -357,7 +357,7 @@ class PerconaClusterColdStartTest(PerconaClusterBaseTest):
'unknown',
negate_match=True)
logging.debug("Wait till model is idle ...")
logging.info("Wait till model is idle ...")
# XXX If a hook was executing on a unit when it was powered off
# it comes back in an error state.
try:
@@ -366,7 +366,7 @@ class PerconaClusterColdStartTest(PerconaClusterBaseTest):
self.resolve_update_status_errors()
zaza.model.block_until_all_units_idle()
logging.debug("Wait for application states ...")
logging.info("Wait for application states ...")
for unit in zaza.model.get_units(self.application):
try:
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
@@ -389,7 +389,7 @@ class PerconaClusterColdStartTest(PerconaClusterBaseTest):
_non_leaders[0],
"bootstrap-pxc",
action_params={})
logging.debug("Wait for application states ...")
logging.info("Wait for application states ...")
for unit in zaza.model.get_units(self.application):
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
states = {"percona-cluster": {
@@ -403,7 +403,7 @@ class PerconaClusterColdStartTest(PerconaClusterBaseTest):
self.application,
"notify-bootstrapped",
action_params={})
logging.debug("Wait for application states ...")
logging.info("Wait for application states ...")
for unit in zaza.model.get_units(self.application):
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
test_config = lifecycle_utils.get_charm_config(fatal=False)
@@ -521,7 +521,7 @@ class MySQLInnoDBClusterColdStartTest(MySQLBaseTest):
zaza.model.resolve_units(
application_name=self.application,
erred_hook='update-status',
wait=True)
wait=True, timeout=180)
def test_100_reboot_cluster_from_complete_outage(self):
"""Reboot cluster from complete outage.
@@ -532,12 +532,12 @@ class MySQLInnoDBClusterColdStartTest(MySQLBaseTest):
juju_utils.get_machine_uuids_for_application(self.application))
# Stop Nodes
# Avoid hitting an update-status hook
logging.debug("Wait till model is idle ...")
logging.info("Wait till model is idle ...")
zaza.model.block_until_all_units_idle()
logging.info("Stopping instances: {}".format(_machines))
for uuid in _machines:
self.nova_client.servers.stop(uuid)
logging.debug("Wait till all machines are shutoff ...")
logging.info("Wait till all machines are shutoff ...")
for uuid in _machines:
openstack_utils.resource_reaches_status(self.nova_client.servers,
uuid,
@@ -550,38 +550,37 @@ class MySQLInnoDBClusterColdStartTest(MySQLBaseTest):
for uuid in _machines:
self.nova_client.servers.start(uuid)
logging.info(
"Wait till all {} units are in state 'unkown' ..."
.format(self.application))
for unit in zaza.model.get_units(self.application):
zaza.model.block_until_unit_wl_status(
unit.entity_id,
'unknown',
negate_match=True)
logging.debug("Wait till model is idle ...")
logging.info("Wait till model is idle ...")
try:
zaza.model.block_until_all_units_idle()
except zaza.model.UnitError:
self.resolve_update_status_errors()
zaza.model.block_until_all_units_idle()
logging.debug("Clear error hooks after reboot ...")
logging.info("Clear error hooks after reboot ...")
for unit in zaza.model.get_units(self.application):
try:
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
except zaza.model.UnitError:
self.resolve_update_status_errors()
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
logging.debug("Wait for application states blocked ...")
states = {
self.application: {
"workload-status": "blocked",
"workload-status-message":
"MySQL InnoDB Cluster not healthy: None"},
"mysql-router": {
"workload-status": "blocked",
"workload-status-message":
"Failed to connect to MySQL"}}
zaza.model.wait_for_application_states(states=states)
logging.info(
"Wait till all {} units are in state 'blocked' ..."
.format(self.application))
for unit in zaza.model.get_units(self.application):
zaza.model.block_until_unit_wl_status(
unit.entity_id,
'blocked')
logging.info("Execute reboot-cluster-from-complete-outage "
"action after cold boot ...")
@@ -592,15 +591,15 @@ class MySQLInnoDBClusterColdStartTest(MySQLBaseTest):
unit.entity_id,
"reboot-cluster-from-complete-outage",
action_params={})
if "Success" in action.data["results"].get("outcome"):
if "Success" in action.data.get("results", {}).get("outcome", ""):
break
else:
logging.info(action.data["results"].get("output"))
logging.info(action.data.get("results", {}).get("output", ""))
assert "Success" in action.data["results"]["outcome"], (
"Reboot cluster from complete outage action failed: {}"
.format(action.data))
logging.debug("Wait for application states ...")
logging.info("Wait for application states ...")
for unit in zaza.model.get_units(self.application):
zaza.model.run_on_unit(unit.entity_id, "hooks/update-status")
test_config = lifecycle_utils.get_charm_config(fatal=False)

119
zaza/openstack/charm_tests/neutron/tests.py
View File

@@ -25,10 +25,7 @@ import logging
import tenacity
import unittest
import novaclient
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
@@ -292,36 +289,44 @@ class NeutronCreateNetworkTest(test_utils.OpenStackBaseTest):
# set up clients
cls.neutron_client = (
openstack_utils.get_neutron_session_client(cls.keystone_session))
cls.neutron_client.format = 'json'
_TEST_NET_NAME = 'test_net'
def test_400_create_network(self):
"""Create a network, verify that it exists, and then delete it."""
self._assert_test_network_doesnt_exist()
self._create_test_network()
net_id = self._assert_test_network_exists_and_return_id()
self._delete_test_network(net_id)
self._assert_test_network_doesnt_exist()
def _create_test_network(self):
logging.debug('Creating neutron network...')
self.neutron_client.format = 'json'
net_name = 'test_net'
# Verify that the network doesn't exist
networks = self.neutron_client.list_networks(name=net_name)
net_count = len(networks['networks'])
assert net_count == 0, (
"Expected zero networks, found {}".format(net_count))
# Create a network and verify that it exists
network = {'name': net_name}
network = {'name': self._TEST_NET_NAME}
self.neutron_client.create_network({'network': network})
networks = self.neutron_client.list_networks(name=net_name)
def _delete_test_network(self, net_id):
logging.debug('Deleting neutron network...')
self.neutron_client.delete_network(net_id)
def _assert_test_network_exists_and_return_id(self):
logging.debug('Confirming new neutron network...')
networks = self.neutron_client.list_networks(name=self._TEST_NET_NAME)
logging.debug('Networks: {}'.format(networks))
net_len = len(networks['networks'])
assert net_len == 1, (
"Expected 1 network, found {}".format(net_len))
logging.debug('Confirming new neutron network...')
network = networks['networks'][0]
assert network['name'] == net_name, "network ext_net not found"
assert network['name'] == self._TEST_NET_NAME, \
"network {} not found".format(self._TEST_NET_NAME)
return network['id']
# Cleanup
logging.debug('Deleting neutron network...')
self.neutron_client.delete_network(network['id'])
def _assert_test_network_doesnt_exist(self):
networks = self.neutron_client.list_networks(name=self._TEST_NET_NAME)
net_count = len(networks['networks'])
assert net_count == 0, (
"Expected zero networks, found {}".format(net_count))
class NeutronApiTest(NeutronCreateNetworkTest):
@@ -600,7 +605,7 @@ class NeutronOpenvSwitchTest(NeutronPluginApiSharedTests):
logging.info('Testing pause resume')
class NeutronNetworkingBase(unittest.TestCase):
class NeutronNetworkingBase(test_utils.OpenStackBaseTest):
"""Base for checking openstack instances have valid networking."""
RESOURCE_PREFIX = 'zaza-neutrontests'
@@ -608,30 +613,10 @@ class NeutronNetworkingBase(unittest.TestCase):
@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))
super(NeutronNetworkingBase, cls).setUpClass(
application_name='neutron-api')
cls.neutron_client = (
openstack_utils.get_neutron_session_client(cls.keystone_session))
# NOTE(fnordahl): in the event of a test failure we do not want to run
# tear down code as it will make debugging a problem virtually
# impossible. To alleviate each test method will set the
# `run_tearDown` instance variable at the end which will let us run
# tear down only when there were no failure.
cls.run_tearDown = False
@classmethod
def tearDown(cls):
"""Remove test resources."""
if cls.run_tearDown:
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")
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60),
reraise=True, stop=tenacity.stop_after_attempt(8))
@@ -729,44 +714,6 @@ class NeutronNetworkingBase(unittest.TestCase):
assert agent['admin_state_up']
assert agent['alive']
def launch_guests(self):
"""Launch two guests to use in tests."""
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))
def retrieve_guest(self, nova_client, guest_name):
"""Return guest matching name.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
try:
return nova_client.servers.find(name=guest_name)
except novaclient.exceptions.NotFound:
return None
def retrieve_guests(self, nova_client):
"""Return test guests.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
instance_1 = self.retrieve_guest(
nova_client,
'{}-ins-1'.format(self.RESOURCE_PREFIX))
instance_2 = self.retrieve_guest(
nova_client,
'{}-ins-1'.format(self.RESOURCE_PREFIX))
return instance_1, instance_2
def check_connectivity(self, instance_1, instance_2):
"""Run North/South and East/West connectivity tests."""
def verify(stdin, stdout, stderr):
@@ -837,9 +784,9 @@ class NeutronNetworkingTest(NeutronNetworkingBase):
def test_instances_have_networking(self):
"""Validate North/South and East/West networking."""
self.launch_guests()
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
self.check_connectivity(instance_1, instance_2)
self.run_tearDown = True
self.run_resource_cleanup = True
class NeutronNetworkingVRRPTests(NeutronNetworkingBase):
@@ -847,10 +794,10 @@ class NeutronNetworkingVRRPTests(NeutronNetworkingBase):
def test_gateway_failure(self):
"""Validate networking in the case of a gateway failure."""
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
if not all([instance_1, instance_2]):
self.launch_guests()
instance_1, instance_2 = self.retrieve_guests(self.nova_client)
instance_1, instance_2 = self.retrieve_guests()
self.check_connectivity(instance_1, instance_2)
routers = self.neutron_client.list_routers(

15
zaza/openstack/charm_tests/neutron_arista/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
# Copyright 2020 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.
"""Collection of code for setting up and testing neutron-api-plugin-arista."""

79
zaza/openstack/charm_tests/neutron_arista/setup.py Normal file
View File

@@ -0,0 +1,79 @@
# Copyright 2020 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.
"""Code for setting up neutron-api-plugin-arista."""
import logging
import os
import tenacity
import zaza
import zaza.openstack.charm_tests.neutron_arista.utils as arista_utils
import zaza.openstack.utilities.openstack as openstack_utils
def download_arista_image():
"""Download arista-cvx-virt-test.qcow2 from a web server.
The download will happen only if the env var TEST_ARISTA_IMAGE_REMOTE has
been set, so you don't have to set it if you already have the image
locally.
If the env var TEST_ARISTA_IMAGE_LOCAL isn't set, it will be set to
`/tmp/arista-cvx-virt-test.qcow2`. This is where the image will be
downloaded to if TEST_ARISTA_IMAGE_REMOTE has been set.
"""
try:
os.environ['TEST_ARISTA_IMAGE_LOCAL']
except KeyError:
os.environ['TEST_ARISTA_IMAGE_LOCAL'] = ''
if not os.environ['TEST_ARISTA_IMAGE_LOCAL']:
os.environ['TEST_ARISTA_IMAGE_LOCAL'] \
= '/tmp/arista-cvx-virt-test.qcow2'
try:
if os.environ['TEST_ARISTA_IMAGE_REMOTE']:
logging.info('Downloading Arista image from {}'
.format(os.environ['TEST_ARISTA_IMAGE_REMOTE']))
openstack_utils.download_image(
os.environ['TEST_ARISTA_IMAGE_REMOTE'],
os.environ['TEST_ARISTA_IMAGE_LOCAL'])
except KeyError:
# TEST_ARISTA_IMAGE_REMOTE isn't set, which means the image is already
# available at TEST_ARISTA_IMAGE_LOCAL
pass
logging.info('Arista image can be found at {}'
.format(os.environ['TEST_ARISTA_IMAGE_LOCAL']))
def test_fixture():
"""Pass arista-virt-test-fixture's IP address to Neutron."""
fixture_ip_addr = arista_utils.fixture_ip_addr()
logging.info(
"{}'s IP address is '{}'. Passing it to {}..."
.format(arista_utils.FIXTURE_APP_NAME, fixture_ip_addr,
arista_utils.PLUGIN_APP_NAME))
zaza.model.set_application_config(arista_utils.PLUGIN_APP_NAME,
{'eapi-host': fixture_ip_addr})
logging.info('Waiting for {} to become ready...'.format(
arista_utils.PLUGIN_APP_NAME))
zaza.model.wait_for_agent_status()
zaza.model.wait_for_application_states()
for attempt in tenacity.Retrying(
wait=tenacity.wait_fixed(10), # seconds
stop=tenacity.stop_after_attempt(30),
reraise=True):
with attempt:
arista_utils.query_fixture_networks(fixture_ip_addr)

74
zaza/openstack/charm_tests/neutron_arista/tests.py Normal file
View File

@@ -0,0 +1,74 @@
#!/usr/bin/env python3
# Copyright 2020 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-api-plugin-arista` testing."""
import logging
import tenacity
import zaza.openstack.charm_tests.neutron.tests as neutron_tests
import zaza.openstack.charm_tests.neutron_arista.utils as arista_utils
class NeutronCreateAristaNetworkTest(neutron_tests.NeutronCreateNetworkTest):
"""Test creating an Arista Neutron network through the API."""
@classmethod
def setUpClass(cls):
"""Run class setup for running Neutron Arista tests."""
super(NeutronCreateAristaNetworkTest, cls).setUpClass()
logging.info('Waiting for Neutron to become ready...')
for attempt in tenacity.Retrying(
wait=tenacity.wait_fixed(5), # seconds
stop=tenacity.stop_after_attempt(12),
reraise=True):
with attempt:
cls.neutron_client.list_networks()
def _assert_test_network_exists_and_return_id(self):
logging.info('Checking that the test network exists on the Arista '
'test fixture...')
# Sometimes the API call from Neutron to Arista fails and Neutron
# retries a couple of seconds later, which is why the newly created
# test network may not be immediately visible on Arista's API.
for attempt in tenacity.Retrying(
wait=tenacity.wait_fixed(10), # seconds
stop=tenacity.stop_after_attempt(3),
reraise=True):
with attempt:
actual_network_names = arista_utils.query_fixture_networks(
arista_utils.fixture_ip_addr())
self.assertEqual(actual_network_names, [self._TEST_NET_NAME])
return super(NeutronCreateAristaNetworkTest,
self)._assert_test_network_exists_and_return_id()
def _assert_test_network_doesnt_exist(self):
logging.info("Checking that the test network doesn't exist on the "
"Arista test fixture...")
for attempt in tenacity.Retrying(
wait=tenacity.wait_fixed(10), # seconds
stop=tenacity.stop_after_attempt(3),
reraise=True):
with attempt:
actual_network_names = arista_utils.query_fixture_networks(
arista_utils.fixture_ip_addr())
self.assertEqual(actual_network_names, [])
super(NeutronCreateAristaNetworkTest,
self)._assert_test_network_doesnt_exist()

68
zaza/openstack/charm_tests/neutron_arista/utils.py Normal file
View File

@@ -0,0 +1,68 @@
# Copyright 2020 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.
"""Common Arista-related utils."""
import json
import requests
import urllib3
import zaza
FIXTURE_APP_NAME = 'arista-virt-test-fixture'
PLUGIN_APP_NAME = 'neutron-api-plugin-arista'
def fixture_ip_addr():
"""Return the public IP address of the Arista test fixture."""
return zaza.model.get_units(FIXTURE_APP_NAME)[0].public_address
_FIXTURE_LOGIN = 'admin'
_FIXTURE_PASSWORD = 'password123'
def query_fixture_networks(ip_addr):
"""Query the Arista test fixture's list of networks."""
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
session = requests.Session()
session.headers['Content-Type'] = 'application/json'
session.headers['Accept'] = 'application/json'
session.verify = False
session.auth = (_FIXTURE_LOGIN, _FIXTURE_PASSWORD)
data = {
'id': 'Zaza {} tests'.format(PLUGIN_APP_NAME),
'method': 'runCmds',
'jsonrpc': '2.0',
'params': {
'timestamps': False,
'format': 'json',
'version': 1,
'cmds': ['show openstack networks']
}
}
response = session.post(
'https://{}/command-api/'.format(ip_addr),
data=json.dumps(data),
timeout=10 # seconds
)
result = []
for region in response.json()['result'][0]['regions'].values():
for tenant in region['tenants'].values():
for network in tenant['tenantNetworks'].values():
result.append(network['networkName'])
return result

10
zaza/openstack/charm_tests/nova/tests.py
View File

@@ -56,6 +56,16 @@ class LTSGuestCreateTest(BaseGuestCreateTest):
glance_setup.LTS_IMAGE_NAME)
class LTSGuestCreateVolumeBackedTest(BaseGuestCreateTest):
"""Tests to launch a LTS image."""
def test_launch_small_instance(self):
"""Launch a Bionic instance and test connectivity."""
zaza.openstack.configure.guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
use_boot_volume=True)
class NovaCompute(test_utils.OpenStackBaseTest):
"""Run nova-compute specific tests."""

19
zaza/openstack/charm_tests/octavia/setup.py
View File

@@ -98,25 +98,6 @@ def configure_octavia():
pass
def prepare_payload_instance():
"""Prepare a instance we can use as payload test."""
session = openstack.get_overcloud_keystone_session()
keystone = openstack.get_keystone_session_client(session)
neutron = openstack.get_neutron_session_client(session)
project_id = openstack.get_project_id(
keystone, 'admin', domain_name='admin_domain')
openstack.add_neutron_secgroup_rules(
neutron,
project_id,
[{'protocol': 'tcp',
'port_range_min': '80',
'port_range_max': '80',
'direction': 'ingress'}])
zaza.openstack.configure.guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
userdata='#cloud-config\npackages:\n - apache2\n')
def centralized_fip_network():
"""Create network with centralized router for connecting lb and fips.

71
zaza/openstack/charm_tests/octavia/tests.py
View File

@@ -38,7 +38,20 @@ class CharmOperationTest(test_utils.OpenStackBaseTest):
Pause service and check services are stopped, then resume and check
they are started.
"""
self.pause_resume(['apache2'])
services = [
'apache2',
'octavia-health-manager',
'octavia-housekeeping',
'octavia-worker',
]
if openstack_utils.ovn_present():
services.append('octavia-driver-agent')
logging.info('Skipping pause resume test LP: #1886202...')
return
logging.info('Testing pause resume (services="{}")'
.format(services))
with self.pause_resume(services, pgrep_full=True):
pass
class LBAASv2Test(test_utils.OpenStackBaseTest):
@@ -48,12 +61,13 @@ class LBAASv2Test(test_utils.OpenStackBaseTest):
def setUpClass(cls):
"""Run class setup for running LBaaSv2 service tests."""
super(LBAASv2Test, cls).setUpClass()
cls.keystone_session = openstack_utils.get_overcloud_keystone_session()
cls.keystone_client = openstack_utils.get_keystone_session_client(
cls.keystone_session)
cls.neutron_client = openstack_utils.get_neutron_session_client(
cls.keystone_session)
cls.octavia_client = openstack_utils.get_octavia_session_client(
cls.keystone_session)
cls.RESOURCE_PREFIX = 'zaza-octavia'
# NOTE(fnordahl): in the event of a test failure we do not want to run
# tear down code as it will make debugging a problem virtually
@@ -63,28 +77,24 @@ class LBAASv2Test(test_utils.OpenStackBaseTest):
cls.run_tearDown = False
# List of load balancers created by this test
cls.loadbalancers = []
# LIst of floating IPs created by this test
# List of floating IPs created by this test
cls.fips = []
@classmethod
def tearDown(cls):
"""Remove resources created during test execution.
Note that resources created in the configure step prior to executing
the test should not be touched here.
"""
if not cls.run_tearDown:
return
for lb in cls.loadbalancers:
cls.octavia_client.load_balancer_delete(lb['id'], cascade=True)
def resource_cleanup(self):
"""Remove resources created during test execution."""
for lb in self.loadbalancers:
self.octavia_client.load_balancer_delete(lb['id'], cascade=True)
try:
cls.wait_for_lb_resource(
cls.octavia_client.load_balancer_show, lb['id'],
self.wait_for_lb_resource(
self.octavia_client.load_balancer_show, lb['id'],
provisioning_status='DELETED')
except osc_lib.exceptions.NotFound:
pass
for fip in cls.fips:
cls.neutron_client.delete_floatingip(fip)
for fip in self.fips:
self.neutron_client.delete_floatingip(fip)
# we run the parent resource_cleanup last as it will remove instances
# referenced as members in the above cleaned up load balancers
super(LBAASv2Test, self).resource_cleanup()
@staticmethod
@tenacity.retry(retry=tenacity.retry_if_exception_type(AssertionError),
@@ -238,12 +248,27 @@ class LBAASv2Test(test_utils.OpenStackBaseTest):
def test_create_loadbalancer(self):
"""Create load balancer."""
nova_client = openstack_utils.get_nova_session_client(
self.keystone_session)
# Prepare payload instances
# First we allow communication to port 80 by adding a security group
# rule
project_id = openstack_utils.get_project_id(
self.keystone_client, 'admin', domain_name='admin_domain')
openstack_utils.add_neutron_secgroup_rules(
self.neutron_client,
project_id,
[{'protocol': 'tcp',
'port_range_min': '80',
'port_range_max': '80',
'direction': 'ingress'}])
# Then we request two Ubuntu instances with the Apache web server
# installed
instance_1, instance_2 = self.launch_guests(
userdata='#cloud-config\npackages:\n - apache2\n')
# Get IP of the prepared payload instances
payload_ips = []
for server in nova_client.servers.list():
for server in (instance_1, instance_2):
payload_ips.append(server.networks['private'][0])
self.assertTrue(len(payload_ips) > 0)
@@ -274,4 +299,4 @@ class LBAASv2Test(test_utils.OpenStackBaseTest):
lb_fp['floating_ip_address']))
# If we get here, it means the tests passed
self.run_tearDown = True
self.run_resource_cleanup = True

91
zaza/openstack/charm_tests/policyd/tests.py
View File

@@ -401,6 +401,10 @@ class BasePolicydSpecialization(PolicydTest,
def test_003_test_overide_is_observed(self):
"""Test that the override is observed by the underlying service."""
if (openstack_utils.get_os_release() <
openstack_utils.get_os_release('groovy_victoria')):
raise unittest.SkipTest(
"Test skipped until Bug #1880959 is fix released")
if self._test_name is None:
logging.info("Doing policyd override for {}"
.format(self._service_name))
@@ -655,7 +659,7 @@ class HeatTests(BasePolicydSpecialization):
class OctaviaTests(BasePolicydSpecialization):
"""Test the policyd override using the octavia client."""
_rule = {'rule.yaml': "{'os_load-balancer_api:loadbalancer:get_one': '!'}"}
_rule = {'rule.yaml': "{'os_load-balancer_api:provider:get_all': '!'}"}
@classmethod
def setUpClass(cls, application_name=None):
@@ -663,89 +667,8 @@ class OctaviaTests(BasePolicydSpecialization):
super(OctaviaTests, cls).setUpClass(application_name="octavia")
cls.application_name = "octavia"
def setup_for_attempt_operation(self, ip):
"""Create a loadbalancer.
This is necessary so that the attempt is to show the load-balancer and
this is an operator that the policy can stop. Unfortunately, octavia,
whilst it has a policy for just listing load-balancers, unfortunately,
it doesn't work; whereas showing the load-balancer can be stopped.
NB this only works if the setup phase of the octavia tests have been
completed.
:param ip: the ip of for keystone.
:type ip: str
"""
logging.info("Setting up loadbalancer.")
auth = openstack_utils.get_overcloud_auth(address=ip)
sess = openstack_utils.get_keystone_session(auth)
octavia_client = openstack_utils.get_octavia_session_client(sess)
neutron_client = openstack_utils.get_neutron_session_client(sess)
if openstack_utils.dvr_enabled():
network_name = 'private_lb_fip_network'
else:
network_name = 'private'
resp = neutron_client.list_networks(name=network_name)
vip_subnet_id = resp['networks'][0]['subnets'][0]
res = octavia_client.load_balancer_create(
json={
'loadbalancer': {
'description': 'Created by Zaza',
'admin_state_up': True,
'vip_subnet_id': vip_subnet_id,
'name': 'zaza-lb-0',
}})
self.lb_id = res['loadbalancer']['id']
# now wait for it to get to the active state
@tenacity.retry(wait=tenacity.wait_fixed(1),
reraise=True, stop=tenacity.stop_after_delay(900))
def wait_for_lb_resource(client, resource_id):
resp = client.load_balancer_show(resource_id)
logging.info(resp['provisioning_status'])
assert resp['provisioning_status'] == 'ACTIVE', (
'load balancer resource has not reached '
'expected provisioning status: {}'
.format(resp))
return resp
logging.info('Awaiting loadbalancer to reach provisioning_status '
'"ACTIVE"')
resp = wait_for_lb_resource(octavia_client, self.lb_id)
logging.info(resp)
logging.info("Setup loadbalancer complete.")
def cleanup_for_attempt_operation(self, ip):
"""Remove the loadbalancer.
:param ip: the ip of for keystone.
:type ip: str
"""
logging.info("Deleting loadbalancer {}.".format(self.lb_id))
auth = openstack_utils.get_overcloud_auth(address=ip)
sess = openstack_utils.get_keystone_session(auth)
octavia_client = openstack_utils.get_octavia_session_client(sess)
octavia_client.load_balancer_delete(self.lb_id)
logging.info("Deleting loadbalancer in progress ...")
@tenacity.retry(wait=tenacity.wait_fixed(1),
reraise=True, stop=tenacity.stop_after_delay(900))
def wait_til_deleted(client, lb_id):
lb_list = client.load_balancer_list()
ids = [lb['id'] for lb in lb_list['loadbalancers']]
assert lb_id not in ids, 'load balancer still deleting'
wait_til_deleted(octavia_client, self.lb_id)
logging.info("Deleted loadbalancer.")
def get_client_and_attempt_operation(self, ip):
"""Attempt to show the loadbalancer as a policyd override.
"""Attempt to list available provider drivers.
This operation should pass normally, and fail when
the rule has been overriden (see the `rule` class variable.
@@ -757,6 +680,6 @@ class OctaviaTests(BasePolicydSpecialization):
octavia_client = openstack_utils.get_octavia_session_client(
self.get_keystone_session_admin_user(ip))
try:
octavia_client.load_balancer_show(self.lb_id)
octavia_client.provider_list()
except octaviaclient.OctaviaClientException:
raise PolicydOperationFailedException()

41
zaza/openstack/charm_tests/swift/tests.py
View File

@@ -178,15 +178,28 @@ class SwiftGlobalReplicationTests(test_utils.OpenStackBaseTest):
logging.info('Deleting container {}'.format(container['name']))
cls.swift_region1.delete_container(container['name'])
def test_two_regions_any_zones_two_replicas(self):
"""Create an object with two replicas across two regions."""
def test_901_two_regions_any_zones_two_replicas(self):
"""Create an object with two replicas across two regions.
We set write affinity to write the first copy in the local
region of the proxy used to perform the write, the other
replica will land in the remote region.
"""
swift_utils.apply_proxy_config(
self.region1_proxy_app,
{
'write-affinity': 'r1, r2',
'write-affinity': 'r1',
'write-affinity-node-count': '1',
'replicas': '2'},
self.region1_model_name)
swift_utils.apply_proxy_config(
self.region2_proxy_app,
{
'write-affinity': 'r2',
'write-affinity-node-count': '1',
'replicas': '2'},
self.region2_model_name)
logging.info('Proxy configs updated in both regions')
container_name, obj_name, obj_replicas = swift_utils.create_object(
self.swift_region1,
self.region1_proxy_app,
@@ -204,15 +217,29 @@ class SwiftGlobalReplicationTests(test_utils.OpenStackBaseTest):
len(obj_replicas.all_zones),
2)
def test_two_regions_any_zones_three_replicas(self):
"""Create an object with three replicas across two regions."""
def test_902_two_regions_any_zones_three_replicas(self):
"""Create an object with three replicas across two regions.
We set write affinity to write the first copy in the local
region of the proxy used to perform the write, at least one
of the other two replicas will end up in the opposite region
based on primary partitions in the ring.
"""
swift_utils.apply_proxy_config(
self.region1_proxy_app,
{
'write-affinity': 'r1, r2',
'write-affinity': 'r1',
'write-affinity-node-count': '1',
'replicas': '3'},
self.region1_model_name)
swift_utils.apply_proxy_config(
self.region2_proxy_app,
{
'write-affinity': 'r2',
'write-affinity-node-count': '1',
'replicas': '3'},
self.region2_model_name)
logging.info('Proxy configs updated in both regions')
container_name, obj_name, obj_replicas = swift_utils.create_object(
self.swift_region1,
self.region1_proxy_app,
@@ -258,7 +285,7 @@ class S3APITest(test_utils.OpenStackBaseTest):
# Create AWS compatible application credentials in Keystone
cls.ec2_creds = ks_client.ec2.create(user_id, project_id)
def test_s3_list_buckets(self):
def test_901_s3_list_buckets(self):
"""Use S3 API to list buckets."""
# We use a mix of the high- and low-level API with common arguments
kwargs = {

3
zaza/openstack/charm_tests/tempest/setup.py
View File

@@ -235,6 +235,9 @@ def get_tempest_context():
'cinder': add_cinder_config,
'keystone': add_keystone_config}
ctxt['enabled_services'] = get_service_list(keystone_session)
if set(['cinderv2', 'cinderv3']) \
.intersection(set(ctxt['enabled_services'])):
ctxt['enabled_services'].append('cinder')
ctxt['disabled_services'] = list(
set(TEMPEST_SVC_LIST) - set(ctxt['enabled_services']))
add_application_ips(ctxt)

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

@@ -14,13 +14,19 @@
"""Module containing base class for implementing charm tests."""
import contextlib
import logging
import ipaddress
import subprocess
import tenacity
import unittest
import novaclient
import zaza.model as model
import zaza.charm_lifecycle.utils as lifecycle_utils
import zaza.openstack.configure.guest as configure_guest
import zaza.openstack.utilities.openstack as openstack_utils
import zaza.openstack.utilities.generic as generic_utils
import zaza.openstack.charm_tests.glance.setup as glance_setup
def skipIfNotHA(service_name):
@@ -94,8 +100,7 @@ class BaseCharmTest(unittest.TestCase):
run_resource_cleanup = False
@classmethod
def resource_cleanup(cls):
def resource_cleanup(self):
"""Cleanup any resources created during the test run.
Override this method with a method which removes any resources
@@ -105,12 +110,13 @@ class BaseCharmTest(unittest.TestCase):
"""
pass
@classmethod
def tearDown(cls):
# this must be a class instance method otherwise descentents will not be
# able to influence if cleanup should be run.
def tearDown(self):
"""Run teardown for test class."""
if cls.run_resource_cleanup:
if self.run_resource_cleanup:
logging.info('Running resource cleanup')
cls.resource_cleanup()
self.resource_cleanup()
@classmethod
def setUpClass(cls, application_name=None, model_alias=None):
@@ -431,3 +437,124 @@ class OpenStackBaseTest(BaseCharmTest):
cls.keystone_session = openstack_utils.get_overcloud_keystone_session(
model_name=cls.model_name)
cls.cacert = openstack_utils.get_cacert()
cls.nova_client = (
openstack_utils.get_nova_session_client(cls.keystone_session))
def resource_cleanup(self):
"""Remove test resources."""
try:
logging.info('Removing instances launched by test ({}*)'
.format(self.RESOURCE_PREFIX))
for server in self.nova_client.servers.list():
if server.name.startswith(self.RESOURCE_PREFIX):
openstack_utils.delete_resource(
self.nova_client.servers,
server.id,
msg="server")
except AttributeError:
# Test did not define self.RESOURCE_PREFIX, ignore.
pass
def launch_guest(self, guest_name, userdata=None):
"""Launch two guests to use in tests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
Also note that this method will remove any already existing instance
with same name as what is requested.
:param guest_name: Name of instance
:type guest_name: str
:param userdata: Userdata to attach to instance
:type userdata: Optional[str]
:returns: Nova instance objects
:rtype: Server
"""
instance_name = '{}-{}'.format(self.RESOURCE_PREFIX, guest_name)
instance = self.retrieve_guest(instance_name)
if instance:
logging.info('Removing already existing instance ({}) with '
'requested name ({})'
.format(instance.id, instance_name))
openstack_utils.delete_resource(
self.nova_client.servers,
instance.id,
msg="server")
return configure_guest.launch_instance(
glance_setup.LTS_IMAGE_NAME,
vm_name=instance_name,
userdata=userdata)
def launch_guests(self, userdata=None):
"""Launch two guests to use in tests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
:param userdata: Userdata to attach to instance
:type userdata: Optional[str]
:returns: List of launched Nova instance objects
:rtype: List[Server]
"""
launched_instances = []
for guest_number in range(1, 2+1):
for attempt in tenacity.Retrying(
stop=tenacity.stop_after_attempt(3),
wait=tenacity.wait_exponential(
multiplier=1, min=2, max=10)):
with attempt:
launched_instances.append(
self.launch_guest(
guest_name='ins-{}'.format(guest_number),
userdata=userdata))
return launched_instances
def retrieve_guest(self, guest_name):
"""Return guest matching name.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
try:
return self.nova_client.servers.find(name=guest_name)
except novaclient.exceptions.NotFound:
return None
def retrieve_guests(self):
"""Return test guests.
Note that it is up to the caller to have set the RESOURCE_PREFIX class
variable prior to calling this method.
:param nova_client: Nova client to use when checking status
:type nova_client: Nova client
:returns: the matching guest
:rtype: Union[novaclient.Server, None]
"""
instance_1 = self.retrieve_guest(
'{}-ins-1'.format(self.RESOURCE_PREFIX))
instance_2 = self.retrieve_guest(
'{}-ins-1'.format(self.RESOURCE_PREFIX))
return instance_1, instance_2
def format_addr(addr):
"""Validate and format IP address.
:param addr: IPv6 or IPv4 address
:type addr: str
:returns: Address string, optionally encapsulated in brackets([])
:rtype: str
:raises: ValueError
"""
ipaddr = ipaddress.ip_address(addr)
if isinstance(ipaddr, ipaddress.IPv6Address):
fmt = '[{}]'
else:
fmt = '{}'
return fmt.format(ipaddr)

19
zaza/openstack/charm_tests/vault/setup.py
View File

@@ -16,6 +16,7 @@
import base64
import functools
import logging
import requests
import tempfile
@@ -99,7 +100,7 @@ async def async_mojo_unseal_by_unit():
unit_name, './hooks/update-status')
def auto_initialize(cacert=None, validation_application='keystone'):
def auto_initialize(cacert=None, validation_application='keystone', wait=True):
"""Auto initialize vault for testing.
Generate a csr and uploading a signed certificate.
@@ -114,6 +115,7 @@ def auto_initialize(cacert=None, validation_application='keystone'):
:returns: None
:rtype: None
"""
logging.info('Running auto_initialize')
basic_setup(cacert=cacert, unseal_and_authorize=True)
action = vault_utils.run_get_csr()
@@ -131,10 +133,11 @@ def auto_initialize(cacert=None, validation_application='keystone'):
root_ca=cacertificate,
allowed_domains='openstack.local')
zaza.model.wait_for_agent_status()
test_config = lifecycle_utils.get_charm_config(fatal=False)
zaza.model.wait_for_application_states(
states=test_config.get('target_deploy_status', {}))
if wait:
zaza.model.wait_for_agent_status()
test_config = lifecycle_utils.get_charm_config(fatal=False)
zaza.model.wait_for_application_states(
states=test_config.get('target_deploy_status', {}))
if validation_application:
validate_ca(cacertificate, application=validation_application)
@@ -163,6 +166,12 @@ auto_initialize_no_validation = functools.partial(
validation_application=None)
auto_initialize_no_validation_no_wait = functools.partial(
auto_initialize,
validation_application=None,
wait=False)
def validate_ca(cacertificate, application="keystone", port=5000):
"""Validate Certificate Authority against application.

3
zaza/openstack/charm_tests/vault/utils.py
View File

@@ -27,6 +27,7 @@ import yaml
import collections
import zaza.model
import zaza.openstack.charm_tests.test_utils as test_utils
AUTH_FILE = "vault_tests.yaml"
CharmVaultClient = collections.namedtuple(
@@ -101,7 +102,7 @@ def get_unit_api_url(ip):
transport = 'http'
if vault_config['ssl-cert']['value']:
transport = 'https'
return '{}://{}:8200'.format(transport, ip)
return '{}://{}:8200'.format(transport, test_utils.format_addr(ip))
def get_hvac_client(vault_url, cacert=None):

34
zaza/openstack/configure/pre_deploy_certs.py
View File

@@ -14,19 +14,19 @@ def set_cidr_certs():
"""Create certs and keys for deploy using IP SANS from CIDR.
Create a certificate authority certificate and key. The CA cert and key
are then base 64 encoded and assigned to the OS_TEST_CAKEY and
OS_TEST_CACERT environment variables.
are then base 64 encoded and assigned to the TEST_CAKEY and
TEST_CACERT environment variables.
Using the CA key a second certificate and key are generated. The new
certificate has a SAN entry for the first 2^11 IPs in the CIDR.
The cert and key are then base 64 encoded and assigned to the OS_TEST_KEY
and OS_TEST_CERT environment variables.
The cert and key are then base 64 encoded and assigned to the TEST_KEY
and TEST_CERT environment variables.
"""
(cakey, cacert) = zaza.openstack.utilities.cert.generate_cert(
ISSUER_NAME,
generate_ca=True)
os.environ['OS_TEST_CAKEY'] = base64.b64encode(cakey).decode()
os.environ['OS_TEST_CACERT'] = base64.b64encode(cacert).decode()
os.environ['TEST_CAKEY'] = base64.b64encode(cakey).decode()
os.environ['TEST_CACERT'] = base64.b64encode(cacert).decode()
# We need to restrain the number of SubjectAlternativeNames we attempt to
# put # in the certificate. There is a hard limit for what length the sum
# of all extensions in the certificate can have.
@@ -34,37 +34,37 @@ def set_cidr_certs():
# - 2^11 ought to be enough for anybody
alt_names = []
for addr in itertools.islice(
ipaddress.IPv4Network(os.environ.get('OS_CIDR_EXT')), 2**11):
ipaddress.IPv4Network(os.environ.get('TEST_CIDR_EXT')), 2**11):
alt_names.append(str(addr))
(key, cert) = zaza.openstack.utilities.cert.generate_cert(
'*.serverstack',
alternative_names=alt_names,
issuer_name=ISSUER_NAME,
signing_key=cakey)
os.environ['OS_TEST_KEY'] = base64.b64encode(key).decode()
os.environ['OS_TEST_CERT'] = base64.b64encode(cert).decode()
os.environ['TEST_KEY'] = base64.b64encode(key).decode()
os.environ['TEST_CERT'] = base64.b64encode(cert).decode()
def set_certs_per_vips():
"""Create certs and keys for deploy using VIPS.
Create a certificate authority certificate and key. The CA cert and key
are then base 64 encoded and assigned to the OS_TEST_CAKEY and
OS_TEST_CACERT environment variables.
are then base 64 encoded and assigned to the TEST_CAKEY and
TEST_CACERT environment variables.
Using the CA key a certificate and key is generated for each VIP specified
via environment variables. eg if OS_VIP06=172.20.0.107 is set in the
via environment variables. eg if TEST_VIP06=172.20.0.107 is set in the
environment then a cert with a SAN entry for 172.20.0.107 is generated.
The cert and key are then base 64 encoded and assigned to the OS_VIP06_KEY
and OS_VIP06_CERT environment variables.
The cert and key are then base 64 encoded and assigned to the
TEST_VIP06_KEY and TEST_VIP06_CERT environment variables.
"""
(cakey, cacert) = zaza.openstack.utilities.cert.generate_cert(
ISSUER_NAME,
generate_ca=True)
os.environ['OS_TEST_CAKEY'] = base64.b64encode(cakey).decode()
os.environ['OS_TEST_CACERT'] = base64.b64encode(cacert).decode()
os.environ['TEST_CAKEY'] = base64.b64encode(cakey).decode()
os.environ['TEST_CACERT'] = base64.b64encode(cacert).decode()
for vip_name, vip_ip in os.environ.items():
if vip_name.startswith('OS_VIP'):
if vip_name.startswith('TEST_VIP'):
(key, cert) = zaza.openstack.utilities.cert.generate_cert(
'*.serverstack',
alternative_names=[vip_ip],

2
zaza/openstack/utilities/generic.py
View File

@@ -23,9 +23,9 @@ import telnetlib
import yaml
from zaza import model
from zaza.openstack.utilities import juju as juju_utils
from zaza.openstack.utilities import exceptions as zaza_exceptions
from zaza.openstack.utilities.os_versions import UBUNTU_OPENSTACK_RELEASE
from zaza.utilities import juju as juju_utils
def assertActionRanOK(action):

237
zaza/openstack/utilities/juju.py
View File

@@ -13,18 +13,31 @@
# 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.
"""Module for interacting with juju."""
import os
from pathlib import Path
import yaml
"""Deprecated, please use zaza.utilities.juju."""
from zaza import (
model,
controller,
)
from zaza.openstack.utilities import generic as generic_utils
import logging
import functools
import zaza.utilities.juju
def deprecate():
"""Add a deprecation warning to wrapped function."""
def wrap(f):
@functools.wraps(f)
def wrapped_f(*args, **kwargs):
msg = (
"{} from zaza.openstack.utilities.juju is deprecated. "
"Please use the equivalent from zaza.utilities.juju".format(
f.__name__))
logging.warning(msg)
return f(*args, **kwargs)
return wrapped_f
return wrap
@deprecate()
def get_application_status(application=None, unit=None, model_name=None):
"""Return the juju status for an application.
@@ -37,39 +50,29 @@ def get_application_status(application=None, unit=None, model_name=None):
:returns: Juju status output for an application
:rtype: dict
"""
status = get_full_juju_status()
if unit and not application:
application = unit.split("/")[0]
if application:
status = status.applications.get(application)
if unit:
status = status.get("units").get(unit)
return status
return zaza.utilities.juju.get_application_status(
application=application,
unit=unit,
model_name=model_name)
def get_application_ip(application):
@deprecate()
def get_application_ip(application, model_name=None):
"""Get the application's IP address.
:param application: Application name
:type application: str
:param model_name: Name of model to query.
:type model_name: str
:returns: Application's IP address
:rtype: str
"""
try:
app_config = model.get_application_config(application)
except KeyError:
return ''
vip = app_config.get("vip").get("value")
if vip:
ip = vip
else:
unit = model.get_units(application)[0]
ip = unit.public_address
return ip
return zaza.utilities.juju.get_application_ip(
application,
model_name=model_name)
@deprecate()
def get_cloud_configs(cloud=None):
"""Get cloud configuration from local clouds.yaml.
@@ -81,14 +84,11 @@ def get_cloud_configs(cloud=None):
:returns: Dictionary of cloud configuration
:rtype: dict
"""
home = str(Path.home())
cloud_config = os.path.join(home, ".local", "share", "juju", "clouds.yaml")
if cloud:
return generic_utils.get_yaml_config(cloud_config)["clouds"].get(cloud)
else:
return generic_utils.get_yaml_config(cloud_config)
return zaza.utilities.juju.get_cloud_configs(
cloud=cloud)
@deprecate()
def get_full_juju_status(model_name=None):
"""Return the full juju status output.
@@ -97,10 +97,11 @@ def get_full_juju_status(model_name=None):
:returns: Full juju status output
:rtype: dict
"""
status = model.get_status(model_name=model_name)
return status
return zaza.utilities.juju.get_full_juju_status(
model_name=model_name)
@deprecate()
def get_machines_for_application(application, model_name=None):
"""Return machines for a given application.
@@ -111,20 +112,12 @@ def get_machines_for_application(application, model_name=None):
:returns: machines for an application
:rtype: Iterator[str]
"""
status = get_application_status(application, model_name=model_name)
if not status:
return
# libjuju juju status no longer has units for subordinate charms
# Use the application it is subordinate-to to find machines
if not status.get("units") and status.get("subordinate-to"):
status = get_application_status(status.get("subordinate-to")[0],
model_name=model_name)
for unit in status.get("units").keys():
yield status.get("units").get(unit).get("machine")
return zaza.utilities.juju.get_machines_for_application(
application,
model_name=model_name)
@deprecate()
def get_unit_name_from_host_name(host_name, application, model_name=None):
"""Return the juju unit name corresponding to a hostname.
@@ -135,17 +128,13 @@ def get_unit_name_from_host_name(host_name, application, model_name=None):
:param model_name: Name of model to query.
:type model_name: str
"""
# Assume that a juju managed hostname always ends in the machine number and
# remove the domain name if it present.
machine_number = host_name.split('-')[-1].split('.')[0]
unit_names = [
u.entity_id
for u in model.get_units(application_name=application,
model_name=model_name)
if int(u.data['machine-id']) == int(machine_number)]
return unit_names[0]
return zaza.utilities.juju.get_unit_name_from_host_name(
host_name,
application,
model_name=model_name)
@deprecate()
def get_machine_status(machine, key=None, model_name=None):
"""Return the juju status for a machine.
@@ -158,17 +147,13 @@ def get_machine_status(machine, key=None, model_name=None):
:returns: Juju status output for a machine
:rtype: dict
"""
status = get_full_juju_status(model_name=model_name)
if "lxd" in machine:
host = machine.split('/')[0]
status = status.machines.get(host)['containers'][machine]
else:
status = status.machines.get(machine)
if key:
status = status.get(key)
return status
return zaza.utilities.juju.get_machine_status(
machine,
key=key,
model_name=model_name)
@deprecate()
def get_machine_series(machine, model_name=None):
"""Return the juju series for a machine.
@@ -179,13 +164,12 @@ def get_machine_series(machine, model_name=None):
:returns: Juju series
:rtype: string
"""
return get_machine_status(
machine=machine,
key='series',
model_name=model_name
)
return zaza.utilities.juju.get_machine_series(
machine,
model_name=model_name)
@deprecate()
def get_machine_uuids_for_application(application, model_name=None):
"""Return machine uuids for a given application.
@@ -196,30 +180,22 @@ def get_machine_uuids_for_application(application, model_name=None):
:returns: machine uuuids for an application
:rtype: Iterator[str]
"""
for machine in get_machines_for_application(application,
model_name=model_name):
yield get_machine_status(machine, key="instance-id",
model_name=model_name)
return zaza.utilities.juju.get_machine_uuids_for_application(
application,
model_name=model_name)
@deprecate()
def get_provider_type():
"""Get the type of the undercloud.
:returns: Name of the undercloud type
:rtype: string
"""
cloud = controller.get_cloud()
if cloud:
# If the controller was deployed from this system with
# the cloud configured in ~/.local/share/juju/clouds.yaml
# Determine the cloud type directly
return get_cloud_configs(cloud)["type"]
else:
# If the controller was deployed elsewhere
# For now assume openstack
return "openstack"
return zaza.utilities.juju.get_provider_type()
@deprecate()
def remote_run(unit, remote_cmd, timeout=None, fatal=None, model_name=None):
"""Run command on unit and return the output.
@@ -238,46 +214,15 @@ def remote_run(unit, remote_cmd, timeout=None, fatal=None, model_name=None):
:rtype: string
:raises: model.CommandRunFailed
"""
if fatal is None:
fatal = True
result = model.run_on_unit(
return zaza.utilities.juju.remote_run(
unit,
remote_cmd,
timeout=timeout,
fatal=fatal,
model_name=model_name)
if result:
if int(result.get("Code")) == 0:
return result.get("Stdout")
else:
if fatal:
raise model.CommandRunFailed(remote_cmd, result)
return result.get("Stderr")
def _get_unit_names(names, model_name=None):
"""Resolve given application names to first unit name of said application.
Helper function that resolves application names to first unit name of
said application. Any already resolved unit names are returned as-is.
:param names: List of units/applications to translate
:type names: list(str)
:param model_name: Name of model to query.
:type model_name: str
:returns: List of units
:rtype: list(str)
"""
result = []
for name in names:
if '/' in name:
result.append(name)
else:
result.append(model.get_first_unit_name(
name,
model_name=model_name))
return result
@deprecate()
def get_relation_from_unit(entity, remote_entity, remote_interface_name,
model_name=None):
"""Get relation data passed between two units.
@@ -302,22 +247,14 @@ def get_relation_from_unit(entity, remote_entity, remote_interface_name,
:rtype: dict
:raises: model.CommandRunFailed
"""
application = entity.split('/')[0]
remote_application = remote_entity.split('/')[0]
rid = model.get_relation_id(application, remote_application,
remote_interface_name=remote_interface_name,
model_name=model_name)
(unit, remote_unit) = _get_unit_names(
[entity, remote_entity],
return zaza.utilities.juju.get_relation_from_unit(
entity,
remote_entity,
remote_interface_name,
model_name=model_name)
cmd = 'relation-get --format=yaml -r "{}" - "{}"' .format(rid, remote_unit)
result = model.run_on_unit(unit, cmd, model_name=model_name)
if result and int(result.get('Code')) == 0:
return yaml.safe_load(result.get('Stdout'))
else:
raise model.CommandRunFailed(cmd, result)
@deprecate()
def leader_get(application, key='', model_name=None):
"""Get leader settings from leader unit of named application.
@@ -331,14 +268,13 @@ def leader_get(application, key='', model_name=None):
:rtype: dict
:raises: model.CommandRunFailed
"""
cmd = 'leader-get --format=yaml {}'.format(key)
result = model.run_on_leader(application, cmd, model_name=model_name)
if result and int(result.get('Code')) == 0:
return yaml.safe_load(result.get('Stdout'))
else:
raise model.CommandRunFailed(cmd, result)
return zaza.utilities.juju.leader_get(
application,
key=key,
model_name=model_name)
@deprecate()
def get_subordinate_units(unit_list, charm_name=None, status=None,
model_name=None):
"""Get a list of all subordinate units associated with units in unit_list.
@@ -369,17 +305,8 @@ def get_subordinate_units(unit_list, charm_name=None, status=None,
:returns: List of matching unit names.
:rtype: []
"""
if not status:
status = model.get_status(model_name=model_name)
sub_units = []
for unit_name in unit_list:
app_name = unit_name.split('/')[0]
subs = status.applications[app_name]['units'][unit_name].get(
'subordinates') or {}
if charm_name:
for unit_name, unit_data in subs.items():
if charm_name in unit_data['charm']:
sub_units.append(unit_name)
else:
sub_units.extend([n for n in subs.keys()])
return sub_units
return zaza.utilities.juju.get_subordinate_units(
unit_list,
charm_name=charm_name,
status=status,
model_name=model_name)

36
zaza/openstack/utilities/openstack.py
View File

@@ -40,6 +40,7 @@ from keystoneauth1.identity import (
)
import zaza.openstack.utilities.cert as cert
import zaza.utilities.deployment_env as deployment_env
import zaza.utilities.juju as juju_utils
from novaclient import client as novaclient_client
from neutronclient.v2_0 import client as neutronclient
from neutronclient.common import exceptions as neutronexceptions
@@ -66,7 +67,6 @@ from zaza import model
from zaza.openstack.utilities import (
exceptions,
generic as generic_utils,
juju as juju_utils,
)
CIRROS_RELEASE_URL = 'http://download.cirros-cloud.net/version/released'
@@ -163,7 +163,7 @@ WORKLOAD_STATUS_EXCEPTIONS = {
KEYSTONE_CACERT = "keystone_juju_ca_cert.crt"
KEYSTONE_REMOTE_CACERT = (
"/usr/local/share/ca-certificates/{}".format(KEYSTONE_CACERT))
KEYSTONE_LOCAL_CACERT = ("/tmp/{}".format(KEYSTONE_CACERT))
KEYSTONE_LOCAL_CACERT = ("tests/{}".format(KEYSTONE_CACERT))
def get_cacert():
@@ -1609,7 +1609,7 @@ def get_undercloud_auth():
'API_VERSION': 3,
}
if domain:
auth_settings['OS_DOMAIN_NAME': 'admin_domain'] = domain
auth_settings['OS_DOMAIN_NAME'] = domain
else:
auth_settings['OS_USER_DOMAIN_NAME'] = (
os.environ.get('OS_USER_DOMAIN_NAME'))
@@ -1621,6 +1621,10 @@ def get_undercloud_auth():
if os_project_id is not None:
auth_settings['OS_PROJECT_ID'] = os_project_id
_os_cacert = os.environ.get('OS_CACERT')
if _os_cacert:
auth_settings.update({'OS_CACERT': _os_cacert})
# Validate settings
for key, settings in list(auth_settings.items()):
if settings is None:
@@ -1733,6 +1737,15 @@ def get_overcloud_auth(address=None, model_name=None):
}
if tls_rid:
unit = model.get_first_unit_name('keystone', model_name=model_name)
# ensure that the path to put the local cacert in actually exists. The
# assumption that 'tests/' exists for, say, mojo is false.
# Needed due to:
# commit: 537473ad3addeaa3d1e4e2d0fd556aeaa4018eb2
_dir = os.path.dirname(KEYSTONE_LOCAL_CACERT)
if not os.path.exists(_dir):
os.makedirs(_dir)
model.scp_from_unit(
unit,
KEYSTONE_REMOTE_CACERT,
@@ -2032,7 +2045,8 @@ def upload_image_to_glance(glance, local_path, image_name, disk_format='qcow2',
return image
def create_image(glance, image_url, image_name, image_cache_dir=None, tags=[]):
def create_image(glance, image_url, image_name, image_cache_dir=None, tags=[],
properties=None):
"""Download the image and upload it to glance.
Download an image from image_url and upload it to glance labelling
@@ -2049,6 +2063,8 @@ def create_image(glance, image_url, image_name, image_cache_dir=None, tags=[]):
:type image_cache_dir: Option[str, None]
:param tags: Tags to add to image
:type tags: list of str
:param properties: Properties and values to add to image
:type properties: dict
:returns: glance image pointer
:rtype: glanceclient.common.utils.RequestIdProxy
"""
@@ -2070,6 +2086,11 @@ def create_image(glance, image_url, image_name, image_cache_dir=None, tags=[]):
logging.debug(
'applying tag to image: glance.image_tags.update({}, {}) = {}'
.format(image.id, tags, result))
logging.info("Setting image properties: {}".format(properties))
if properties:
result = glance.images.update(image.id, **properties)
return image
@@ -2198,7 +2219,8 @@ def get_private_key_file(keypair_name):
:returns: Path to file containing key
:rtype: str
"""
return 'tests/id_rsa_{}'.format(keypair_name)
tmp_dir = deployment_env.get_tmpdir()
return '{}/id_rsa_{}'.format(tmp_dir, keypair_name)
def write_private_key(keypair_name, key):
@@ -2279,7 +2301,7 @@ def get_ports_from_device_id(neutron_client, device_id):
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=120),
reraise=True, stop=tenacity.stop_after_attempt(12))
reraise=True, stop=tenacity.stop_after_delay(1800))
def cloud_init_complete(nova_client, vm_id, bootstring):
"""Wait for cloud init to complete on the given vm.
@@ -2396,7 +2418,7 @@ def ssh_command(username,
ssh.connect(ip, username=username, password=password)
else:
key = paramiko.RSAKey.from_private_key(io.StringIO(privkey))
ssh.connect(ip, username=username, password='', pkey=key)
ssh.connect(ip, username=username, password=None, pkey=key)
logging.info("Running {} on {}".format(command, vm_name))
stdin, stdout, stderr = ssh.exec_command(command)
if verify and callable(verify):

23
zaza/openstack/utilities/os_versions.py
View File

@@ -35,6 +35,7 @@ UBUNTU_OPENSTACK_RELEASE = OrderedDict([
('disco', 'stein'),
('eoan', 'train'),
('focal', 'ussuri'),
('groovy', 'victoria'),
])
@@ -57,6 +58,7 @@ OPENSTACK_CODENAMES = OrderedDict([
('2019.1', 'stein'),
('2019.2', 'train'),
('2020.1', 'ussuri'),
('2020.2', 'victoria'),
])
OPENSTACK_RELEASES_PAIRS = [
@@ -66,7 +68,8 @@ OPENSTACK_RELEASES_PAIRS = [
'xenial_pike', 'artful_pike', 'xenial_queens',
'bionic_queens', 'bionic_rocky', 'cosmic_rocky',
'bionic_stein', 'disco_stein', 'bionic_train',
'eoan_train', 'bionic_ussuri', 'focal_ussuri']
'eoan_train', 'bionic_ussuri', 'focal_ussuri',
'focal_victoria', 'groovy_victoria']
# The ugly duckling - must list releases oldest to newest
SWIFT_CODENAMES = OrderedDict([
@@ -105,7 +108,9 @@ SWIFT_CODENAMES = OrderedDict([
('train',
['2.22.0']),
('ussuri',
['2.24.0']),
['2.24.0', '2.25.0']),
('victoria',
['2.25.0']),
])
# >= Liberty version->codename mapping
@@ -121,6 +126,7 @@ PACKAGE_CODENAMES = {
('19', 'stein'),
('20', 'train'),
('21', 'ussuri'),
('22', 'victoria'),
]),
'neutron-common': OrderedDict([
('7', 'liberty'),
@@ -133,6 +139,7 @@ PACKAGE_CODENAMES = {
('14', 'stein'),
('15', 'train'),
('16', 'ussuri'),
('17', 'victoria'),
]),
'cinder-common': OrderedDict([
('7', 'liberty'),
@@ -145,6 +152,7 @@ PACKAGE_CODENAMES = {
('14', 'stein'),
('15', 'train'),
('16', 'ussuri'),
('17', 'victoria'),
]),
'keystone': OrderedDict([
('8', 'liberty'),
@@ -157,6 +165,7 @@ PACKAGE_CODENAMES = {
('15', 'stein'),
('16', 'train'),
('17', 'ussuri'),
('18', 'victoria'),
]),
'horizon-common': OrderedDict([
('8', 'liberty'),
@@ -168,7 +177,8 @@ PACKAGE_CODENAMES = {
('14', 'rocky'),
('15', 'stein'),
('16', 'train'),
('17', 'ussuri'),
('18', 'ussuri'),
('19', 'victoria'),
]),
'ceilometer-common': OrderedDict([
('5', 'liberty'),
@@ -181,6 +191,7 @@ PACKAGE_CODENAMES = {
('12', 'stein'),
('13', 'train'),
('14', 'ussuri'),
('15', 'victoria'),
]),
'heat-common': OrderedDict([
('5', 'liberty'),
@@ -193,6 +204,7 @@ PACKAGE_CODENAMES = {
('12', 'stein'),
('13', 'train'),
('14', 'ussuri'),
('15', 'victoria'),
]),
'glance-common': OrderedDict([
('11', 'liberty'),
@@ -205,6 +217,7 @@ PACKAGE_CODENAMES = {
('18', 'stein'),
('19', 'train'),
('20', 'ussuri'),
('21', 'victoria'),
]),
'openstack-dashboard': OrderedDict([
('8', 'liberty'),
@@ -216,7 +229,8 @@ PACKAGE_CODENAMES = {
('14', 'rocky'),
('15', 'stein'),
('16', 'train'),
('17', 'ussuri'),
('18', 'ussuri'),
('19', 'victoria'),
]),
'designate-common': OrderedDict([
('1', 'liberty'),
@@ -229,6 +243,7 @@ PACKAGE_CODENAMES = {
('8', 'stein'),
('9', 'train'),
('10', 'ussuri'),
('11', 'victoria'),
]),
'ovn-common': OrderedDict([
('2', 'train'),