#!/usr/bin/python3
"""
// © University of Southampton IT Innovation Centre, 2018
//
// Copyright in this software belongs to University of Southampton
// IT Innovation Centre of Gamma House, Enterprise Road,
// Chilworth Science Park, Southampton, SO16 7NS, UK.
//
// This software may not be used, sold, licensed, transferred, copied
// or reproduced in whole or in part in any manner or form or in or
// on any media by any person other than in accordance with the terms
// of the Licence Agreement supplied with the software, or otherwise
// without the prior written consent of the copyright owners.
//
// This software is distributed WITHOUT ANY WARRANTY, without even the
// implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
// PURPOSE, except where stated in the Licence Agreement supplied with
// the software.
//
// Created By : Nikolay Stanchev
// Created Date : 14-08-2018
// Created for Project : FLAME
"""
# Python standard libs
from os import listdir
from os.path import isfile, join
from urllib.parse import urlparse
# PIP installed libs
import pytest
from yaml import load
from pyramid import testing
from requests import get, delete
from toscaparser.tosca_template import ToscaTemplate
# CLMC-service imports
from clmcservice.alertsapi.utilities import adjust_tosca_definitions_import
from clmcservice.alertsapi.alerts_specification_schema import validate_clmc_alerts_specification
from clmcservice.alertsapi.views import AlertsConfigurationAPI
from clmcservice import ROOT_DIR
class TestAlertsConfigurationAPI(object):
"""
A pytest-implementation test for the Alerts Configuration API endpoints.
"""
@pytest.fixture(autouse=True)
def print_fixture(self):
"""
Fixture to adjust the printing format when running pytest with the "-s" flag - by default print messages mix up with pytest's output
"""
print()
@pytest.fixture()
def app_config(self):
"""
A fixture to implement setUp/tearDown functionality for all tests by initializing configuration structure for the web service
"""
self.registry = testing.setUp()
yield
testing.tearDown()
def test_alerts_config_tosca_parsing(self):
"""
Tests that what we consider a valid/invalid alerts specification is successfully/unsuccessfully parsed by the TOSCA-parser.
"""
for path_suffix, valid_expected in (("valid", True), ("invalid", False)):
test_data_path = join(ROOT_DIR, *["resources", "tosca", "test-data", "tosca-parser", path_suffix])
for test_file_path in listdir(test_data_path):
alert_config_abs_path = join(test_data_path, test_file_path)
if not isfile(alert_config_abs_path):
continue # skip directories
if not test_file_path.lower().endswith('.yaml'):
continue # non-yaml files are not intended for being tested
print(alert_config_abs_path, valid_expected)
with open(alert_config_abs_path, 'r') as fh:
yaml_content = load(fh)
adjust_tosca_definitions_import(yaml_content)
valid_real = True
try:
ToscaTemplate(yaml_dict_tpl=yaml_content)
except Exception:
valid_real = False
assert valid_expected == valid_real, "TOSCA parser test failed for file: {0}".format(alert_config_abs_path)
def test_alerts_config_clmc_validation(self):
"""
Tests the custom CLMC validation of the TOSCA alerts specification.
"""
for path_suffix, valid_expected in (("valid", True), ("invalid", False)):
test_data_path = join(ROOT_DIR, *["resources", "tosca", "test-data", "clmc-validator", path_suffix])
for test_file_path in listdir(test_data_path):
alert_config_abs_path = join(test_data_path, test_file_path)
if not isfile(alert_config_abs_path):
continue # skip directories
if not test_file_path.lower().endswith('.yaml'):
continue # non-yaml files are not intended for being tested
print(alert_config_abs_path, valid_expected)
with open(alert_config_abs_path, 'r') as fh:
yaml_content = load(fh)
adjust_tosca_definitions_import(yaml_content)
# do not catch exceptions here since we are testing the clmc validator, the tosca parsing is tested in the previous test method
alert_tosca_spec = ToscaTemplate(yaml_dict_tpl=yaml_content)
valid_real, err = validate_clmc_alerts_specification(alert_tosca_spec.tpl, include_error=True)
assert valid_expected == valid_real, "CLMC alerts specification validator test failed for file: {0}".format(alert_config_abs_path)
def test_alerts_config_api_post(self, app_config):
"""
Tests the POST API endpoint of the alerts configuration API responsible for receiving alerts specifications.
Test steps are:
* Traverse all valid TOSCA Alerts Specifications in the
src/service/clmcservice/resources/tosca/test-data/clmc-validator/valid and src/service/clmcservice/resources/tosca/test-data/tosca-parser/valid
* Sending a valid TOSCA Alert Specification to the view responsible for configuring Kapacitor
* Check that Kapacitor alerts, topics and handlers are created with the correct identifier and arguments
:param app_config: fixture for setUp/tearDown of the web service registry
"""
for test_folder in ("clmc-validator", "tosca-parser"):
test_data_path = join(ROOT_DIR, *["resources", "tosca", "test-data", test_folder, "valid"])
for test_file_path in listdir(test_data_path):
alert_spec_abs_path = join(test_data_path, test_file_path)
if not isfile(alert_spec_abs_path):
continue # skip directories
if not test_file_path.lower().endswith('.yaml'):
continue # non-yaml files are not intended for being tested
print("Testing file {0} in folder {1}".format(test_file_path, test_folder))
request = testing.DummyRequest()
with open(alert_spec_abs_path) as alert_spec:
sfc, sfc_instance, alert_ids, topic_handlers = extract_alert_spec_data(alert_spec)
alert_spec.seek(0)
request.POST['alert-spec'] = FieldStorageMock(test_file_path, alert_spec) # a simple mock class is used to mimic the FieldStorage class
clmc_service_response = AlertsConfigurationAPI(request).post_alerts_specification()
assert (sfc, sfc_instance) == (clmc_service_response["service_function_chain_id"], clmc_service_response["service_function_chain_instance_id"]), \
"Incorrect extraction of metadata for file {0}". format(test_file_path)
# traverse through all alert IDs and check that they are created within Kapacitor
for alert_id in alert_ids:
kapacitor_response = get("http://localhost:9092/kapacitor/v1/tasks/{0}".format(alert_id))
assert kapacitor_response.status_code == 200, "Alert with ID {0} was not created - test file {1}.".format(alert_id, test_file_path)
kapacitor_response_json = kapacitor_response.json()
assert "link" in kapacitor_response_json, "Incorrect response from kapacitor for alert with ID {0} - test file {1}".format(alert_id, test_file_path)
assert kapacitor_response_json["status"] == "enabled", "Alert with ID {0} was created but is disabled - test file {1}".format(alert_id, test_file_path)
# check that all topic IDs were registered within Kapacitor
topic_ids = list(topic_handlers.keys())
kapacitor_response = get("http://localhost:9092/kapacitor/v1/alerts/topics")
assert kapacitor_response.status_code == 200, "Kapacitor couldn't return the list of created topics - test file {0}".format(test_file_path)
kapacitor_response_json = kapacitor_response.json()
kapacitor_defined_topics = [topic["id"] for topic in kapacitor_response_json["topics"]]
assert set(topic_ids).issubset(kapacitor_defined_topics), "Not all topic IDs were created within kapacitor - test file {0}".format(test_file_path)
# check that all handler IDs were created and each of them is subscribed to the correct topic ID
for topic_id in topic_handlers:
for handler_id, handler_url in topic_handlers[topic_id]:
kapacitor_response = get("http://localhost:9092/kapacitor/v1/alerts/topics/{0}/handlers/{1}".format(topic_id, handler_id))
assert kapacitor_response.status_code == 200, "Handler with ID {0} for topic with ID {1} doesn't exist - test file {2}".format(handler_id, topic_id, test_file_path)
kapacitor_response_json = kapacitor_response.json()
assert kapacitor_response_json["id"] == handler_id, "Incorrect ID of handler {0} in the Kapacitor response - test file {1}".format(handler_id, test_file_path)
assert kapacitor_response_json["kind"] == "post", "Incorrect kind of handler {0} in the Kapacitor response - test file {1}".format(handler_id, test_file_path)
assert kapacitor_response_json["options"]["url"], "Incorrect url of handler {0} in the Kapacitor response - test file {1}".format(handler_id, test_file_path)
clear_kapacitor_alerts(alert_ids, topic_handlers)
class FieldStorageMock(object):
def __init__(self, filename, file):
"""
Used to mock the behaviour of the cgi.FieldStorage class - two attributes needed only to forward a file to the view.
:param filename: file name
:param file: file object
"""
self.filename = filename
self.file = file
def extract_alert_spec_data(alert_spec):
"""
A utility function to extract the expected alert, handler and topic identifiers from a given alert specification.
:param alert_spec: the alert specification file (file object)
:return: a tuple containing sfc_id and sfc_instance_id along with a list and a dictionary of generated IDs (alert IDs (list), topic IDs linked to handler IDs (dict))
"""
yaml_alert_spec = load(alert_spec)
adjust_tosca_definitions_import(yaml_alert_spec)
tosca_tpl = ToscaTemplate(yaml_dict_tpl=yaml_alert_spec)
sfc, sfc_instance = tosca_tpl.tpl["metadata"]["sfc"], tosca_tpl.tpl["metadata"]["sfci"]
alert_ids = [] # saves all alert IDs in a list
topic_handlers = {} # saves all topics in a dictionary, each topic is linked to a list of handler pairs (a handler pair consists of handler id and handler url)
for policy in tosca_tpl.policies:
policy_id = policy.name
for trigger in policy.triggers:
trigger_id = trigger.name
topic_id = "{0}.{1}.{2}".format(sfc, sfc_instance, trigger_id)
topic_handlers[topic_id] = []
alert_id = "{0}.{1}.{2}.{3}".format(sfc, sfc_instance, policy_id, trigger_id)
alert_ids.append(alert_id)
for handler_url in trigger.trigger_tpl["action"]["implementation"]:
handler_host = urlparse(handler_url).hostname
handler_id = "{0}.{1}.{2}".format(policy_id, trigger_id, handler_host)
topic_handlers[topic_id].append((handler_id, handler_url))
return sfc, sfc_instance, alert_ids, topic_handlers
def clear_kapacitor_alerts(alert_ids, topic_handlers):
"""
A utility function to clean up Kapacitor from the configured alerts, topics and handlers.
:param alert_ids: the list of alert IDs to delete
:param topic_handlers: the dictionary of topic and handlers to delete
"""
for alert_id in alert_ids:
kapacitor_response = delete("http://localhost:9092/kapacitor/v1/tasks/{0}".format(alert_id)) # delete alert
assert kapacitor_response.status_code == 204
for topic_id in topic_handlers:
for handler_id, handler_url in topic_handlers[topic_id]:
kapacitor_response = delete("http://localhost:9092/kapacitor/v1/alerts/topics/{0}/handlers/{1}".format(topic_id, handler_id)) # delete handler
assert kapacitor_response.status_code == 204
kapacitor_response = delete("http://localhost:9092/kapacitor/v1/alerts/topics/{0}".format(topic_id)) # delete topic
assert kapacitor_response.status_code == 204