yardstick/tests/unit/network_services/collector/test_subscriber.py

   1 # Copyright (c) 2016-2017 Intel Corporation
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 #
  15
  16 import unittest
  17 import mock
  18
  19 from yardstick.network_services.collector import subscriber
  20 from yardstick import ssh
  21
  22
  23 class MockVnfAprrox(object):
  24
  25     def __init__(self):
  26         self.result = {}
  27         self.name = "vnf__1"
  28
  29     def collect_kpi(self):
  30         self.result = {
  31             'pkt_in_up_stream': 100,
  32             'pkt_drop_up_stream': 5,
  33             'pkt_in_down_stream': 50,
  34             'pkt_drop_down_stream': 40
  35         }
  36         return self.result
  37
  38
  39 class CollectorTestCase(unittest.TestCase):
  40
  41     def setUp(self):
  42         vnf = MockVnfAprrox()
  43         vnf.start_collect = mock.Mock()
  44         vnf.stop_collect = mock.Mock()
  45         self.ssh_patch = mock.patch.object(ssh, 'AutoConnectSSH')
  46         mock_ssh = self.ssh_patch.start()
  47         mock_instance = mock.Mock()
  48         mock_instance.execute.return_value = 0, '', ''
  49         mock_ssh.from_node.return_value = mock_instance
  50         self.collector = subscriber.Collector([vnf])
  51
  52     def tearDown(self):
  53         self.ssh_patch.stop()
  54
  55     def test___init__(self, *_):
  56         vnf = MockVnfAprrox()
  57         collector = subscriber.Collector([vnf])
  58         self.assertEqual(len(collector.vnfs), 1)
  59
  60     def test_start(self, *_):
  61         self.assertIsNone(self.collector.start())
  62         for vnf in self.collector.vnfs:
  63             vnf.start_collect.assert_called_once()
  64
  65     def test_stop(self, *_):
  66         self.assertIsNone(self.collector.stop())
  67         for vnf in self.collector.vnfs:
  68             vnf.stop_collect.assert_called_once()
  69
  70     def test_get_kpi(self, *_):
  71         result = self.collector.get_kpi()
  72
  73         self.assertEqual(1, len(result))
  74         self.assertEqual(4, len(result["vnf__1"]))
  75         self.assertEqual(result["vnf__1"]["pkt_in_up_stream"], 100)
  76         self.assertEqual(result["vnf__1"]["pkt_drop_up_stream"], 5)
  77         self.assertEqual(result["vnf__1"]["pkt_in_down_stream"], 50)
  78         self.assertEqual(result["vnf__1"]["pkt_drop_down_stream"], 40)