openstack/vitrage
Code Issues Proposed changes

tests optimizations - add assert validations

Browse Source 
Change-Id: Id1cadc8d756bb7caa4e8279edbc7049c070be266
This commit is contained in:
Marina Koushnir 2016-06-07 14:59:18 +00:00
parent bffa8dd42b
commit e96c579d79
7 changed files with 266 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
vitrage_tempest_tests/tests/api/alarms/base.py
View File

@ -11,16 +11,15 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import random
import time
from oslo_log import log as logging
from vitrage import clients
from vitrage_tempest_tests.tests.api.base import BaseApiTest
import vitrage_tempest_tests.tests.utils as utils
LOG = logging.getLogger(__name__)
TEMPLATES_RESOURCES_PATH = 'resources/templates/'
@ -35,21 +34,6 @@ class BaseAlarmsTest(BaseApiTest):
super(BaseAlarmsTest, cls).setUpClass()
cls.ceilometer_client = clients.ceilometer_client(cls.conf)
@staticmethod
def _filter_alarms_by_parameter(alarms_list,
keys, values):
filtered_alarms_list = []
for item in alarms_list:
verification = 0
for index in range(len(keys)):
if utils.uni2str(item[keys[index]]) == values[index]:
verification += 1
else:
break
if verification == len(keys):
filtered_alarms_list.append(item)
return filtered_alarms_list
def _create_ceilometer_alarm(self, resource_id=None,
name=None, unic=True):
if not name:

38
vitrage_tempest_tests/tests/api/alarms/test_alarms.py
View File

