-#!/usr/bin/env python
-
# Copyright (c) 2016-2017 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# limitations under the License.
#
-from __future__ import absolute_import
-
import errno
import os
import unittest
'packets_in',
'packets_fwd',
'packets_dropped',
+ 'curr_packets_fwd',
+ 'curr_packets_in'
],
},
'connection-point': [
}
@mock.patch(SSH_HELPER)
- def test___init__(self, ssh, mock_time):
+ def test___init__(self, ssh, *args):
mock_ssh(ssh)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
self.assertIsNone(prox_approx_vnf._vnf_process)
@mock.patch(SSH_HELPER)
- def test_collect_kpi_no_client(self, ssh, mock_time):
+ def test_collect_kpi_no_client(self, ssh, *args):
mock_ssh(ssh)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
'packets_in': 0,
'packets_dropped': 0,
'packets_fwd': 0,
- 'collect_stats': {'core': {}},
+ 'collect_stats': {'core': {}}
}
result = prox_approx_vnf.collect_kpi()
self.assertEqual(result, expected)
@mock.patch(SSH_HELPER)
- def test_collect_kpi(self, ssh, mock_time):
+ def test_collect_kpi(self, ssh, *args):
mock_ssh(ssh)
resource_helper = mock.MagicMock()
'collect_stats': {'core': {'result': 234}},
}
result = prox_approx_vnf.collect_kpi()
- self.assertEqual(result, expected)
+ self.assertEqual(result['packets_in'], expected['packets_in'])
+ self.assertEqual(result['packets_dropped'], expected['packets_dropped'])
+ self.assertEqual(result['packets_fwd'], expected['packets_fwd'])
+ self.assertNotEqual(result['packets_fwd'], 0)
+ self.assertNotEqual(result['packets_fwd'], 0)
@mock.patch(SSH_HELPER)
- def test_collect_kpi_error(self, ssh, mock_time):
+ def test_collect_kpi_error(self, ssh, *args):
mock_ssh(ssh)
resource_helper = mock.MagicMock()
with self.assertRaises(RuntimeError):
prox_approx_vnf.collect_kpi()
- def _get_file_abspath(self, filename, mock_time):
+ def _get_file_abspath(self, filename, *args):
curr_path = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(curr_path, filename)
return file_path
+ @mock.patch('yardstick.common.utils.open', create=True)
@mock.patch('yardstick.benchmark.scenarios.networking.vnf_generic.open', create=True)
@mock.patch('yardstick.network_services.helpers.iniparser.open', create=True)
@mock.patch(SSH_HELPER)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf.scenario_helper.scenario_cfg = self.SCENARIO_CFG
- prox_approx_vnf.ssh_helper.provision_tool.return_value = '/tool_path12/tool_file34'
+ prox_approx_vnf.ssh_helper.join_bin_path.return_value = '/tool_path12/tool_file34'
prox_approx_vnf.setup_helper.remote_path = 'configs/file56.cfg'
expected = "sudo bash -c 'cd /tool_path12; " \
self.assertEqual(result, expected)
@mock.patch(SSH_HELPER)
- def bad_test_instantiate(self, ssh, mock_time):
+ def bad_test_instantiate(self, *args):
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf.scenario_helper = mock.MagicMock()
prox_approx_vnf.setup_helper = mock.MagicMock()
prox_approx_vnf.setup_helper.build_config.assert_called_once()
@mock.patch(SSH_HELPER)
- def test_wait_for_instantiate_panic(self, ssh, mock_time):
+ def test_wait_for_instantiate_panic(self, ssh, *args):
mock_ssh(ssh, exec_result=(1, "", ""))
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf._vnf_process = mock.MagicMock(**{"is_alive.return_value": True})
with self.assertRaises(RuntimeError):
prox_approx_vnf.wait_for_instantiate()
- @mock.patch(SSH_HELPER)
- def test_scale(self, ssh, mock_time):
- mock_ssh(ssh)
- prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
- with self.assertRaises(NotImplementedError):
- prox_approx_vnf.scale()
-
@mock.patch('yardstick.network_services.vnf_generic.vnf.prox_helpers.socket')
@mock.patch(SSH_HELPER)
- def test_terminate(self, ssh, mock_socket, mock_time):
+ def test_terminate(self, ssh, *args):
mock_ssh(ssh)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf._vnf_process = mock.MagicMock()
self.assertIsNone(prox_approx_vnf.terminate())
@mock.patch(SSH_HELPER)
- def test__vnf_up_post(self, ssh, mock_time):
+ def test__vnf_up_post(self, ssh, *args):
mock_ssh(ssh)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf.resource_helper = resource_helper = mock.Mock()
self.assertEqual(resource_helper.up_post.call_count, 1)
@mock.patch(SSH_HELPER)
- def test_vnf_execute_oserror(self, ssh, mock_time):
+ def test_vnf_execute_oserror(self, ssh, *args):
mock_ssh(ssh)
prox_approx_vnf = ProxApproxVnf(NAME, self.VNFD0)
prox_approx_vnf.resource_helper = resource_helper = mock.Mock()
resource_helper.execute.side_effect = OSError(errno.EADDRINUSE, "")
with self.assertRaises(OSError):
prox_approx_vnf.vnf_execute("", _ignore_errors=True)
-
-if __name__ == '__main__':
- unittest.main()