Newer
Older
#!/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
# PIP installed libs
import pytest
from pyramid import testing
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
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