openstack/vitrage
Code Issues Proposed changes

nagios transformer - adding tests

Browse Source 
Change-Id: I3e3bfeb51e538c8af60f8954e9537f280d4cf605
This commit is contained in:
Liat Har-Tal 2016-02-16 06:52:04 +00:00
parent aafc1c17f6
commit f2f9879d0a
3 changed files with 109 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
vitrage/synchronizer/plugins/nagios/transformer.py
View File

@ -104,8 +104,8 @@ class NagiosTransformer(tbase.TransformerBase):
def _extract_action_type(self, entity_event):
sync_mode = entity_event[SyncProps.SYNC_MODE]
if sync_mode in (SyncMode.UPDATE, SyncMode.SNAPSHOT):
if entity_event[self.STATUS] == 'OK':
return EventAction.CREATE
if entity_event[NagiosProperties.STATUS] == self.STATUS_OK:
return EventAction.DELETE
else:
return EventAction.UPDATE
if SyncMode.INIT_SNAPSHOT == sync_mode:
@ -114,12 +114,12 @@ class NagiosTransformer(tbase.TransformerBase):
def extract_key(self, entity_event):
type = entity_event[SyncProps.SYNC_TYPE]
sync_type = entity_event[SyncProps.SYNC_TYPE]
alarm_name = entity_event[NagiosProperties.SERVICE]
resource_name = entity_event[NagiosProperties.RESOURCE_NAME]
return tbase.build_key(self.key_values([type,
resource_name,
alarm_name]))
return tbase.build_key(self.key_values([sync_type,
resource_name,
alarm_name]))
def key_values(self, mutable_fields=[]):
return [EntityCategory.ALARM] + mutable_fields

101
vitrage/tests/unit/synchronizer/nagios/test_nagios_alarm_transformer.py
View File

