aodh/tests/api/v2/test_compute_duration_by_resource.py

# -*- encoding: utf-8 -*-
#
# Copyright © 2012 New Dream Network, LLC (DreamHost)
#
# Author: Doug Hellmann <doug.hellmann@dreamhost.com>
#
# 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.
"""Test listing raw events.
"""

import datetime
import logging

from ceilometer.openstack.common import timeutils
from ceilometer.storage import impl_test
from .base import FunctionalTest

LOG = logging.getLogger(__name__)


class TestComputeDurationByResource(FunctionalTest):

    def setUp(self):
        super(TestComputeDurationByResource, self).setUp()

        # Create events relative to the range and pretend
        # that the intervening events exist.

        self.early1 = datetime.datetime(2012, 8, 27, 7, 0)
        self.early2 = datetime.datetime(2012, 8, 27, 17, 0)

        self.start = datetime.datetime(2012, 8, 28, 0, 0)

        self.middle1 = datetime.datetime(2012, 8, 28, 8, 0)
        self.middle2 = datetime.datetime(2012, 8, 28, 18, 0)

        self.end = datetime.datetime(2012, 8, 28, 23, 59)

        self.late1 = datetime.datetime(2012, 8, 29, 9, 0)
        self.late2 = datetime.datetime(2012, 8, 29, 19, 0)

    def _stub_interval_func(self, func):
        self.stubs.Set(impl_test.TestConnection,
                       'get_meter_statistics',
                       func)

    def _set_interval(self, start, end):
        def get_interval(ignore_self, event_filter, period):
            assert event_filter.start
            assert event_filter.end
            if (event_filter.start > end
                or event_filter.end < start):
                return []
            return [{'count': 0,
                     # ...
                     'duration_start': max(event_filter.start, start),
                     'duration_end': min(event_filter.end, end)}]
        self._stub_interval_func(get_interval)

    def _invoke_api(self):
        return self.get_json('/meters/instance:m1.tiny/statistics',
                             q=[{'field': 'timestamp',
                                 'op': 'ge',
                                 'value': self.start.isoformat()},
                                {'field': 'timestamp',
                                 'op': 'le',
                                 'value': self.end.isoformat()},
                                {'field': 'search_offset',
                                 'value': 10}])

    def test_before_range(self):
        self._set_interval(self.early1, self.early2)
        data = self._invoke_api()
        self.assertEqual(data, [])

    def _assert_times_match(self, actual, expected):
        if actual:
            actual = timeutils.parse_isotime(actual)
        actual = actual.replace(tzinfo=None)
        assert actual == expected

    def test_overlap_range_start(self):
        self._set_interval(self.early1, self.middle1)
        data = self._invoke_api()
        self._assert_times_match(data[0]['duration_start'], self.start)
        self._assert_times_match(data[0]['duration_end'], self.middle1)
        self.assertEqual(data[0]['duration'], 8 * 60 * 60)

    def test_within_range(self):
        self._set_interval(self.middle1, self.middle2)
        data = self._invoke_api()
        self._assert_times_match(data[0]['duration_start'], self.middle1)
        self._assert_times_match(data[0]['duration_end'], self.middle2)
        self.assertEqual(data[0]['duration'], 10 * 60 * 60)

    def test_within_range_zero_duration(self):
        self._set_interval(self.middle1, self.middle1)
        data = self._invoke_api()
        self._assert_times_match(data[0]['duration_start'], self.middle1)
        self._assert_times_match(data[0]['duration_end'], self.middle1)
        self.assertEqual(data[0]['duration'], 0)

    def test_overlap_range_end(self):
        self._set_interval(self.middle2, self.late1)
        data = self._invoke_api()
        self._assert_times_match(data[0]['duration_start'], self.middle2)
        self._assert_times_match(data[0]['duration_end'], self.end)
        self.assertEqual(data[0]['duration'], ((6 * 60) - 1) * 60)

    def test_after_range(self):
        self._set_interval(self.late1, self.late2)
        data = self._invoke_api()
        self.assertEqual(data, [])

    def test_without_end_timestamp(self):
        def get_interval(ignore_self, event_filter, period):
            return [{'count': 0,
                     'min': None,
                     'max': None,
                     'avg': None,
                     'qty': None,
                     'duration': None,
                     'duration_start': self.late1,
                     'duration_end': self.late2}]
        self._stub_interval_func(get_interval)
        data = self.get_json('/meters/instance:m1.tiny/statistics',
                             q=[{'field': 'timestamp',
                                 'op': 'ge',
                                 'value': self.late1.isoformat()},
                                {'field': 'resource_id',
                                 'value': 'resource-id'},
                                {'field': 'search_offset',
                                 'value': 10}])
        self._assert_times_match(data[0]['duration_start'], self.late1)
        self._assert_times_match(data[0]['duration_end'], self.late2)

    def test_without_start_timestamp(self):
        def get_interval(ignore_self, event_filter, period):
            return [{'count': 0,
                     'min': None,
                     'max': None,
                     'avg': None,
                     'qty': None,
                     'duration': None,
                     'duration_start': self.early1,
                     'duration_end': self.early2}]
            return (self.early1, self.early2)
        self._stub_interval_func(get_interval)
        data = self.get_json('/meters/instance:m1.tiny/statistics',
                             q=[{'field': 'timestamp',
                                 'op': 'le',
                                 'value': self.early2.isoformat()},
                                {'field': 'resource_id',
                                 'value': 'resource-id'},
                                {'field': 'search_offset',
                                 'value': 10}])
        self._assert_times_match(data[0]['duration_start'], self.early1)
        self._assert_times_match(data[0]['duration_end'], self.early2)