@ -11,13 +11,14 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
from oslo_log import log as logging
from vitrage.common.constants import VertexProperties as VProps
from vitrage.datasources.aodh import AODH_DATASOURCE
from vitrage_tempest_tests.tests.api.alarms.base import BaseAlarmsTest
import vitrage_tempest_tests.tests.utils as utils
LOG = logging.getLogger(__name__)
@ -33,16 +34,18 @@ class TestAlarms(BaseAlarmsTest):
def test_compare_cli_vs_api_alarms(self):
"""Wrapper that returns a test graph."""
try:
resources = self._create_instances(num_instances=1)
self._create_ceilometer_alarm(resource_id=resources[0].id,
instances = self._create_instances(num_instances=1)
self.assertNotEqual(len(instances), 0,
'The instances list is empty')
self._create_ceilometer_alarm(resource_id=instances[0].id,
name='tempest_aodh_test')
api_alarms = self.vitrage_client.alarms.list(vitrage_id=None)
cli_alarms = utils.run_vitrage_command(
'vitrage alarms list', self.conf)
self.assertTrue(self._compare_alarms_lists(
self._compare_alarms_lists(
api_alarms, cli_alarms, AODH_DATASOURCE,
utils.uni2str(resources[0].id)))
utils.uni2str(instances[0].id))
finally:
self._delete_ceilometer_alarms()
@ -51,12 +54,10 @@ class TestAlarms(BaseAlarmsTest):
def _compare_alarms_lists(self, api_alarms, cli_alarms,
resource_type, resource_id):
"""Validate alarm existence """
if not api_alarms:
LOG.error("The alarms list taken from api is empty")
return False
if cli_alarms is None:
LOG.error("The alarms list taken from cli is empty")
return False
self.assertNotEqual(len(api_alarms), 0,
'The alarms list taken from api is empty')
self.assertIsNotNone(cli_alarms,
'The alarms list taken from cli is empty')
LOG.debug("The alarms list taken from cli is : %s", cli_alarms)
LOG.debug("The alarms list taken by api is : %s",
@ -64,13 +65,14 @@ class TestAlarms(BaseAlarmsTest):
cli_items = cli_alarms.splitlines()
api_by_type = self._filter_alarms_by_parameter(
api_alarms, ['type'], [resource_type])
api_by_type = self._filter_list_by_pairs_parameters(
api_alarms, [VProps.TYPE], [resource_type])
cli_by_type = cli_alarms.count(' ' + resource_type + ' ')
api_by_id = self._filter_alarms_by_parameter(
api_by_id = self._filter_list_by_pairs_parameters(
api_alarms, ['resource_id'], [resource_id])
cli_by_id = cli_alarms.count(resource_id)
return (len(cli_items) - 4 == len(api_alarms) and
cli_by_type == len(api_by_type) and
cli_by_id == len(api_by_id))
self.assertEqual(len(cli_items), len(api_alarms) + 4)
self.assertEqual(cli_by_type, len(api_by_type))
self.assertEqual(cli_by_id, len(api_by_id))

23
vitrage_tempest_tests/tests/api/base.py
View File

@ -11,6 +11,7 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import time
from oslo_log import log as logging
@ -31,9 +32,8 @@ from vitrage.graph import NXGraph
from vitrage.graph import Vertex
from vitrage import keystone_client
from vitrage_tempest_tests.tests import OPTS
from vitrageclient import client as v_client
import vitrage_tempest_tests.tests.utils as utils
from vitrageclient import client as v_client
LOG = logging.getLogger(__name__)
@ -56,6 +56,21 @@ class BaseApiTest(base.BaseTestCase):
cls.nova_client = clients.nova_client(cls.conf)
cls.cinder_client = clients.cinder_client(cls.conf)
@staticmethod
def _filter_list_by_pairs_parameters(origin_list,
keys, values):
filtered_list = []
for item in origin_list:
verification = 0
for index in range(len(keys)):
if utils.uni2str(item[keys[index]]) == values[index]:
verification += 1
else:
break
if verification == len(keys):
filtered_list.append(item)
return filtered_list
def _create_volume_and_attach(self, name, size, instance_id, mount_point):
volume = self.cinder_client.volumes.create(display_name=name,
size=size)
@ -250,8 +265,8 @@ class BaseApiTest(base.BaseTestCase):
num_entities,
num_edges,
entities):
self.assertIsNot(None, graph)
self.assertIsNot(None, entities)
self.assertIsNotNone(graph)
self.assertIsNotNone(entities)
self.assertEqual(num_entities, graph.num_vertices())
self.assertEqual(num_edges, graph.num_edges())

199
vitrage_tempest_tests/tests/api/rca/base.py Normal file
View File

@ -0,0 +1,199 @@
# Copyright 2016 - Nokia
#
# 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.
import json
from oslo_log import log as logging
from vitrage.common.constants import VertexProperties as VProps
from vitrage.datasources import AODH_DATASOURCE
from vitrage.datasources import NOVA_HOST_DATASOURCE
from vitrage.datasources import NOVA_INSTANCE_DATASOURCE
from vitrage.entity_graph.mappings.operational_alarm_severity \
import OperationalAlarmSeverity
from vitrage.entity_graph.mappings.operational_resource_state \
import OperationalResourceState
from vitrage.evaluator.actions.evaluator_event_transformer \
import VITRAGE_TYPE
from vitrage_tempest_tests.tests.api.alarms.base import BaseAlarmsTest
import vitrage_tempest_tests.tests.utils as utils
LOG = logging.getLogger(__name__)
RCA_ALARM_NAME = 'rca_test_host_alarm'
VITRAGE_ALARM_NAME = 'instance_deduce'
class BaseRcaTest(BaseAlarmsTest):
@classmethod
def setUpClass(cls):
super(BaseRcaTest, cls).setUpClass()
def _clean_all(self):
self._delete_instances()
self._delete_ceilometer_alarms()
def _create_alarm(self, resource_id, alarm_name, unic=False):
self._create_ceilometer_alarm(resource_id=resource_id,
name=alarm_name, unic=unic)
list_alarms = self.vitrage_client.alarms.list(vitrage_id=None)
expected_alarm = self._filter_list_by_pairs_parameters(
list_alarms, ['resource_id', VProps.TYPE],
[resource_id, AODH_DATASOURCE])
if not expected_alarm:
return None
return expected_alarm[0]
def _compare_rca(self, api_rca, cli_rca):
self.assertNotEqual(len(api_rca), 0, 'The rca taken from api is empty')
self.assertIsNotNone(cli_rca, 'The rca taken from cli is empty')
LOG.debug("The rca taken from cli is : %s", cli_rca)
LOG.debug("The rca taken by api is : %s",
json.dumps(api_rca))
parsed_rca = json.loads(cli_rca)
sorted_cli_graph = self._clean_timestamps(sorted(parsed_rca.items()))
sorted_api_graph = self._clean_timestamps(sorted(api_rca.items()))
self.assertEqual(sorted_cli_graph, sorted_api_graph)
def _validate_rca(self, rca):
self.assertNotEqual(len(rca), 0, 'The rca is empty')
LOG.debug("The rca alarms list is : %s",
json.dumps(rca))
resource_alarm = self._filter_list_by_pairs_parameters(
rca, [VProps.TYPE, VProps.NAME],
[AODH_DATASOURCE, RCA_ALARM_NAME])
deduce_alarms = self._filter_list_by_pairs_parameters(
rca, [VProps.TYPE, VProps.NAME, VProps.SEVERITY],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
OperationalAlarmSeverity.WARNING])
self.assertEqual(len(rca), 3)
self.assertEqual(len(resource_alarm), 1)
self.assertEqual(len(deduce_alarms), 2)
def _validate_deduce_alarms(self, alarms, instances):
"""Validate alarm existence """
self.assertNotEqual(len(alarms), 0, 'The alarms list is empty')
LOG.debug("The alarms list is : %s",
json.dumps(alarms))
deduce_alarms_1 = self._filter_list_by_pairs_parameters(
alarms,
[VProps.TYPE, VProps.NAME, 'resource_type', 'resource_id'],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[0].id)])
deduce_alarms_2 = self._filter_list_by_pairs_parameters(
alarms,
[VProps.TYPE, VProps.NAME, 'resource_type', 'resource_id'],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[1].id)])
self.assertEqual(len(alarms), 3)
self.assertEqual(len(deduce_alarms_1), 1)
self.assertEqual(len(deduce_alarms_2), 1)
def _validate_relationship(self, links, alarms):
self.assertNotEqual(len(links), 0, 'The links list is empty')
self.assertNotEqual(len(alarms), 0, 'The alarms list is empty')
flag = True
for item in links:
source_alarm_name = self._get_value(
alarms[item['source']], VProps.NAME)
target_alarm_name = self._get_value(
alarms[item['target']], VProps.NAME)
if self._get_value(item, 'key') != 'causes' \
or self._get_value(item, 'relationship_type') != 'causes' \
or source_alarm_name != RCA_ALARM_NAME \
or target_alarm_name != VITRAGE_ALARM_NAME:
flag = False
self.assertEqual(len(alarms), 3)
self.assertTrue(flag)
def _validate_set_state(self, topology, instances):
self.assertNotEqual(len(topology), 0, 'The topology graph is empty')
host = self._filter_list_by_pairs_parameters(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_HOST_DATASOURCE,
self._get_hostname(),
OperationalResourceState.ERROR,
OperationalResourceState.ERROR])
vm1 = self._filter_list_by_pairs_parameters(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[0].id),
OperationalResourceState.SUBOPTIMAL,
OperationalResourceState.SUBOPTIMAL])
vm2 = self._filter_list_by_pairs_parameters(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[1].id),
OperationalResourceState.SUBOPTIMAL,
OperationalResourceState.SUBOPTIMAL])
self.assertEqual(len(host), 1)
self.assertEqual(len(vm1), 1)
self.assertEqual(len(vm2), 1)
def _validate_notifier(self, alarms, vitrage_alarms):
self.assertNotEqual(len(alarms), 0, 'The aodh alarms list is empty')
self.assertNotEqual(len(vitrage_alarms), 0,
'The vitrage alarms list is empty')
validation = 0
for itemC in alarms:
vitrage_id = filter(
lambda item: item['field'] == VProps.VITRAGE_ID,
itemC.event_rule['query'])
for itemV in vitrage_alarms:
if not vitrage_id:
if itemC.name == itemV[VProps.NAME]:
validation += 1
break
elif vitrage_id[0]['value'] == itemV[VProps.VITRAGE_ID]:
validation += 1
break
self.assertEqual(len(vitrage_alarms), validation)
self.assertEqual(len(alarms), 3)
def _get_hostname(self):
return self._get_value(item=self._get_host(), key=VProps.ID)
@staticmethod
def _clean_timestamps(alist):
try:
del alist[5][1][0][VProps.SAMPLE_TIMESTAMP]
del alist[5][1][0][VProps.UPDATE_TIMESTAMP]
except Exception:
pass
return alist

