Source code for heat_integrationtests.scenario.test_aodh_alarm

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from oslo_log import log as logging

from heat_integrationtests.common import test
from heat_integrationtests.scenario import scenario_base

LOG = logging.getLogger(__name__)


[docs]class AodhAlarmTest(scenario_base.ScenarioTestsBase):
    """Class is responsible for testing of aodh usage."""
[docs]    def setUp(self):
        super(AodhAlarmTest, self).setUp()
        self.template = self._load_template(__file__,
                                            'test_aodh_alarm.yaml',
                                            'templates')

[docs]    def check_instance_count(self, stack_identifier, expected):
        stack = self.client.stacks.get(stack_identifier)
        actual = self._stack_output(stack, 'asg_size')
        if actual != expected:
            LOG.warning('check_instance_count exp:%d, act:%s' % (expected,
                                                                 actual))
        return actual == expected

[docs]    def test_alarm(self):
        """Confirm we can create an alarm and trigger it."""

        # 1. create the stack
        stack_identifier = self.stack_create(template=self.template)

        # 2. send ceilometer a metric (should cause the alarm to fire)
        sample = {}
        sample['counter_type'] = 'gauge'
        sample['counter_name'] = 'test_meter'
        sample['counter_volume'] = 1
        sample['counter_unit'] = 'count'
        sample['resource_metadata'] = {'metering.stack_id':
                                       stack_identifier.split('/')[-1]}
        sample['resource_id'] = 'shouldnt_matter'
        self.metering_client.samples.create(**sample)

        # 3. confirm we get a scaleup.
        # Note: there is little point waiting more than 60s+time to scale up.
        self.assertTrue(test.call_until_true(
            120, 2, self.check_instance_count, stack_identifier, 2))