Source code for ceilometer.tests.functional.publisher.test_direct

#
# Copyright 2015 Red Hat
#
# 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.
"""Tests for ceilometer/publisher/direct.py
"""

import datetime
import uuid

from oslo_utils import netutils

from ceilometer.event.storage import models as event
from ceilometer.publisher import direct
from ceilometer import sample
from ceilometer.tests import db as tests_db


[docs]class TestDirectPublisher(tests_db.TestBase): resource_id = str(uuid.uuid4()) test_data = [ sample.Sample( name='alpha', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id=resource_id, timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='beta', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id=resource_id, timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='gamma', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id=resource_id, timestamp=datetime.datetime.now().isoformat(), resource_metadata={'name': 'TestPublish'}, ), ]
[docs] def test_direct_publisher(self): """Test samples are saved.""" self.CONF.set_override('connection', self.db_manager.url, group='database') parsed_url = netutils.urlsplit('direct://') publisher = direct.DirectPublisher(parsed_url) publisher.publish_samples(self.test_data) meters = list(self.conn.get_meters(resource=self.resource_id)) names = sorted([meter.name for meter in meters]) self.assertEqual(3, len(meters), 'There should be 3 samples') self.assertEqual(['alpha', 'beta', 'gamma'], names)
[docs]class TestEventDirectPublisher(tests_db.TestBase): test_data = [event.Event(message_id=str(uuid.uuid4()), event_type='event_%d' % i, generated=datetime.datetime.utcnow(), traits=[], raw={}) for i in range(0, 5)]
[docs] def test_direct_publisher(self): parsed_url = netutils.urlsplit('direct://dispatcher=database') publisher = direct.DirectPublisher(parsed_url) publisher.publish_events(self.test_data) e_types = list(self.event_conn.get_event_types()) self.assertEqual(5, len(e_types)) self.assertEqual(['event_%d' % i for i in range(0, 5)], sorted(e_types))

Project Source