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 : 22-08-2018
## Created for Project : FLAME
"""
from requests import post, get
from os import listdir
from os.path import join, dirname
from json import load
from schema import Schema, And, Or, Optional, SchemaError
from clmctest.alerts.alert_handler_server import LOG_TEST_FOLDER_PATH
CLMC_SERVICE_PORT = 9080
NGINX_PORT = 80
38
39
40
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
def is_valid_timestamp(str_timestamp):
try:
strptime(str_timestamp, "%Y-%m-%dT%H:%M:%SZ")
return True
except ValueError:
return False
def is_valid_details_string(details):
return "db=" in details and "sfc=" in details and "sfci=" in details and "policy" in details
JSON_BODY_SCHEMA = Schema({
"message": "TRUE",
"id": str,
"level": "CRITICAL",
"duration": int,
"previousLevel": str,
"details": And(str, is_valid_details_string),
"time": And(str, is_valid_timestamp),
"data": {
"series": [
{
"name": str,
Optional("tags"): {
str: str
},
"columns": [
str
],
"values": [
[
Or(str, int)
]
]
}
]
}
})
class TestAlerts(object):
Nikolay Stanchev
committed
def test_alert_triggers(self, rspec_config, set_up_tear_down_fixture):
"""
Test is implemented using the following steps:
* Send clmc service a TOSCA alert spec. file
* Wait 15 seconds for Kapacitor to configure and start executing the defined tasks
* Send some test requests to nginx to increase the load
* Wait 20 seconds for alerts to be triggered
* Check that 4 log files have been created - one for each alert defined in the alert spec.
:param rspec_config: fixture from conftest.py
"""
global CLMC_SERVICE_PORT, NGINX_PORT, JSON_BODY_SCHEMA
clmc_service_host, nginx_host = None, None
for host in rspec_config:
if host["name"] == "clmc-service":
clmc_service_host = host["ip_address"]
elif host["name"] == "nginx":
nginx_host = host["ip_address"]
if clmc_service_host is not None and nginx_host is not None:
break
print("Sending alerts specification to clmc service...")
alerts_spec = join(dirname(__file__), "alerts_test_config.yaml")
resources_spec = join(dirname(__file__), "resources_test_config.yaml")
with open(alerts_spec, 'rb') as alerts:
with open(resources_spec, 'rb') as resources:
files = {'alert-spec': alerts, 'resource-spec': resources}
response = post("http://{0}:{1}/alerts".format(clmc_service_host, CLMC_SERVICE_PORT), files=files)
assert response.status_code == 200
clmc_service_response = response.json()
assert "triggers_specification_errors" not in clmc_service_response, "Unexpected error was returned for triggers specification"
assert "triggers_action_errors" not in clmc_service_response, "Unexpected error was returned for handlers specification"
print("Alert spec sent successfully")
print("Wait 10 seconds for Kapacitor stream/batch tasks to start working...")
sleep(10)
print("Sending test requests to nginx...")
response = get("http://{0}:{1}/".format(nginx_host, NGINX_PORT))
assert response.status_code == 200
print("Wait 15 seconds for Kapacitor to trigger alerts...")
sleep(15)
alert_logs = listdir(LOG_TEST_FOLDER_PATH)
assert len(alert_logs) == 4, "4 log files must have been created - one for each alert defined in the specification."
for alert_log in alert_logs:
alert_log_path = join(LOG_TEST_FOLDER_PATH, alert_log)
with open(alert_log_path) as fh:
alert_json = load(fh)
try:
JSON_BODY_SCHEMA.validate(alert_json)
valid = True
except SchemaError:
valid = False
assert valid, "Alert log content is invalid - {0}".format(alert_log_path)