# Unittest for yardstick.benchmark.scenarios.networking.test_vnf_generic
from __future__ import absolute_import
+
+import os
+import errno
import unittest
import mock
-import os
from yardstick.benchmark.scenarios.networking.vnf_generic import \
- ssh_manager, NetworkServiceTestCase, IncorrectConfig, IncorrectSetup
+ SshManager, NetworkServiceTestCase, IncorrectConfig, \
+ IncorrectSetup, open_relative_file
from yardstick.network_services.collector.subscriber import Collector
from yardstick.network_services.vnf_generic.vnf.base import \
GenericTrafficGen, GenericVNF
+STL_MOCKS = {
+ 'stl': mock.MagicMock(),
+ 'stl.trex_stl_lib': mock.MagicMock(),
+ 'stl.trex_stl_lib.base64': mock.MagicMock(),
+ 'stl.trex_stl_lib.binascii': mock.MagicMock(),
+ 'stl.trex_stl_lib.collections': mock.MagicMock(),
+ 'stl.trex_stl_lib.copy': mock.MagicMock(),
+ 'stl.trex_stl_lib.datetime': mock.MagicMock(),
+ 'stl.trex_stl_lib.functools': mock.MagicMock(),
+ 'stl.trex_stl_lib.imp': mock.MagicMock(),
+ 'stl.trex_stl_lib.inspect': mock.MagicMock(),
+ 'stl.trex_stl_lib.json': mock.MagicMock(),
+ 'stl.trex_stl_lib.linecache': mock.MagicMock(),
+ 'stl.trex_stl_lib.math': mock.MagicMock(),
+ 'stl.trex_stl_lib.os': mock.MagicMock(),
+ 'stl.trex_stl_lib.platform': mock.MagicMock(),
+ 'stl.trex_stl_lib.pprint': mock.MagicMock(),
+ 'stl.trex_stl_lib.random': mock.MagicMock(),
+ 'stl.trex_stl_lib.re': mock.MagicMock(),
+ 'stl.trex_stl_lib.scapy': mock.MagicMock(),
+ 'stl.trex_stl_lib.socket': mock.MagicMock(),
+ 'stl.trex_stl_lib.string': mock.MagicMock(),
+ 'stl.trex_stl_lib.struct': mock.MagicMock(),
+ 'stl.trex_stl_lib.sys': mock.MagicMock(),
+ 'stl.trex_stl_lib.threading': mock.MagicMock(),
+ 'stl.trex_stl_lib.time': mock.MagicMock(),
+ 'stl.trex_stl_lib.traceback': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_async_client': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_client': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_exceptions': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_ext': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_jsonrpc_client': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_packet_builder_interface': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_packet_builder_scapy': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_port': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_stats': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_streams': mock.MagicMock(),
+ 'stl.trex_stl_lib.trex_stl_types': mock.MagicMock(),
+ 'stl.trex_stl_lib.types': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.argparse': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.collections': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.common': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.json': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.os': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.parsing_opts': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.pwd': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.random': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.re': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.string': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.sys': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.text_opts': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.text_tables': mock.MagicMock(),
+ 'stl.trex_stl_lib.utils.texttable': mock.MagicMock(),
+ 'stl.trex_stl_lib.warnings': mock.MagicMock(),
+ 'stl.trex_stl_lib.yaml': mock.MagicMock(),
+ 'stl.trex_stl_lib.zlib': mock.MagicMock(),
+ 'stl.trex_stl_lib.zmq': mock.MagicMock(),
+}
+
COMPLETE_TREX_VNFD = \
{'vnfd:vnfd-catalog':
{'vnfd':
"""
TRAFFIC_PROFILE = {
- "schema": "isb:traffic_profile:0.1",
- "name": "fixed",
- "description": "Fixed traffic profile to run UDP traffic",
- "traffic_profile": {
- "traffic_type": "FixedTraffic",
- "frame_rate": 100, # pps
- "flow_number": 10,
- "frame_size": 64}}
+ "schema": "isb:traffic_profile:0.1",
+ "name": "fixed",
+ "description": "Fixed traffic profile to run UDP traffic",
+ "traffic_profile": {
+ "traffic_type": "FixedTraffic",
+ "frame_rate": 100, # pps
+ "flow_number": 10,
+ "frame_size": 64}}
class TestNetworkServiceTestCase(unittest.TestCase):
'if': 'xe1'}],
'password': 'r00t'}}}
- self.topology = \
- {'short-name': 'trex-tg-topology',
- 'constituent-vnfd':
- [{'member-vnf-index': '1',
- 'VNF model': 'tg_trex_tpl.yaml',
- 'vnfd-id-ref': 'trexgen__1'},
- {'member-vnf-index': '2',
- 'VNF model': 'tg_trex_tpl.yaml',
- 'vnfd-id-ref': 'trexvnf__1'}],
- 'description': 'trex-tg-topology',
- 'name': 'trex-tg-topology',
- 'vld': [{'vnfd-connection-point-ref': [
- {'vnfd-connection-point-ref': 'xe0',
- 'member-vnf-index-ref': '1',
- 'vnfd-id-ref': 'trexgen'},
- {'vnfd-connection-point-ref': 'xe0',
- 'member-vnf-index-ref': '2',
- 'vnfd-id-ref': 'trexgen'}],
- 'type': 'ELAN',
- 'id': 'private',
- 'name': 'trexgen__1 to trexvnf__1 link 1'},
- {'vnfd-connection-point-ref': [
- {'vnfd-connection-point-ref': 'xe1',
- 'member-vnf-index-ref': '1',
- 'vnfd-id-ref': 'trexgen'},
- {'vnfd-connection-point-ref': 'xe1',
- 'member-vnf-index-ref': '2',
- 'vnfd-id-ref': 'trexgen'}],
- 'type': 'ELAN',
- 'id': 'public',
- 'name': 'trexvnf__1 to trexgen__1 link 2'}],
- 'id': 'trex-tg-topology'}
-
- self.scenario_cfg = {'tc_options':
- {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
- 'task_id': 'a70bdf4a-8e67-47a3-9dc1-273c14506eb7',
- 'tc': 'tc_ipv4_1Mflow_64B_packetsize',
- 'runner': {'object': 'NetworkServiceTestCase',
- 'interval': 35,
- 'output_filename': 'yardstick.out',
- 'runner_id': 74476,
- 'duration': 400, 'type': 'Duration'},
- 'traffic_profile': 'ipv4_throughput_vpe.yaml',
- 'traffic_options':
- {'flow': 'ipv4_1flow_Packets_vpe.yaml',
- 'imix': 'imix_voice.yaml'},
- 'type': 'ISB',
- 'nodes': {'tg__2': 'trafficgen_2.yardstick',
- 'tg__1': 'trafficgen_1.yardstick',
- 'vnf__1': 'vnf.yardstick'},
- 'topology': 'vpe_vnf_topology.yaml'}
-
- self.scenario_cfg["topology"] = \
- self._get_file_abspath("vpe_vnf_topology.yaml")
+ self.topology = {
+ 'short-name': 'trex-tg-topology',
+ 'constituent-vnfd':
+ [{'member-vnf-index': '1',
+ 'VNF model': 'tg_trex_tpl.yaml',
+ 'vnfd-id-ref': 'trexgen__1'},
+ {'member-vnf-index': '2',
+ 'VNF model': 'tg_trex_tpl.yaml',
+ 'vnfd-id-ref': 'trexvnf__1'}],
+ 'description': 'trex-tg-topology',
+ 'name': 'trex-tg-topology',
+ 'vld': [
+ {
+ 'vnfd-connection-point-ref': [
+ {
+ 'vnfd-connection-point-ref': 'xe0',
+ 'member-vnf-index-ref': '1',
+ 'vnfd-id-ref': 'trexgen'
+ },
+ {
+ 'vnfd-connection-point-ref': 'xe0',
+ 'member-vnf-index-ref': '2',
+ 'vnfd-id-ref': 'trexgen'
+ }
+ ],
+ 'type': 'ELAN',
+ 'id': 'private',
+ 'name': 'trexgen__1 to trexvnf__1 link 1'
+ },
+ {
+ 'vnfd-connection-point-ref': [
+ {
+ 'vnfd-connection-point-ref': 'xe1',
+ 'member-vnf-index-ref': '1',
+ 'vnfd-id-ref': 'trexgen'
+ },
+ {
+ 'vnfd-connection-point-ref': 'xe1',
+ 'member-vnf-index-ref': '2',
+ 'vnfd-id-ref': 'trexgen'
+ }
+ ],
+ 'type': 'ELAN',
+ 'id': 'public',
+ 'name': 'trexvnf__1 to trexgen__1 link 2'
+ }],
+ 'id': 'trex-tg-topology',
+ }
+
+ self.scenario_cfg = {
+ 'task_path': "",
+ 'tc_options': {'rfc2544': {'allowed_drop_rate': '0.8 - 1'}},
+ 'task_id': 'a70bdf4a-8e67-47a3-9dc1-273c14506eb7',
+ 'tc': 'tc_ipv4_1Mflow_64B_packetsize',
+ 'runner': {'object': 'NetworkServiceTestCase',
+ 'interval': 35,
+ 'output_filename': 'yardstick.out',
+ 'runner_id': 74476,
+ 'duration': 400, 'type': 'Duration'},
+ 'traffic_profile': 'ipv4_throughput_vpe.yaml',
+ 'traffic_options': {'flow': 'ipv4_1flow_Packets_vpe.yaml',
+ 'imix': 'imix_voice.yaml'}, 'type': 'ISB',
+ 'nodes': {'tg__2': 'trafficgen_2.yardstick',
+ 'tg__1': 'trafficgen_1.yardstick',
+ 'vnf__1': 'vnf.yardstick'},
+ "topology": self._get_file_abspath("vpe_vnf_topology.yaml")}
+
self.s = NetworkServiceTestCase(self.scenario_cfg, self.context_cfg)
def _get_file_abspath(self, filename):
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(0, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
for node, node_dict in self.context_cfg["nodes"].items():
- with ssh_manager(node_dict) as conn:
+ with SshManager(node_dict) as conn:
self.assertIsNotNone(conn)
def test___init__(self):
def test_get_vnf_imp(self):
vnfd = COMPLETE_TREX_VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- self.assertIsNotNone(self.s.get_vnf_impl(vnfd))
+ with mock.patch.dict("sys.modules", STL_MOCKS):
+ self.assertIsNotNone(self.s.get_vnf_impl(vnfd))
def test_load_vnf_models_invalid(self):
self.context_cfg["nodes"]['trexgen__1']['VNF model'] = \
vnf = mock.Mock(autospec=GenericVNF)
self.s.get_vnf_impl = mock.Mock(return_value=vnf)
- self.assertIsNotNone(self.s.load_vnf_models(self.context_cfg))
+ self.assertIsNotNone(
+ self.s.load_vnf_models(self.scenario_cfg, self.context_cfg))
def test_map_topology_to_infrastructure(self):
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(0, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
self.s.map_topology_to_infrastructure(self.context_cfg,
self.topology)
self.assertEqual("tg_trex_tpl.yaml",
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(1, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(1, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
self.assertRaises(IncorrectSetup,
self.s.map_topology_to_infrastructure,
self.context_cfg, self.topology)
def test_map_topology_to_infrastructure_config_invalid(self):
- del self.context_cfg\
- ['nodes']['trexvnf__1']['interfaces']['xe0']['local_mac']
+ cfg = dict(self.context_cfg)
+ del cfg['nodes']['trexvnf__1']['interfaces']['xe0']['local_mac']
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(0, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
self.assertRaises(IncorrectConfig,
self.s.map_topology_to_infrastructure,
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(0, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
del self.context_cfg['nodes']
self.assertRaises(IncorrectConfig, self.s._resolve_topology,
with mock.patch("yardstick.ssh.SSH") as ssh:
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
- mock.Mock(return_value=(0, SYS_CLASS_NET+IP_ADDR_SHOW, ""))
- ssh.return_value = ssh_mock
+ mock.Mock(return_value=(0, SYS_CLASS_NET + IP_ADDR_SHOW, ""))
+ ssh.from_node.return_value = ssh_mock
tgen = mock.Mock(autospec=GenericTrafficGen)
tgen.traffic_finished = True
self.context_cfg))
def test__get_traffic_profile_exception(self):
- self.assertRaises(IOError, self.s._get_traffic_profile,
- self.scenario_cfg, self.context_cfg)
+ cfg = dict(self.scenario_cfg)
+ cfg["traffic_profile"] = ""
+ self.assertRaises(IOError, self.s._get_traffic_profile, cfg,
+ self.context_cfg)
def test___get_traffic_imix_exception(self):
- self.assertEqual({}, self.s._get_traffic_imix(self.scenario_cfg))
+ cfg = dict(self.scenario_cfg)
+ cfg["traffic_options"]["imix"] = ""
+ self.assertEqual({}, self.s._get_traffic_imix(cfg))
def test__fill_traffic_profile(self):
- self.scenario_cfg["traffic_profile"] = \
- self._get_file_abspath("ipv4_throughput_vpe.yaml")
- self.scenario_cfg["traffic_options"]["flow"] = \
- self._get_file_abspath("ipv4_1flow_Packets_vpe.yaml")
- self.scenario_cfg["traffic_options"]["imix"] = \
- self._get_file_abspath("imix_voice.yaml")
- self.assertIsNotNone(self.s._fill_traffic_profile(self.scenario_cfg,
- self.context_cfg))
+ with mock.patch.dict("sys.modules", STL_MOCKS):
+ self.scenario_cfg["traffic_profile"] = \
+ self._get_file_abspath("ipv4_throughput_vpe.yaml")
+ self.scenario_cfg["traffic_options"]["flow"] = \
+ self._get_file_abspath("ipv4_1flow_Packets_vpe.yaml")
+ self.scenario_cfg["traffic_options"]["imix"] = \
+ self._get_file_abspath("imix_voice.yaml")
+ self.assertIsNotNone(self.s._fill_traffic_profile(self.scenario_cfg,
+ self.context_cfg))
def test_teardown(self):
vnf = mock.Mock(autospec=GenericVNF)
self.s.collector.stop = \
mock.Mock(return_value=True)
self.assertIsNone(self.s.teardown())
+
+ SAMPLE_NETDEVS = {
+ 'enp11s0': {
+ 'address': '0a:de:ad:be:ef:f5',
+ 'device': '0x1533',
+ 'driver': 'igb',
+ 'ifindex': '2',
+ 'interface_name': 'enp11s0',
+ 'operstate': 'down',
+ 'pci_bus_id': '0000:0b:00.0',
+ 'subsystem_device': '0x1533',
+ 'subsystem_vendor': '0x15d9',
+ 'vendor': '0x8086'
+ },
+ 'lan': {
+ 'address': '0a:de:ad:be:ef:f4',
+ 'device': '0x153a',
+ 'driver': 'e1000e',
+ 'ifindex': '3',
+ 'interface_name': 'lan',
+ 'operstate': 'up',
+ 'pci_bus_id': '0000:00:19.0',
+ 'subsystem_device': '0x153a',
+ 'subsystem_vendor': '0x15d9',
+ 'vendor': '0x8086'
+ }
+ }
+ SAMPLE_VM_NETDEVS = {
+ 'eth1': {
+ 'address': 'fa:de:ad:be:ef:5b',
+ 'device': '0x0001',
+ 'driver': 'virtio_net',
+ 'ifindex': '3',
+ 'interface_name': 'eth1',
+ 'operstate': 'down',
+ 'pci_bus_id': '0000:00:04.0',
+ 'vendor': '0x1af4'
+ }
+ }
+
+ def test_parse_netdev_info(self):
+ output = """\
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/ifindex:2
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/address:0a:de:ad:be:ef:f5
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/operstate:down
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/vendor:0x8086
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/device:0x1533
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_vendor:0x15d9
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/device/subsystem_device:0x1533
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/driver:igb
+/sys/devices/pci0000:00/0000:00:1c.3/0000:0b:00.0/net/enp11s0/pci_bus_id:0000:0b:00.0
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/ifindex:3
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/address:0a:de:ad:be:ef:f4
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/operstate:up
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/vendor:0x8086
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/device:0x153a
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_vendor:0x15d9
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/device/subsystem_device:0x153a
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/driver:e1000e
+/sys/devices/pci0000:00/0000:00:19.0/net/lan/pci_bus_id:0000:00:19.0
+"""
+ res = NetworkServiceTestCase.parse_netdev_info(output)
+ assert res == self.SAMPLE_NETDEVS
+
+ def test_parse_netdev_info_virtio(self):
+ output = """\
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/ifindex:3
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/address:fa:de:ad:be:ef:5b
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/operstate:down
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/vendor:0x1af4
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/device/device:0x0001
+/sys/devices/pci0000:00/0000:00:04.0/virtio1/net/eth1/driver:virtio_net
+"""
+ res = NetworkServiceTestCase.parse_netdev_info(output)
+ assert res == self.SAMPLE_VM_NETDEVS
+
+ def test_sort_dpdk_port_num(self):
+ netdevs = self.SAMPLE_NETDEVS.copy()
+ NetworkServiceTestCase._sort_dpdk_port_num(netdevs)
+ assert netdevs['lan']['dpdk_port_num'] == 1
+ assert netdevs['enp11s0']['dpdk_port_num'] == 2
+
+ def test_probe_missing_values(self):
+ netdevs = self.SAMPLE_NETDEVS.copy()
+ NetworkServiceTestCase._sort_dpdk_port_num(netdevs)
+ network = {'local_mac': '0a:de:ad:be:ef:f5'}
+ NetworkServiceTestCase._probe_missing_values(netdevs, network, set())
+ assert network['dpdk_port_num'] == 2
+
+ network = {'local_mac': '0a:de:ad:be:ef:f4'}
+ NetworkServiceTestCase._probe_missing_values(netdevs, network, set())
+ assert network['dpdk_port_num'] == 1
+
+ def test_open_relative_path(self):
+ mock_open = mock.mock_open()
+ mock_open_result = mock_open()
+ mock_open_call_count = 1 # initial call to get result
+
+ module_name = \
+ 'yardstick.benchmark.scenarios.networking.vnf_generic.open'
+
+ # test
+ with mock.patch(module_name, mock_open, create=True):
+ self.assertEqual(open_relative_file('foo', 'bar'), mock_open_result)
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertNotIn('bar', mock_open.call_args_list[-1][0][0])
+
+ def open_effect(*args, **kwargs):
+ if kwargs.get('name', args[0]) == os.path.join('bar', 'foo'):
+ return mock_open_result
+ raise IOError(errno.ENOENT, 'not found')
+
+ mock_open.side_effect = open_effect
+ self.assertEqual(open_relative_file('foo', 'bar'), mock_open_result)
+
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError of type ENOENT
+ mock_open.side_effect = IOError(errno.ENOENT, 'not found')
+ with self.assertRaises(IOError):
+ # the second call still raises
+ open_relative_file('foo', 'bar')
+
+ mock_open_call_count += 2 # two more calls expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
+ self.assertIn('foo', mock_open.call_args_list[-1][0][0])
+ self.assertIn('bar', mock_open.call_args_list[-1][0][0])
+
+ # test an IOError other than ENOENT
+ mock_open.side_effect = IOError(errno.EBUSY, 'busy')
+ with self.assertRaises(IOError):
+ open_relative_file('foo', 'bar')
+
+ mock_open_call_count += 1 # one more call expected
+ self.assertEqual(mock_open.call_count, mock_open_call_count)
Code Review / yardstick.git / blobdiff