"Fossies" - the Fresh Open Source Software Archive

Member "monasca-api-3.1.0/monasca_api/tests/test_validation.py" (27 Sep 2019, 16395 Bytes) of package /linux/misc/openstack/monasca-api-3.1.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_validation.py": 3.1.0_vs_4.0.0.

    1 # (C) Copyright 2015-2017 Hewlett Packard Enterprise Development LP
    2 # Copyright 2015 Cray Inc. All Rights Reserved.
    3 # Copyright 2017 Fujitsu LIMITED
    4 #
    5 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    6 # not use this file except in compliance with the License. You may obtain
    7 # a copy of the License at
    8 #
    9 #      http://www.apache.org/licenses/LICENSE-2.0
   10 #
   11 # Unless required by applicable law or agreed to in writing, software
   12 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   13 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   14 # License for the specific language governing permissions and limitations
   15 # under the License.
   16 
   17 import falcon
   18 import mock
   19 
   20 from monasca_api.tests import base
   21 import monasca_api.v2.common.exceptions as common_exceptions
   22 import monasca_api.v2.common.schemas.alarm_definition_request_body_schema as schemas_alarm_defs
   23 import monasca_api.v2.common.schemas.exceptions as schemas_exceptions
   24 import monasca_api.v2.common.schemas.notifications_request_body_schema as schemas_notifications
   25 import monasca_api.v2.common.validation as validation
   26 import monasca_api.v2.reference.helpers as helpers
   27 
   28 
   29 def mock_req_can(authorised_rule):
   30     if authorised_rule != 'authorized':
   31         raise Exception
   32 
   33 
   34 class TestStateValidation(base.BaseTestCase):
   35 
   36     VALID_STATES = "OK", "ALARM", "UNDETERMINED"
   37 
   38     def test_valid_states(self):
   39         for state in self.VALID_STATES:
   40             validation.validate_alarm_state(state)
   41 
   42     def test_valid_states_lower_case(self):
   43         for state in self.VALID_STATES:
   44             validation.validate_alarm_state(state.lower())
   45 
   46     def test_invalid_state(self):
   47         self.assertRaises(common_exceptions.HTTPUnprocessableEntityError,
   48                           validation.validate_alarm_state, 'BOGUS')
   49 
   50 
   51 class TestSeverityValidation(base.BaseTestCase):
   52 
   53     VALID_SEVERITIES = "LOW", "MEDIUM", "HIGH", "CRITICAL"
   54 
   55     def test_valid_severities(self):
   56         for state in self.VALID_SEVERITIES:
   57             validation.validate_severity_query(state)
   58 
   59     def test_valid_severities_lower_case(self):
   60         for state in self.VALID_SEVERITIES:
   61             validation.validate_severity_query(state.lower())
   62 
   63     def test_valid_multi_severities(self):
   64         validation.validate_severity_query('|'.join(self.VALID_SEVERITIES))
   65 
   66     def test_valid_multi_severities_lower_case(self):
   67         validation.validate_severity_query('|'.join(self.VALID_SEVERITIES)
   68                                            .lower())
   69 
   70     def test_invalid_state(self):
   71         self.assertRaises(common_exceptions.HTTPUnprocessableEntityError,
   72                           validation.validate_severity_query,
   73                           'BOGUS')
   74         self.assertRaises(common_exceptions.HTTPUnprocessableEntityError,
   75                           validation.validate_severity_query,
   76                           '|'.join([self.VALID_SEVERITIES[0], 'BOGUS']))
   77 
   78 
   79 class TestRuleValidation(base.BaseApiTestCase):
   80     def test_rule_valid(self):
   81         req = mock.Mock()
   82         req.can = mock_req_can
   83         test_rules = ['Rule1', 'authorized']
   84         helpers.validate_authorization(req, test_rules)
   85 
   86     def test_rule_invalid(self):
   87         req = mock.Mock()
   88         req.can = mock_req_can
   89         test_rules = ['rule1', 'rule2']
   90         self.assertRaises(
   91             falcon.HTTPUnauthorized,
   92             helpers.validate_authorization, req, test_rules)
   93 
   94 
   95 class TestTimestampsValidation(base.BaseTestCase):
   96 
   97     def test_valid_timestamps(self):
   98         start_time = '2015-01-01T00:00:00Z'
   99         end_time = '2015-01-01T00:00:01Z'
  100         start_timestamp = helpers._convert_time_string(start_time)
  101         end_timestamp = helpers._convert_time_string(end_time)
  102 
  103         try:
  104             helpers.validate_start_end_timestamps(start_timestamp,
  105                                                   end_timestamp)
  106         except:
  107             self.fail("shouldn't happen")
  108 
  109     def test_same_timestamps(self):
  110         start_time = '2015-01-01T00:00:00Z'
  111         end_time = start_time
  112         start_timestamp = helpers._convert_time_string(start_time)
  113         end_timestamp = helpers._convert_time_string(end_time)
  114 
  115         self.assertRaises(
  116             falcon.HTTPBadRequest,
  117             helpers.validate_start_end_timestamps,
  118             start_timestamp, end_timestamp)
  119 
  120     def test_end_before_than_start(self):
  121         start_time = '2015-01-01T00:00:00Z'
  122         end_time = '2014-12-31T23:59:59Z'
  123         start_timestamp = helpers._convert_time_string(start_time)
  124         end_timestamp = helpers._convert_time_string(end_time)
  125 
  126         self.assertRaises(
  127             falcon.HTTPBadRequest,
  128             helpers.validate_start_end_timestamps,
  129             start_timestamp, end_timestamp)
  130 
  131 
  132 class TestConvertTimeString(base.BaseTestCase):
  133 
  134     def test_valid_date_time_string(self):
  135         date_time_string = '2015-01-01T00:00:00Z'
  136 
  137         timestamp = helpers._convert_time_string(date_time_string)
  138         self.assertEqual(1420070400., timestamp)
  139 
  140     def test_valid_date_time_string_with_mills(self):
  141         date_time_string = '2015-01-01T00:00:00.025Z'
  142 
  143         timestamp = helpers._convert_time_string(date_time_string)
  144         self.assertEqual(1420070400.025, timestamp)
  145 
  146     def test_valid_date_time_string_with_timezone(self):
  147         date_time_string = '2015-01-01T09:00:00+09:00'
  148 
  149         timestamp = helpers._convert_time_string(date_time_string)
  150         self.assertEqual(1420070400., timestamp)
  151 
  152     def test_invalid_date_time_string(self):
  153         date_time_string = '2015-01-01T00:00:000Z'
  154 
  155         self.assertRaises(
  156             ValueError,
  157             helpers._convert_time_string, date_time_string)
  158 
  159 valid_periods = [0, 60]
  160 
  161 
  162 class TestNotificationValidation(base.BaseTestCase):
  163 
  164     def test_validation_for_email(self):
  165         notification = {"name": "MyEmail", "type": "EMAIL", "address": "name@domain.com"}
  166         try:
  167             schemas_notifications.parse_and_validate(notification, valid_periods)
  168         except schemas_exceptions.ValidationException:
  169             self.fail("shouldn't happen")
  170 
  171     def test_validation_exception_for_invalid_email_address(self):
  172         notification = {"name": "MyEmail", "type": "EMAIL", "address": "name@"}
  173         ex = self.assertRaises(schemas_exceptions.ValidationException,
  174                                schemas_notifications.parse_and_validate,
  175                                notification, valid_periods)
  176         self.assertEqual("Address name@ is not of correct format", str(ex))
  177 
  178     def test_validation_exception_for_invalid_period_for_email(self):
  179         notification = {
  180             "name": "MyEmail",
  181             "type": "EMAIL",
  182             "address": "name@domain.com",
  183             "period": "60"}
  184         ex = self.assertRaises(schemas_exceptions.ValidationException,
  185                                schemas_notifications.parse_and_validate,
  186                                notification, valid_periods)
  187         self.assertEqual("Period can only be set with webhooks", str(ex))
  188 
  189     def test_validation_for_webhook(self):
  190         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "http://somedomain.com"}
  191         try:
  192             schemas_notifications.parse_and_validate(notification, valid_periods)
  193         except schemas_exceptions.ValidationException:
  194             self.fail("shouldn't happen")
  195 
  196     def test_validation_for_webhook_non_zero_period(self):
  197         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "http://somedomain.com",
  198                         "period": 60}
  199         try:
  200             schemas_notifications.parse_and_validate(notification, valid_periods)
  201         except schemas_exceptions.ValidationException:
  202             self.fail("shouldn't happen")
  203 
  204     def test_validation_exception_for_webhook_no_scheme(self):
  205         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "//somedomain.com"}
  206         ex = self.assertRaises(schemas_exceptions.ValidationException,
  207                                schemas_notifications.parse_and_validate,
  208                                notification, valid_periods)
  209         self.assertEqual("Address //somedomain.com does not have URL scheme", str(ex))
  210 
  211     def test_validation_exception_for_webhook_no_netloc(self):
  212         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "http://"}
  213         ex = self.assertRaises(schemas_exceptions.ValidationException,
  214                                schemas_notifications.parse_and_validate,
  215                                notification, valid_periods)
  216         self.assertEqual("Address http:// does not have network location", str(ex))
  217 
  218     def test_validation_exception_for_webhook_invalid_scheme(self):
  219         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "ftp://somedomain.com"}
  220         ex = self.assertRaises(schemas_exceptions.ValidationException,
  221                                schemas_notifications.parse_and_validate,
  222                                notification, valid_periods)
  223         self.assertEqual("Address ftp://somedomain.com scheme is not in ['http', 'https']", str(ex))
  224 
  225     def test_validation_exception_for_webhook_invalid_period(self):
  226         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "//somedomain.com",
  227                         "period": "10"}
  228         ex = self.assertRaises(schemas_exceptions.ValidationException,
  229                                schemas_notifications.parse_and_validate,
  230                                notification, valid_periods)
  231         self.assertEqual("10 is not a valid period, not in [0, 60]", str(ex))
  232 
  233     def test_validation_for_pagerduty(self):
  234         notification = {"name": "MyPagerduty", "type": "PAGERDUTY",
  235                         "address": "nzH2LVRdMzun11HNC2oD"}
  236         try:
  237             schemas_notifications.parse_and_validate(notification, valid_periods)
  238         except schemas_exceptions.ValidationException:
  239             self.fail("shouldn't happen")
  240 
  241     def test_validation_exception_for_invalid_period_for_pagerduty(self):
  242         notification = {"name": "MyPagerduty", "type": "PAGERDUTY",
  243                         "address": "nzH2LVRdMzun11HNC2oD", "period": 60}
  244         ex = self.assertRaises(schemas_exceptions.ValidationException,
  245                                schemas_notifications.parse_and_validate,
  246                                notification, valid_periods)
  247         self.assertEqual("Period can only be set with webhooks", str(ex))
  248 
  249     def test_validation_for_max_name_address(self):
  250         name = "A" * 250
  251         self.assertEqual(250, len(name))
  252         address = "http://" + "A" * 502 + ".io"
  253         self.assertEqual(512, len(address))
  254         notification = {"name": name, "type": "WEBHOOK", "address": address}
  255         try:
  256             schemas_notifications.parse_and_validate(notification, valid_periods)
  257         except schemas_exceptions.ValidationException:
  258             self.fail("shouldn't happen")
  259 
  260     def test_validation_exception_for_exceeded_name_length(self):
  261         name = "A" * 251
  262         self.assertEqual(251, len(name))
  263         notification = {"name": name, "type": "WEBHOOK", "address": "http://somedomain.com"}
  264         self.assertRaises(
  265             schemas_exceptions.ValidationException,
  266             schemas_notifications.parse_and_validate,
  267             notification, valid_periods)
  268 
  269     def test_validation_exception_for_exceeded_address_length(self):
  270         address = "http://" + "A" * 503 + ".io"
  271         self.assertEqual(513, len(address))
  272         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": address}
  273         self.assertRaises(
  274             schemas_exceptions.ValidationException,
  275             schemas_notifications.parse_and_validate, notification, valid_periods)
  276 
  277     def test_validation_exception_for_invalid_period_float(self):
  278         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "//somedomain.com",
  279                         "period": 1.2}
  280         ex = self.assertRaises(schemas_exceptions.ValidationException,
  281                                schemas_notifications.parse_and_validate,
  282                                notification, valid_periods)
  283         self.assertEqual("expected int for dictionary value @ data['period']", str(ex))
  284 
  285     def test_validation_exception_for_invalid_period_non_int(self):
  286         notification = {"name": "MyWebhook", "type": "WEBHOOK", "address": "//somedomain.com",
  287                         "period": "zero"}
  288         ex = self.assertRaises(schemas_exceptions.ValidationException,
  289                                schemas_notifications.parse_and_validate,
  290                                notification, valid_periods)
  291         self.assertEqual("Period zero must be a valid integer", str(ex))
  292 
  293     def test_validation_exception_for_missing_period(self):
  294         notification = {"name": "MyEmail", "type": "EMAIL", "address": "name@domain."}
  295         ex = self.assertRaises(schemas_exceptions.ValidationException,
  296                                schemas_notifications.parse_and_validate,
  297                                notification, valid_periods, require_all=True)
  298         self.assertEqual("Period is required", str(ex))
  299 
  300 
  301 class TestAlarmDefinitionValidation(base.BaseTestCase):
  302 
  303     def setUp(self):
  304         super(TestAlarmDefinitionValidation, self).setUp()
  305         self.full_alarm_definition = (
  306             {"name": self._create_string_of_length(255),
  307              "expression": "min(cpu.idle_perc) < 10",
  308              "description": self._create_string_of_length(255),
  309              "severity": "MEDIUM",
  310              "match_by": ["hostname"],
  311              "ok_actions:": [self._create_string_of_length(50)],
  312              "undetermined_actions": [self._create_string_of_length(50)],
  313              "alarm_actions": [self._create_string_of_length(50)],
  314              "actions_enabled": True})
  315 
  316     def _create_string_of_length(self, length):
  317         s = ''
  318         for i in range(0, length):
  319             s += str(i % 10)
  320         return s
  321 
  322     def test_validation_good_minimum(self):
  323         alarm_definition = {"name": "MyAlarmDefinition", "expression": "min(cpu.idle_perc) < 10"}
  324         try:
  325             schemas_alarm_defs.validate(alarm_definition)
  326         except schemas_exceptions.ValidationException as e:
  327             self.fail("shouldn't happen: {}".format(str(e)))
  328 
  329     def test_validation_good_full(self):
  330         alarm_definition = self.full_alarm_definition
  331         try:
  332             schemas_alarm_defs.validate(alarm_definition)
  333         except schemas_exceptions.ValidationException as e:
  334             self.fail("shouldn't happen: {}".format(str(e)))
  335 
  336     def _ensure_fails_with_new_value(self, name, value):
  337         alarm_definition = self.full_alarm_definition.copy()
  338         alarm_definition[name] = value
  339         self._ensure_validation_fails(alarm_definition)
  340 
  341     def _ensure_validation_fails(self, alarm_definition):
  342         self.assertRaises(
  343             schemas_exceptions.ValidationException,
  344             schemas_alarm_defs.validate, alarm_definition)
  345 
  346     def _run_duplicate_action_test(self, actions_type):
  347         actions = ["a", "b", "a"]
  348         self._ensure_fails_with_new_value(actions_type, actions)
  349 
  350     def test_validation_too_long_name(self):
  351         self._ensure_fails_with_new_value("name",
  352                                           self._create_string_of_length(256))
  353 
  354     def test_validation_too_long_description(self):
  355         self._ensure_fails_with_new_value("description",
  356                                           self._create_string_of_length(256))
  357 
  358     def test_validation_duplicate_ok_actions(self):
  359         self._run_duplicate_action_test("ok_actions")
  360 
  361     def test_validation_duplicate_alarm_actions(self):
  362         self._run_duplicate_action_test("alarm_actions")
  363 
  364     def test_validation_duplicate_undetermined_actions(self):
  365         self._run_duplicate_action_test("undetermined_actions")
  366 
  367     def test_validation_too_many_actions(self):
  368         actions = [self._create_string_of_length(51)]
  369         self._ensure_fails_with_new_value("ok_actions", actions)
  370 
  371     def test_validation_invalid_severity(self):
  372         self._ensure_fails_with_new_value("severity", "BOGUS")
  373 
  374     def test_validation_invalid_match_by(self):
  375         self._ensure_fails_with_new_value("match_by", "NOT_A_LIST")
  376 
  377     def test_validation_invalid_actions_enabled(self):
  378         self._ensure_fails_with_new_value("actions_enabled", 42)