228
vitrage_tempest_tests/tests/api/rca/test_rca.py
View File

@ -11,29 +11,18 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
from oslo_log import log as logging
from vitrage.common.constants import VertexProperties as VProps
from vitrage.datasources import AODH_DATASOURCE
from vitrage.datasources import NOVA_HOST_DATASOURCE
from vitrage.datasources import NOVA_INSTANCE_DATASOURCE
from vitrage.entity_graph.mappings.operational_alarm_severity \
import OperationalAlarmSeverity
from vitrage.entity_graph.mappings.operational_resource_state \
import OperationalResourceState
from vitrage.evaluator.actions.evaluator_event_transformer \
import VITRAGE_TYPE
from vitrage_tempest_tests.tests.api.alarms.base import BaseAlarmsTest
from vitrage.common.constants import VertexProperties as VProps
from vitrage_tempest_tests.tests.api.rca.base import BaseRcaTest
from vitrage_tempest_tests.tests.api.rca.base import RCA_ALARM_NAME
import vitrage_tempest_tests.tests.utils as utils
LOG = logging.getLogger(__name__)
RCA_ALARM_NAME = 'rca_test_host_alarm'
VITRAGE_ALARM_NAME = 'instance_deduce'
class TestRca(BaseAlarmsTest):
class TestRca(BaseRcaTest):
"""RCA test class for Vitrage API tests."""
@classmethod
@ -48,12 +37,10 @@ class TestRca(BaseAlarmsTest):
"""
try:
instances = self._create_instances(num_instances=1)
if not instances:
LOG.error('Failed to create instance', False)
return False
self.assertNotEqual(len(instances), 0, 'Failed to create instance')
instance_alarm = self._create_alarm(
resource_id=utils.uni2str(text=instances[0].id),
resource_id=instances[0].id,
alarm_name='instance_rca_alarm', unic=True)
vitrage_id = self._get_value(
@ -62,7 +49,7 @@ class TestRca(BaseAlarmsTest):
cli_rca = utils.run_vitrage_command(
'vitrage rca show ' + vitrage_id, self.conf)
self.assertTrue(self._compare_rca(api_rca, cli_rca))
self._compare_rca(api_rca, cli_rca)
except Exception:
LOG.error('Got exception', False)
finally:
@ -87,10 +74,9 @@ class TestRca(BaseAlarmsTest):
alarm_id=self._get_value(host_alarm,
VProps.VITRAGE_ID))
self.assertTrue(self._validate_rca(rca=api_rca['nodes']))
self.assertTrue(self._validate_relationship(
links=api_rca['links'],
alarms=api_rca['nodes']))
self._validate_rca(rca=api_rca['nodes'])
self._validate_relationship(links=api_rca['links'],
alarms=api_rca['nodes'])
except Exception:
LOG.error('Got exception', False)
finally:
@ -110,9 +96,8 @@ class TestRca(BaseAlarmsTest):
alarm_name=RCA_ALARM_NAME)
api_alarms = self.vitrage_client.alarms.list(vitrage_id=None)
self.assertTrue(self._validate_deduce_alarms(
alarms=api_alarms,
instances=instances))
self._validate_deduce_alarms(alarms=api_alarms,
instances=instances)
except Exception:
LOG.error('Got exception', False)
finally:
@ -133,9 +118,8 @@ class TestRca(BaseAlarmsTest):
alarm_name=RCA_ALARM_NAME)
topology = self.vitrage_client.topology.get()
self.assertTrue(self._validate_set_state(
topology=topology['nodes'],
instances=instances))
self._validate_set_state(topology=topology['nodes'],
instances=instances)
except Exception:
LOG.error('Got exception', False)
finally:
@ -157,189 +141,9 @@ class TestRca(BaseAlarmsTest):
vitrage_alarms = self.vitrage_client.alarms.list(vitrage_id=None)
ceilometer_alarms = self.ceilometer_client.alarms.list()
self.assertTrue(self._validate_notifier(
alarms=ceilometer_alarms,
vitrage_alarms=vitrage_alarms))
self._validate_notifier(alarms=ceilometer_alarms,
vitrage_alarms=vitrage_alarms)
except Exception:
LOG.error('Got exception', False)
finally:
self._clean_all()
def _clean_all(self):
self._delete_instances()
self._delete_ceilometer_alarms()
def _create_alarm(self, resource_id, alarm_name, unic=False):
if not resource_id:
raise Exception("Can't create alarm with empty resource id")
self._create_ceilometer_alarm(resource_id=resource_id,
name=alarm_name, unic=unic)
list_alarms = self.vitrage_client.alarms.list(vitrage_id=None)
expected_alarm = self._filter_alarms_by_parameter(
list_alarms, ['resource_id', 'type'],
[resource_id, AODH_DATASOURCE])
if not expected_alarm:
return None
return expected_alarm[0]
def _compare_rca(self, api_rca, cli_rca):
if not api_rca:
LOG.error("The rca taken from api is empty")
return False
if cli_rca is None:
LOG.error("The rca taken from cli is empty")
return False
LOG.debug("The rca taken from cli is : %s", cli_rca)
LOG.debug("The rca taken by api is : %s",
json.dumps(api_rca))
parsed_rca = json.loads(cli_rca)
sorted_cli_graph = self._clean_timestamps(sorted(parsed_rca.items()))
sorted_api_graph = self._clean_timestamps(sorted(api_rca.items()))
return sorted_cli_graph == sorted_api_graph
def _validate_rca(self, rca):
if not rca:
LOG.error("The alarms list is empty")
return False
LOG.debug("The rca alarms list is : %s",
json.dumps(rca))
resource_alarm = self._filter_alarms_by_parameter(
rca, ['type', 'name'],
[AODH_DATASOURCE, RCA_ALARM_NAME])
deduce_alarms = self._filter_alarms_by_parameter(
rca, ['type', 'name', 'severity'],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
OperationalAlarmSeverity.WARNING])
return (len(rca) == 3 and
len(resource_alarm) == 1 and
len(deduce_alarms) == 2)
def _validate_deduce_alarms(self, alarms, instances):
"""Validate alarm existence """
if not alarms:
LOG.error("The alarms list is empty")
return False
LOG.debug("The alarms list is : %s",
json.dumps(alarms))
deduce_alarms_1 = self._filter_alarms_by_parameter(
alarms,
['type', 'name', 'resource_type', 'resource_id'],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[0].id)])
deduce_alarms_2 = self._filter_alarms_by_parameter(
alarms,
['type', 'name', 'resource_type', 'resource_id'],
[VITRAGE_TYPE, VITRAGE_ALARM_NAME,
NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[1].id)])
return (len(alarms) == 3 and
len(deduce_alarms_1) == 1 and
len(deduce_alarms_2) == 1)
def _validate_relationship(self, links, alarms):
if not links:
LOG.error("The links list is empty")
return False
if not alarms:
LOG.error("The alarms list is empty")
return False
flag = True
for item in links:
source_alarm_name = self._get_value(
alarms[item['source']], VProps.NAME)
target_alarm_name = self._get_value(
alarms[item['target']], VProps.NAME)
if self._get_value(item, 'key') != 'causes' \
or self._get_value(item, 'relationship_type') != 'causes' \
or source_alarm_name != RCA_ALARM_NAME \
or target_alarm_name != VITRAGE_ALARM_NAME:
flag = False
return flag and len(alarms) == 3
def _validate_set_state(self, topology, instances):
if not topology:
LOG.error("The topology list is empty")
return False
host = self._filter_alarms_by_parameter(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_HOST_DATASOURCE,
self._get_hostname(),
OperationalResourceState.ERROR,
OperationalResourceState.ERROR])
vm1 = self._filter_alarms_by_parameter(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[0].id),
OperationalResourceState.SUBOPTIMAL,
OperationalResourceState.SUBOPTIMAL])
vm2 = self._filter_alarms_by_parameter(
topology,
[VProps.TYPE, VProps.ID, VProps.VITRAGE_STATE,
VProps.AGGREGATED_STATE],
[NOVA_INSTANCE_DATASOURCE,
utils.uni2str(instances[1].id),
OperationalResourceState.SUBOPTIMAL,
OperationalResourceState.SUBOPTIMAL])
return len(host) == 1 and len(vm1) == 1 and len(vm2) == 1
@staticmethod
def _validate_notifier(alarms, vitrage_alarms):
if not alarms:
LOG.error("The alarms list is empty")
return False
if not vitrage_alarms:
LOG.error("The vitrage alarms list is empty")
return False
validation = 0
for itemC in alarms:
vitrage_id = filter(
lambda item: item['field'] == VProps.VITRAGE_ID,
itemC.event_rule['query'])
for itemV in vitrage_alarms:
if not vitrage_id:
if itemC.name == itemV[VProps.NAME]:
validation += 1
break
elif vitrage_id[0]['value'] == itemV[VProps.VITRAGE_ID]:
validation += 1
break
return (len(vitrage_alarms) == validation
and len(alarms) == 3)
@staticmethod
def _clean_timestamps(alist):
try:
del alist[5][1][0][VProps.SAMPLE_TIMESTAMP]
del alist[5][1][0][VProps.UPDATE_TIMESTAMP]
except Exception:
pass
return alist
def _get_hostname(self):
return self._get_value(item=self._get_host(), key=VProps.ID)

19
vitrage_tempest_tests/tests/api/topology/base.py
View File

@ -20,6 +20,7 @@ from oslo_log import log as logging
from vitrage.common.constants import VertexProperties as VProps
from vitrage_tempest_tests.tests.api.base import BaseApiTest
LOG = logging.getLogger(__name__)
@ -41,9 +42,10 @@ class BaseTopologyTest(BaseApiTest):
if num_instances > 0:
resources = self._create_instances(num_instances)
self.assertNotEqual(len(resources), 0, 'The instances list is empty')
if num_volumes > 0:
self._create_volume_and_attach('volume-1', 1,
resources[0].__dict__['id'],
resources[0].id,
'/tmp/vda')
# waiting until all the entities creation were processed by the
@ -58,15 +60,12 @@ class BaseTopologyTest(BaseApiTest):
# entity graph processor
time.sleep(2)
@staticmethod
def _compare_graphs(api_graph, cli_graph):
def _compare_graphs(self, api_graph, cli_graph):
"""Compare Graph object to graph form terminal """
if not api_graph:
LOG.error("The topology graph taken from rest api is empty")
return False
if not cli_graph:
LOG.error("The topology graph taken from terminal is empty")
return False
self.assertNotEqual(len(api_graph), 0,
'The topology graph taken from rest api is empty')
self.assertNotEqual(len(cli_graph), 0,
'The topology graph taken from terminal is empty')
parsed_topology = json.loads(cli_graph)
@ -79,7 +78,7 @@ class BaseTopologyTest(BaseApiTest):
for item in sorted_api_graph[4][1]:
item.pop(VProps.UPDATE_TIMESTAMP, None)
return sorted_cli_graph == sorted_api_graph
self.assertEqual(sorted_cli_graph, sorted_api_graph)
@staticmethod
def _graph_query():

2
vitrage_tempest_tests/tests/api/topology/test_topology.py
View File

@ -40,7 +40,7 @@ class TestTopology(BaseTopologyTest):
api_graph = self.vitrage_client.topology.get()
cli_graph = utils.run_vitrage_command('vitrage topology show',
self.conf)
self.assertEqual(True, self._compare_graphs(api_graph, cli_graph))
self._compare_graphs(api_graph, cli_graph)
def test_default_graph(self):
try: