diff options
Diffstat (limited to 'ceilometer/tests/unit/hardware/pollsters/test_generic.py')
| -rw-r--r-- | ceilometer/tests/unit/hardware/pollsters/test_generic.py | 193 |
1 files changed, 0 insertions, 193 deletions
diff --git a/ceilometer/tests/unit/hardware/pollsters/test_generic.py b/ceilometer/tests/unit/hardware/pollsters/test_generic.py deleted file mode 100644 index 99ac2bfb..00000000 --- a/ceilometer/tests/unit/hardware/pollsters/test_generic.py +++ /dev/null @@ -1,193 +0,0 @@ -# -# 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']) |