@ -13,9 +13,17 @@
# under the License.
from oslo_log import log as logging
from vitrage.common.constants import EdgeLabels
from vitrage.common.constants import EntityCategory
from vitrage.common.constants import EntityType
from vitrage.common.constants import EventAction
from vitrage.common.constants import SynchronizerProperties as SyncProps
from vitrage.common.constants import SyncMode
from vitrage.common.constants import VertexProperties as VProps
from vitrage.synchronizer.plugins.nagios.properties import NagiosProperties
from vitrage.synchronizer.plugins.nagios.transformer import NagiosTransformer
from vitrage.synchronizer.plugins.nova.host.transformer import HostTransformer
from vitrage.synchronizer.plugins.transformer_base import TransformerBase
from vitrage.tests import base
from vitrage.tests.mocks import mock_syncronizer as mock_sync
@ -32,14 +40,101 @@ class NagiosTransformerTest(base.BaseTest):
host_transformer = HostTransformer(self.transformers)
self.transformers[EntityType.NOVA_HOST] = host_transformer
def test_extract_key(self):
LOG.debug('Test get key from nova instance transformer')
# Test setup
spec_list = mock_sync.simple_nagios_alarm_generators(host_num=1,
events_num=1)
nagios_alarms = mock_sync.generate_sequential_events_list(spec_list)
transformer = NagiosTransformer(self.transformers)
event = nagios_alarms[0]
# Test action
observed_key = transformer.extract_key(event)
# Test assertions
observed_key_fields = observed_key.split(
TransformerBase.KEY_SEPARATOR)
self.assertEqual(EntityCategory.ALARM, observed_key_fields[0])
self.assertEqual(event[SyncProps.SYNC_TYPE], observed_key_fields[1])
self.assertEqual(event[NagiosProperties.RESOURCE_NAME],
observed_key_fields[2])
self.assertEqual(event[NagiosProperties.SERVICE],
observed_key_fields[3])
def test_nagios_alarm_transform(self):
LOG.debug('Nagios alarm transformer test: transform entity event')
# Test setup
spec_list = mock_sync.simple_nagios_alarm_generators(host_num=4,
events_num=4)
events_num=10)
nagios_alarms = mock_sync.generate_sequential_events_list(spec_list)
# Test action
for alarm in nagios_alarms:
NagiosTransformer(self.transformers)._create_entity_vertex(alarm)
# Test action
wrapper = NagiosTransformer(self.transformers).transform(alarm)
self._validate_vertex(wrapper.vertex, alarm)
neighbors = wrapper.neighbors
self.assertEqual(1, len(neighbors))
neighbor = neighbors[0]
# Right now we are support only host as a resource
if neighbor.vertex[VProps.TYPE] == EntityType.NOVA_HOST:
self._validate_host_neighbor(neighbors[0], alarm)
self._validate_action(alarm, wrapper)
def _validate_action(self, alarm, wrapper):
sync_mode = alarm[SyncProps.SYNC_MODE]
if sync_mode in (SyncMode.SNAPSHOT, SyncMode.UPDATE):
if alarm[NagiosProperties.STATUS] == 'OK':
self.assertEqual(EventAction.DELETE, wrapper.action)
else:
self.assertEqual(EventAction.UPDATE, wrapper.action)
else:
self.assertEqual(EventAction.CREATE, wrapper.action)
def _validate_vertex(self, vertex, event):
self.assertEqual(EntityCategory.ALARM, vertex[VProps.CATEGORY])
self.assertEqual(event[SyncProps.SYNC_TYPE], vertex[VProps.TYPE])
self.assertEqual(event[NagiosProperties.SERVICE], vertex[VProps.NAME])
self.assertEqual(NagiosTransformer.NAGIOS_ALARM_STATE,
vertex[VProps.STATE])
self.assertEqual(event[NagiosProperties.STATUS],
vertex[VProps.SEVERITY])
self.assertEqual(event[NagiosProperties.STATUS_INFO],
vertex[VProps.INFO])
self.assertFalse(vertex[VProps.IS_DELETED])
self.assertFalse(vertex[VProps.IS_PLACEHOLDER])
def _validate_host_neighbor(self, neighbor, event):
host_vertex = neighbor.vertex
key_fields = host_vertex.vertex_id.split(TransformerBase.KEY_SEPARATOR)
self.assertEqual(EntityCategory.RESOURCE, key_fields[0])
self.assertEqual(EntityType.NOVA_HOST, key_fields[1])
self.assertEqual(event[NagiosProperties.RESOURCE_NAME], key_fields[2])
self.assertFalse(host_vertex[VProps.IS_DELETED])
self.assertTrue(host_vertex[VProps.IS_PLACEHOLDER])
self.assertEqual(EntityCategory.RESOURCE, host_vertex[VProps.CATEGORY])
self.assertEqual(event[NagiosProperties.RESOURCE_NAME],
host_vertex[VProps.ID])
self.assertEqual(EntityType.NOVA_HOST, host_vertex[VProps.TYPE])
edge = neighbor.edge
self.assertEqual(EdgeLabels.ON, edge.label)
alarm_key = NagiosTransformer(self.transformers).extract_key(event)
self.assertEqual(alarm_key, edge.source_id)
self.assertEqual(host_vertex.vertex_id, edge.target_id)

11
vitrage/tests/unit/synchronizer/nova/test_nova_instance_transformers.py
View File

@ -49,20 +49,19 @@ class NovaInstanceTransformerTest(base.BaseTest):
# Tests setup
instance_id = 'Instance123'
timestamp = datetime.datetime.utcnow()
# Test action
instance_transformer = InstanceTransformer(self.transformers)
properties = {
VertexProperties.ID: instance_id,
VertexProperties.UPDATE_TIMESTAMP: timestamp
}
placeholder = instance_transformer.create_placeholder_vertex(
properties)
transformer = InstanceTransformer(self.transformers)
# Test action
placeholder = transformer.create_placeholder_vertex(properties)
# Test assertions
observed_id_values = placeholder.vertex_id.split(
TransformerBase.KEY_SEPARATOR)
expected_id_values = instance_transformer.key_values([instance_id])
expected_id_values = transformer.key_values([instance_id])
self.assertEqual(observed_id_values, expected_id_values)
observed_time = placeholder.get(VertexProperties.UPDATE_TIMESTAMP)