3cdac9f3d0
The meter signing and publishing code was tightly coupled to the global configuration object, making it more difficult to reuse in an outside project. This change adds arguments for the configuration settings actually being used in those libraries and moves use of the global configuration object higher up the stack to the manager. It also organizes the definition of the options related to signing and publishing metering messages so they can more easily by registered on other configuration objects. Change-Id: I8b1df0f228d0c2658841c4b7f7f4527414efc9a6 Signed-off-by: Doug Hellmann <doug.hellmann@dreamhost.com>
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
# -*- 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.
|
|
"""Tests for ceilometer/publish.py
|
|
"""
|
|
|
|
import datetime
|
|
|
|
from ceilometer.openstack.common import rpc
|
|
from ceilometer.tests import base
|
|
|
|
from ceilometer import counter
|
|
from ceilometer import publish
|
|
|
|
|
|
class TestPublish(base.TestCase):
|
|
|
|
test_data = counter.Counter(
|
|
source='test',
|
|
name='test',
|
|
type='cumulative',
|
|
volume=1,
|
|
user_id='test',
|
|
project_id='test',
|
|
resource_id='test_run_tasks',
|
|
timestamp=datetime.datetime.utcnow().isoformat(),
|
|
duration=0,
|
|
resource_metadata={'name': 'TestPublish',
|
|
},
|
|
)
|
|
|
|
def faux_notify(self, context, topic, msg):
|
|
self.notifications.append((topic, msg))
|
|
|
|
def setUp(self):
|
|
super(TestPublish, self).setUp()
|
|
self.notifications = []
|
|
self.stubs.Set(rpc, 'cast', self.faux_notify)
|
|
publish.publish_counter(None,
|
|
self.test_data,
|
|
'metering',
|
|
'not-so-secret')
|
|
|
|
def test_notify(self):
|
|
assert len(self.notifications) == 2
|
|
|
|
def test_notify_topics(self):
|
|
topics = [n[0] for n in self.notifications]
|
|
assert topics == ['metering', 'metering.test']
|