rwmon/test/utest_rwmon.py

#!/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