Source code for ceilometer.tests.functional.api.v2.test_compute_duration_by_resource_scenarios

#
# Copyright 2012 New Dream Network, LLC (DreamHost)
#
# 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 mock
from oslo_utils import timeutils

from ceilometer.storage import models
from ceilometer.tests.functional.api import v2


[docs]class TestComputeDurationByResource(v2.FunctionalTest):
[docs] 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 _patch_get_interval(self, start, end): def get_interval(sample_filter, period, groupby, aggregate): self.assertIsNotNone(sample_filter.start_timestamp) self.assertIsNotNone(sample_filter.end_timestamp) if (sample_filter.start_timestamp > end or sample_filter.end_timestamp < start): return [] duration_start = max(sample_filter.start_timestamp, start) duration_end = min(sample_filter.end_timestamp, end) duration = timeutils.delta_seconds(duration_start, duration_end) return [ models.Statistics( unit='', min=0, max=0, avg=0, sum=0, count=0, period=None, period_start=None, period_end=None, duration=duration, duration_start=duration_start, duration_end=duration_end, groupby=None, ) ] return mock.patch.object(type(self.conn), 'get_meter_statistics', side_effect=get_interval) def _invoke_api(self): return self.get_json('/meters/instance/statistics', q=[{'field': 'timestamp', 'op': 'ge', 'value': self.start.isoformat()}, {'field': 'timestamp', 'op': 'le', 'value': self.end.isoformat()}, {'field': 'search_offset', 'value': 10}])
[docs] def test_before_range(self): with self._patch_get_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) self.assertEqual(expected, actual)
[docs] def test_overlap_range_start(self): with self._patch_get_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(8 * 60 * 60, data[0]['duration'])
[docs] def test_within_range(self): with self._patch_get_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(10 * 60 * 60, data[0]['duration'])
[docs] def test_within_range_zero_duration(self): with self._patch_get_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(0, data[0]['duration'])
[docs] def test_overlap_range_end(self): with self._patch_get_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(((6 * 60) - 1) * 60, data[0]['duration'])
[docs] def test_after_range(self): with self._patch_get_interval(self.late1, self.late2): data = self._invoke_api() self.assertEqual([], data)
[docs] def test_without_end_timestamp(self): statistics = [ models.Statistics( unit=None, count=0, min=None, max=None, avg=None, duration=None, duration_start=self.late1, duration_end=self.late2, sum=0, period=None, period_start=None, period_end=None, groupby=None, ) ] with mock.patch.object(type(self.conn), 'get_meter_statistics', return_value=statistics): data = self.get_json('/meters/instance/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)
[docs] def test_without_start_timestamp(self): statistics = [ models.Statistics( unit=None, count=0, min=None, max=None, avg=None, duration=None, duration_start=self.early1, duration_end=self.early2, sum=0, period=None, period_start=None, period_end=None, groupby=None, ) ] with mock.patch.object(type(self.conn), 'get_meter_statistics', return_value=statistics): data = self.get_json('/meters/instance/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)

Project Source