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

Add octavia policyd tests

Browse Source 
Providing a policy.d test for Octavia is a bit more complicated as the
obvious policy is just to prevent listing of load balancers;
unfortunately it doesn't work in the upstream, as the code doesn't check
the policy in octavia.  However, "showing" an load balancer does involve
a policy check and so the test is more complicated than 'normal' policy
tests in that a resource is created, then the policy override is checked
and then the resource is deleted.
This commit is contained in:
Alex Kavanagh
2019-11-07 18:34:46 +00:00
parent 57a7c8c90f
commit b00e570e57
2 changed files with 142 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
requirements.txt
View File

@@ -1,9 +1,8 @@
aiounittest
async_generator
juju
juju_wait
PyYAML<=4.2,>=3.0
PyYAML<=4.2,>=3.0
flake8>=2.2.4,<=3.5.0
flake8-docstrings
flake8-per-file-ignores
@@ -35,6 +34,7 @@ python-swiftclient
tenacity
distro-info
paramiko
# Documentation requirements
sphinx
sphinxcontrib-asyncio

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

@@ -35,13 +35,15 @@ import logging
import os
import shutil
import tempfile
import tenacity
import unittest
import zipfile
import keystoneauth1
import glanceclient.common.exceptions
from octaviaclient.api.v2 import octavia as octaviaclient
import cinderclient.exceptions
import heatclient.exc
import glanceclient.common.exceptions
import keystoneauth1
import zaza.model as zaza_model
@@ -285,6 +287,28 @@ class BasePolicydSpecialization(PolicydTest,
"not valid if {}.rule is not configured"
.format(cls.__name__))
def setup_for_attempt_operation(self, ip):
"""Set-up for the policy override if needed.
This method allows the test being performed in
get_client_and_attempt_operation() to have some setup done before the
test is performed. This is because the method
get_client_and_attempt_operation() is run twice; once to succeed and
once to fail.
:param ip: the ip of for keystone.
:type ip: str
"""
pass
def cleanup_for_attempt_operation(self, ip):
"""Clean-up after a successful (or not) policy override operation.
:param ip: the ip of for keystone.
:type ip: str
"""
pass
def get_client_and_attempt_operation(self, keystone_session):
"""Override this method to perform the operation.
@@ -385,6 +409,10 @@ class BasePolicydSpecialization(PolicydTest,
# note policyd override only works with Xenial-queens and so keystone
# is already v3
# Allow the overriden class to setup the environment before the policyd
# test is performed.
self.setup_for_attempt_operation(self.keystone_ips[0])
# verify that the operation works before performing the policyd
# override.
zaza_model.block_until_wl_status_info_starts_with(
@@ -394,6 +422,7 @@ class BasePolicydSpecialization(PolicydTest,
try:
self.get_client_and_attempt_operation(self.keystone_ips[0])
except Exception as e:
self.cleanup_for_attempt_operation(self.keystone_ips[0])
raise zaza_exceptions.PolicydError(
'Service action failed and should have passed. "{}"'
.format(str(e)))
@@ -418,6 +447,7 @@ class BasePolicydSpecialization(PolicydTest,
logging.info("exception was: {}".format(e.__class__.__name__))
import traceback
logging.info(traceback.format_exc())
self.cleanup_for_attempt_operation(self.keystone_ips[0])
raise zaza_exceptions.PolicydError(
'Service action failed in an unexpected way: {}'
.format(str(e)))
@@ -446,6 +476,8 @@ class BasePolicydSpecialization(PolicydTest,
'Service action failed and should have passed after removing '
'policy override: "{}"'
.format(str(e)))
finally:
self.cleanup_for_attempt_operation(self.keystone_ips[0])
logging.info('OK')
@@ -607,3 +639,109 @@ class HeatTests(BasePolicydSpecialization):
list(heat_client.stacks.list())
except heatclient.exc.HTTPForbidden:
raise PolicydOperationFailedException()
class OctaviaTests(BasePolicydSpecialization):
"""Test the policyd override using the octavia client."""
_rule = "{'os_load-balancer_api:loadbalancer:get_one': '!'}"
@classmethod
def setUpClass(cls, application_name=None):
"""Run class setup for running OctaviaTests charm operation tests."""
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)
resp = neutron_client.list_networks(name="private_lb_fip_network")
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.
This operation should pass normally, and fail when
the rule has been overriden (see the `rule` class variable.
:param ip: the IP address to get the session against.
:type ip: str
:raises: PolicydOperationFailedException if operation fails.
"""
octavia_client = openstack_utils.get_octavia_session_client(
self.get_keystone_session_admin_user(ip))
try:
octavia_client.load_balancer_show(self.lb_id)
except octaviaclient.OctaviaClientException:
raise PolicydOperationFailedException()