3 # Copyright (c) 2016-2017 Intel Corporation
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
18 # Unittest for yardstick.network_services.vnf_generic.vnf.test_base
20 from __future__ import absolute_import
24 from multiprocessing import Queue
26 from yardstick.network_services.vnf_generic.vnf.base import \
27 QueueFileWrapper, GenericVNF, GenericTrafficGen
28 from yardstick.ssh import SSH
30 IP_PIPELINE_CFG_FILE_TPL = """
31 arp_route_tbl = ({port0_local_ip_hex},{port0_netmask_hex},1,"""
32 """{port1_local_ip_hex}) ({port1_local_ip_hex},{port1_netmask_hex},0,"""
33 """{port0_local_ip_hex})"""
35 IP_PIPELINE_ND_CFG_FILE_TPL = """
36 nd_route_tbl = ({port1_dst_ip_hex6},"""
37 """{port1_dst_netmask_hex6},1,{port1_dst_ip_hex6})"""
39 _LOCAL_OBJECT = object()
42 class FileAbsPath(object):
43 def __init__(self, module_file):
44 super(FileAbsPath, self).__init__()
45 self.module_path = os.path.dirname(os.path.abspath(module_file))
47 def get_path(self, filename):
48 file_path = os.path.join(self.module_path, filename)
52 def mock_ssh(mock_ssh_type, spec=None, exec_result=_LOCAL_OBJECT, run_result=_LOCAL_OBJECT):
56 if exec_result is _LOCAL_OBJECT:
57 exec_result = 0, "", ""
59 if run_result is _LOCAL_OBJECT:
60 run_result = 0, "", ""
62 mock_ssh_instance = mock.Mock(autospec=spec)
63 mock_ssh_instance._get_client.return_value = mock.Mock()
64 mock_ssh_instance.execute.return_value = exec_result
65 mock_ssh_instance.run.return_value = run_result
66 mock_ssh_type.from_node.return_value = mock_ssh_instance
67 return mock_ssh_instance
70 class TestQueueFileWrapper(unittest.TestCase):
72 self.prompt = "pipeline>"
76 def test___init__(self):
77 queue_file_wrapper = \
78 QueueFileWrapper(self.q_in, self.q_out, self.prompt)
79 self.assertEqual(queue_file_wrapper.prompt, self.prompt)
82 queue_file_wrapper = \
83 QueueFileWrapper(self.q_in, self.q_out, self.prompt)
84 queue_file_wrapper.bufsize = 5
85 queue_file_wrapper.write("pipeline>")
86 queue_file_wrapper.close()
87 self.assertIsNone(queue_file_wrapper.clear())
88 self.assertIsNotNone(queue_file_wrapper.q_out.empty())
91 queue_file_wrapper = \
92 QueueFileWrapper(self.q_in, self.q_out, self.prompt)
93 self.assertEqual(None, queue_file_wrapper.close())
96 queue_file_wrapper = \
97 QueueFileWrapper(self.q_in, self.q_out, self.prompt)
98 queue_file_wrapper.q_in.put("pipeline>")
99 self.assertEqual("pipeline>", queue_file_wrapper.read(20))
101 def test_write(self):
102 queue_file_wrapper = \
103 QueueFileWrapper(self.q_in, self.q_out, self.prompt)
104 queue_file_wrapper.write("pipeline>")
105 self.assertIsNotNone(queue_file_wrapper.q_out.empty())
108 class TestGenericVNF(unittest.TestCase):
111 'short-name': 'VpeVnf',
116 'network': '152.16.100.20',
117 'netmask': '255.255.255.0',
118 'gateway': '152.16.100.20',
122 'network': '152.16.40.20',
123 'netmask': '255.255.255.0',
124 'gateway': '152.16.40.20',
128 'description': 'VPE approximation using DPDK',
129 'name': 'vpevnf-baremetal',
132 'network': '0064:ff9b:0:0:0:0:9810:6414',
134 'gateway': '0064:ff9b:0:0:0:0:9810:6414',
138 'network': '0064:ff9b:0:0:0:0:9810:2814',
140 'gateway': '0064:ff9b:0:0:0:0:9810:2814',
144 'id': 'vpevnf-baremetal',
145 'external-interface': [
147 'virtual-interface': {
148 'dst_mac': '00:00:00:00:00:03',
149 'vpci': '0000:05:00.0',
150 'local_ip': '152.16.100.19',
151 'type': 'PCI-PASSTHROUGH',
152 'netmask': '255.255.255.0',
154 'bandwidth': '10 Gbps',
155 'dst_ip': '152.16.100.20',
156 'local_mac': '00:00:00:00:00:01'
158 'vnfd-connection-point-ref': 'xe0',
162 'virtual-interface': {
163 'dst_mac': '00:00:00:00:00:04',
164 'vpci': '0000:05:00.1',
165 'local_ip': '152.16.40.19',
166 'type': 'PCI-PASSTHROUGH',
167 'netmask': '255.255.255.0',
169 'bandwidth': '10 Gbps',
170 'dst_ip': '152.16.40.20',
171 'local_mac': '00:00:00:00:00:02'
173 'vnfd-connection-point-ref': 'xe1',
179 'description': 'Vpe approximation using DPDK',
181 'vdu-id': 'vpevnf-baremetal',
194 'connection-point': [
204 'id': 'VpeApproxVnf', 'name': 'VPEVnfSsh'
208 'vnfd:vnfd-catalog': {
215 def test___init__(self):
216 generic_vnf = GenericVNF('vnf1', self.VNFD_0)
217 assert generic_vnf.kpi
219 def test_collect_kpi(self):
220 generic_vnf = GenericVNF('vnf1', self.VNFD_0)
221 self.assertRaises(NotImplementedError, generic_vnf.collect_kpi)
223 def test__get_kpi_definition(self):
224 vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
225 generic_vnf = GenericVNF('vnf1', vnfd)
226 kpi = generic_vnf._get_kpi_definition()
227 self.assertEqual(kpi, ['packets_in', 'packets_fwd', 'packets_dropped'])
229 def test_instantiate(self):
230 generic_vnf = GenericVNF('vnf1', self.VNFD['vnfd:vnfd-catalog']['vnfd'][0])
231 with self.assertRaises(NotImplementedError):
232 generic_vnf.instantiate({}, {})
234 def test_scale(self):
235 generic_vnf = GenericVNF('vnf1', self.VNFD['vnfd:vnfd-catalog']['vnfd'][0])
236 with self.assertRaises(NotImplementedError):
239 def test_terminate(self):
240 generic_vnf = GenericVNF('vnf1', self.VNFD['vnfd:vnfd-catalog']['vnfd'][0])
241 with self.assertRaises(NotImplementedError):
242 generic_vnf.terminate()
245 class TestGenericTrafficGen(unittest.TestCase):
246 def test___init__(self):
247 vnfd = TestGenericVNF.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
248 generic_traffic_gen = GenericTrafficGen('vnf1', vnfd)
249 assert generic_traffic_gen.name == "vnf1"
251 def test_listen_traffic(self):
252 vnfd = TestGenericVNF.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
253 generic_traffic_gen = GenericTrafficGen('vnf1', vnfd)
255 self.assertIsNone(generic_traffic_gen.listen_traffic(traffic_profile))
257 def test_run_traffic(self):
258 vnfd = TestGenericVNF.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
259 generic_traffic_gen = GenericTrafficGen('vnf1', vnfd)
261 self.assertRaises(NotImplementedError,
262 generic_traffic_gen.run_traffic, traffic_profile)
264 def test_terminate(self):
265 vnfd = TestGenericVNF.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
266 generic_traffic_gen = GenericTrafficGen('vnf1', vnfd)
267 self.assertRaises(NotImplementedError, generic_traffic_gen.terminate)
269 def test_verify_traffic(self):
270 vnfd = TestGenericVNF.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
271 generic_traffic_gen = GenericTrafficGen('vnf1', vnfd)
273 self.assertIsNone(generic_traffic_gen.verify_traffic(traffic_profile))