1 # Copyright (c) 2016-2017 Intel Corporation
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
15 from __future__ import absolute_import
17 import multiprocessing
20 from yardstick.network_services.nfvi.resource import ResourceProfile
21 from yardstick.network_services.nfvi import resource, collectd
24 class TestResourceProfile(unittest.TestCase):
25 VNFD = {'vnfd:vnfd-catalog':
27 [{'short-name': 'VpeVnf',
30 [{'network': '172.16.100.20',
31 'netmask': '255.255.255.0',
32 'gateway': '172.16.100.20',
34 {'network': '172.16.40.20',
35 'netmask': '255.255.255.0',
36 'gateway': '172.16.40.20',
38 'description': 'VPE approximation using DPDK',
39 'name': 'vpevnf-baremetal',
41 [{'network': '0064:ff9b:0:0:0:0:9810:6414',
43 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
45 {'network': '0064:ff9b:0:0:0:0:9810:2814',
47 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
49 'id': 'vpevnf-baremetal',
51 [{'virtual-interface':
52 {'dst_mac': '3c:fd:fe:9e:64:38',
53 'vpci': '0000:05:00.0',
54 'local_ip': '172.16.100.19',
55 'type': 'PCI-PASSTHROUGH',
56 'netmask': '255.255.255.0',
58 'bandwidth': '10 Gbps',
59 'dst_ip': '172.16.100.20',
60 'local_mac': '3c:fd:fe:a1:2b:80'},
61 'vnfd-connection-point-ref': 'xe0',
64 {'dst_mac': '00:1e:67:d0:60:5c',
65 'vpci': '0000:05:00.1',
66 'local_ip': '172.16.40.19',
67 'type': 'PCI-PASSTHROUGH',
68 'netmask': '255.255.255.0',
70 'bandwidth': '10 Gbps',
71 'dst_ip': '172.16.40.20',
72 'local_mac': '3c:fd:fe:a1:2b:81'},
73 'vnfd-connection-point-ref': 'xe1',
75 'description': 'Vpe approximation using DPDK',
77 {'vdu-id': 'vpevnf-baremetal',
83 {'kpi': ['packets_in', 'packets_fwd', 'packets_dropped']},
84 'connection-point': [{'type': 'VPORT', 'name': 'xe0'},
85 {'type': 'VPORT', 'name': 'xe1'}],
86 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'}]}}
89 with mock.patch("yardstick.ssh.SSH") as ssh:
90 ssh_mock = mock.Mock(autospec=ssh.SSH)
92 mock.Mock(return_value=(0, {}, ""))
93 ssh.return_value = ssh_mock
95 self.resource_profile = \
96 ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
99 def test___init__(self):
100 self.assertEqual(True, self.resource_profile.enable)
102 def test_check_if_sa_running(self):
103 self.assertEqual(self.resource_profile.check_if_sa_running("collectd"),
106 def test_get_cpu_data(self):
107 reskey = ["", "cpufreq", "cpufreq-0"]
109 val = self.resource_profile.get_cpu_data(reskey, value)
110 self.assertEqual(val, ['0', 'cpufreq', '10', 'metric'])
112 def test_get_cpu_data_error(self):
113 reskey = ["", "", ""]
115 val = self.resource_profile.get_cpu_data(reskey, value)
116 self.assertEqual(val, ['error', 'Invalid', ''])
118 def test__start_collectd(self):
119 with mock.patch("yardstick.ssh.SSH") as ssh:
120 ssh_mock = mock.Mock(autospec=ssh.SSH)
122 mock.Mock(return_value=(0, "", ""))
123 ssh.return_value = ssh_mock
125 ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
128 resource_profile._start_collectd(ssh_mock, "/opt/nsb_bin"))
130 def test_initiate_systemagent(self):
131 with mock.patch("yardstick.ssh.SSH") as ssh:
132 ssh_mock = mock.Mock(autospec=ssh.SSH)
134 mock.Mock(return_value=(0, "", ""))
135 ssh.return_value = ssh_mock
137 ResourceProfile(self.VNFD['vnfd:vnfd-catalog']['vnfd'][0],
140 resource_profile.initiate_systemagent("/opt/nsb_bin"))
142 def test_parse_collectd_result(self):
143 res = self.resource_profile.parse_collectd_result({}, [0, 1, 2])
144 self.assertDictEqual(res, {'timestamp': '', 'cpu': {}, 'memory': {}})
146 def test_run_collectd_amqp(self):
147 _queue = multiprocessing.Queue()
148 resource.AmqpConsumer = mock.Mock(autospec=collectd)
149 self.assertIsNone(self.resource_profile.run_collectd_amqp(_queue))
151 def test_start(self):
152 self.assertIsNone(self.resource_profile.start())
155 self.assertIsNone(self.resource_profile.stop())
157 if __name__ == '__main__':