--- /dev/null
+#!/usr/bin/env python3
+
+#
+# Copyright 2016 RIFT.IO Inc
+#
+# 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 os
+import unittest
+import xmlrunner
+
+import rw_peas
+
+import gi
+gi.require_version("RwcalYang", "1.0")
+gi.require_version("RwMon", "1.0")
+gi.require_version("RwmonYang", "1.0")
+gi.require_version("RwTypes", "1.0")
+
+
+from gi.repository import (
+ RwmonYang,
+ RwcalYang,
+ RwTypes,
+ )
+
+
+class TestNullDataSource(unittest.TestCase):
+ def setUp(self):
+ plugin = rw_peas.PeasPlugin("rwmon_mock", 'RwMon-1.0')
+ self.plugin = plugin.get_interface("Monitoring")
+
+ self.account = RwcalYang.CloudAccount()
+ self.vim_id = "test-vim-id"
+
+ def test_null_data_source(self):
+ """
+ By default, the NFVI metrics plugin mock installs a 'null'
+ implementation that simply returns empty NFVI structures.
+
+ """
+ status, metrics = self.plugin.nfvi_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics, RwmonYang.NfviMetrics())
+
+ status, metrics = self.plugin.nfvi_vcpu_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics, RwmonYang.NfviMetrics_Vcpu())
+
+ status, metrics = self.plugin.nfvi_memory_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics, RwmonYang.NfviMetrics_Memory())
+
+ status, metrics = self.plugin.nfvi_storage_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics, RwmonYang.NfviMetrics_Storage())
+
+ status, result = self.plugin.nfvi_metrics_available(self.account)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertTrue(result)
+
+
+class TestMockDataSource(unittest.TestCase):
+ def setUp(self):
+ plugin = rw_peas.PeasPlugin("rwmon_mock", 'RwMon-1.0')
+ self.plugin = plugin.get_interface("Monitoring")
+ self.plugin.set_impl(MockDataSource())
+
+ self.account = RwcalYang.CloudAccount()
+ self.vim_id = "test-vim-id"
+
+ def test_mock_data_source(self):
+ """
+ This test installs a mock data source implementation in the plugin,
+ which returns known values. This test simply checks the expected values
+ are indeed returned.
+
+ """
+ expected_vcpu_metrics = RwmonYang.NfviMetrics_Vcpu()
+ expected_vcpu_metrics.utilization = 50.0
+ expected_vcpu_metrics.total = 100
+
+ status, metrics = self.plugin.nfvi_vcpu_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics.total, expected_vcpu_metrics.total)
+ self.assertEqual(metrics.utilization, expected_vcpu_metrics.utilization)
+
+ expected_memory_metrics = RwmonYang.NfviMetrics_Memory()
+ expected_memory_metrics.used = 90
+ expected_memory_metrics.total = 100
+ expected_memory_metrics.utilization = 90/100
+
+ status, metrics = self.plugin.nfvi_memory_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics.used, expected_memory_metrics.used)
+ self.assertEqual(metrics.total, expected_memory_metrics.total)
+ self.assertEqual(metrics.utilization, expected_memory_metrics.utilization)
+
+ expected_storage_metrics = RwmonYang.NfviMetrics_Storage()
+ expected_storage_metrics.used = 300
+ expected_storage_metrics.total = 500
+ expected_storage_metrics.utilization = 300/500
+
+ status, metrics = self.plugin.nfvi_storage_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics.used, expected_storage_metrics.used)
+ self.assertEqual(metrics.total, expected_storage_metrics.total)
+ self.assertEqual(metrics.utilization, expected_storage_metrics.utilization)
+
+ status, metrics = self.plugin.nfvi_metrics(self.account, self.vim_id)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertEqual(metrics.vcpu.total, expected_vcpu_metrics.total)
+ self.assertEqual(metrics.vcpu.utilization, expected_vcpu_metrics.utilization)
+ self.assertEqual(metrics.storage.used, expected_storage_metrics.used)
+ self.assertEqual(metrics.storage.total, expected_storage_metrics.total)
+ self.assertEqual(metrics.storage.utilization, expected_storage_metrics.utilization)
+ self.assertEqual(metrics.memory.used, expected_memory_metrics.used)
+ self.assertEqual(metrics.memory.total, expected_memory_metrics.total)
+ self.assertEqual(metrics.memory.utilization, expected_memory_metrics.utilization)
+
+ status, result = self.plugin.nfvi_metrics_available(self.account)
+ self.assertEqual(status, RwTypes.RwStatus.SUCCESS)
+ self.assertTrue(result)
+
+
+class TestMockAlarms(unittest.TestCase):
+ def setUp(self):
+ plugin = rw_peas.PeasPlugin("rwmon_mock", 'RwMon-1.0')
+
+ self.mock = MockAlarmInterface()
+ self.plugin = plugin.get_interface("Monitoring")
+ self.plugin.set_impl(self.mock)
+
+ self.account = RwcalYang.CloudAccount()
+ self.alarm = RwmonYang.Alarm(name='test-alarm')
+ self.vim_id = 'test-vim-id'
+
+ def test(self):
+ """
+ This test uses a simple, mock implementation of the alarm interface to
+ check that create, update, delete, and list work correctly.
+
+ """
+ # In the beginning, there were no alarms
+ _, alarms = self.plugin.do_alarm_list(self.account)
+ self.assertEqual(0, len(alarms))
+
+ # Create two alarms
+ self.plugin.do_alarm_create(self.account, self.vim_id, RwmonYang.Alarm())
+ self.plugin.do_alarm_create(self.account, self.vim_id, RwmonYang.Alarm())
+
+ _, alarms = self.plugin.do_alarm_list(self.account)
+ self.assertEqual(2, len(alarms))
+
+ # The alarms should have no names
+ alarms.sort(key=lambda a: a.alarm_id)
+ self.assertEqual('test-alarm-id-1', alarms[0].alarm_id)
+ self.assertEqual('test-alarm-id-2', alarms[1].alarm_id)
+ self.assertTrue(all(a.name is None for a in alarms))
+
+ # Give names to the alarms
+ alarms[0].name = 'test-alarm'
+ alarms[1].name = 'test-alarm'
+ self.plugin.do_alarm_update(self.account, alarms[0])
+ self.plugin.do_alarm_update(self.account, alarms[1])
+ self.assertTrue(all(a.name == 'test-alarm' for a in alarms))
+
+ # Delete the alarms
+ self.plugin.do_alarm_delete(self.account, alarms[0].alarm_id)
+ self.plugin.do_alarm_delete(self.account, alarms[1].alarm_id)
+ _, alarms = self.plugin.do_alarm_list(self.account)
+ self.assertEqual(0, len(alarms))
+
+
+class MockAlarmInterface(object):
+ """
+ This class is mock impementation for the alarm interface on the monitoring
+ plugin.
+ """
+
+ def __init__(self):
+ self.count = 0
+ self.alarms = dict()
+
+ def alarm_create(self, account, vim_id, alarm):
+ self.count += 1
+ alarm_id = 'test-alarm-id-{}'.format(self.count)
+ alarm.alarm_id = alarm_id
+ self.alarms[alarm_id] = alarm
+
+ def alarm_update(self, account, alarm):
+ assert alarm.alarm_id is not None
+ self.alarms[alarm.alarm_id] = alarm
+
+ def alarm_delete(self, account, alarm_id):
+ del self.alarms[alarm_id]
+
+ def alarm_list(self, account):
+ return list(self.alarms.values())
+
+
+class MockDataSource(object):
+ """
+ This class implements the data source interface used by the monitoring
+ plugin and provides mock data for testing.
+ """
+
+ def nfvi_metrics(self, account, vm_id):
+ metrics = RwmonYang.NfviMetrics()
+ metrics.vcpu = self.nfvi_vcpu_metrics(account, vm_id)
+ metrics.memory = self.nfvi_memory_metrics(account, vm_id)
+ metrics.storage = self.nfvi_storage_metrics(account, vm_id)
+ return metrics
+
+ def nfvi_vcpu_metrics(self, account, vm_id):
+ metrics = RwmonYang.NfviMetrics_Vcpu()
+ metrics.total = 100
+ metrics.utilization = 50.0
+ return metrics
+
+ def nfvi_memory_metrics(self, account, vm_id):
+ metrics = RwmonYang.NfviMetrics_Memory()
+ metrics.used = 90
+ metrics.total = 100
+ metrics.utilization = 90/100
+ return metrics
+
+ def nfvi_storage_metrics(self, account, vm_id):
+ metrics = RwmonYang.NfviMetrics_Storage()
+ metrics.used = 300
+ metrics.total = 500
+ metrics.utilization = 300/500
+ return metrics
+
+ def nfvi_metrics_available(self, account):
+ return True
+
+
+def main():
+ runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
+ unittest.main(testRunner=runner)
+
+
+if __name__ == '__main__':
+ main()
+
+# vim: sw=4
projects / osm / SO.git / blobdiff