# # Copyright 2015 Intel Corp. # # 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 unittest import mock import fixtures from oslo_utils import fileutils import yaml from ceilometer import declarative from ceilometer.hardware.inspector import base as inspector_base from ceilometer.hardware.pollsters import generic from ceilometer import sample from ceilometer import service from ceilometer.tests import base as test_base class TestMeterDefinition(test_base.BaseTestCase): def test_config_definition(self): cfg = dict(name='test', type='gauge', unit='B', snmp_inspector={}) definition = generic.MeterDefinition(cfg) self.assertEqual('test', definition.name) self.assertEqual('gauge', definition.type) self.assertEqual('B', definition.unit) self.assertEqual({}, definition.snmp_inspector) def test_config_missing_field(self): cfg = dict(name='test', type='gauge') try: generic.MeterDefinition(cfg) except declarative.MeterDefinitionException as e: self.assertEqual("Missing field unit", e.brief_message) def test_config_invalid_field(self): cfg = dict(name='test', type='gauge', unit='B', invalid={}) definition = generic.MeterDefinition(cfg) self.assertEqual("foobar", getattr(definition, 'invalid', 'foobar')) def test_config_invalid_type_field(self): cfg = dict(name='test', type='invalid', unit='B', snmp_inspector={}) try: generic.MeterDefinition(cfg) except declarative.MeterDefinitionException as e: self.assertEqual("Unrecognized type value invalid", e.brief_message) def test_config_missing_unit_field(self): cfg = dict(name='hardware.cpu.user', snmp_inspector={"matching_type": "type_exact", "oid": "1.3.6.1.4.1.2021.11.50.0", "type": "int"}) try: generic.MeterDefinition(cfg) except declarative.MeterDefinitionException as e: self.assertEqual("Missing field unit", e.brief_message) @mock.patch('ceilometer.hardware.pollsters.generic.LOG') def test_bad_metric_skip(self, LOG): cfg = {'metric': [dict(name='test1', type='gauge', unit='B', snmp_inspector={}), dict(name='test_bad', type='invalid', unit='B', snmp_inspector={}), dict(name='test2', type='gauge', unit='B', snmp_inspector={})]} data = generic.load_definition(cfg) self.assertEqual(2, len(data)) LOG.error.assert_called_with( "Error loading meter definition: %s", "Unrecognized type value invalid") class FakeInspector(inspector_base.Inspector): net_metadata = dict(name='test.teest', mac='001122334455', ip='10.0.0.2', speed=1000) DATA = { 'test': (0.99, {}, {}), 'test2': (90, net_metadata, {}), } def inspect_generic(self, host, cache, extra_metadata=None, param=None): yield self.DATA[host.hostname] class TestGenericPollsters(test_base.BaseTestCase): @staticmethod def faux_get_inspector(url, namespace=None): return FakeInspector() def setUp(self): super(TestGenericPollsters, self).setUp() self.conf = service.prepare_service([], []) self.resources = ["snmp://test", "snmp://test2"] self.useFixture(fixtures.MockPatch( 'ceilometer.hardware.inspector.get_inspector', self.faux_get_inspector)) self.pollster = generic.GenericHardwareDeclarativePollster(self.conf) def _setup_meter_def_file(self, cfg): cfg = cfg.encode('utf-8') meter_cfg_file = fileutils.write_to_tempfile(content=cfg, prefix="snmp", suffix="yaml") self.conf.set_override( 'meter_definitions_file', meter_cfg_file, group='hardware') cfg = declarative.load_definitions( self.conf, {}, self.conf.hardware.meter_definitions_file) return cfg def _check_get_samples(self, name, definition, expected_value, expected_type, expected_unit=None): self.pollster._update_meter_definition(definition) cache = {} samples = list(self.pollster.get_samples(None, cache, self.resources)) self.assertTrue(samples) self.assertIn(self.pollster.CACHE_KEY, cache) for resource in self.resources: self.assertIn(resource, cache[self.pollster.CACHE_KEY]) self.assertEqual(set([name]), set([s.name for s in samples])) match = [s for s in samples if s.name == name] self.assertEqual(expected_value, match[0].volume) self.assertEqual(expected_type, match[0].type) if expected_unit: self.assertEqual(expected_unit, match[0].unit) def test_get_samples(self): param = dict(matching_type='type_exact', oid='1.3.6.1.4.1.2021.10.1.3.1', type='lambda x: float(str(x))') meter_def = generic.MeterDefinition(dict(type='gauge', name='hardware.test1', unit='process', snmp_inspector=param)) self._check_get_samples('hardware.test1', meter_def, 0.99, sample.TYPE_GAUGE, expected_unit='process') def test_get_pollsters_extensions(self): param = dict(matching_type='type_exact', oid='1.3.6.1.4.1.2021.10.1.3.1', type='lambda x: float(str(x))') meter_cfg = yaml.dump( {'metric': [dict(type='gauge', name='hardware.test1', unit='process', snmp_inspector=param), dict(type='gauge', name='hardware.test2.abc', unit='process', snmp_inspector=param)]}) self._setup_meter_def_file(meter_cfg) pollster = generic.GenericHardwareDeclarativePollster # Clear cached mapping pollster.mapping = None exts = pollster.get_pollsters_extensions(self.conf) self.assertEqual(2, len(exts)) self.assertIn(exts[0].name, ['hardware.test1', 'hardware.test2.abc']) self.assertIn(exts[1].name, ['hardware.test1', 'hardware.test2.abc'])