ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
self.assertIsNotNone(trex_traffic_gen._terminated)
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
restult = trex_traffic_gen.collect_kpi()
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
self.assertEqual(None, trex_traffic_gen.listen_traffic({}))
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
trex_traffic_gen._start_server = mock.Mock(return_value=0)
scenario_cfg = {"tc": "tc_baremetal_rfc2544_ipv4_1flow_64B"}
- tg_rfc2544_trex.WAIT_TIME = 3
- self.assertEqual(0, trex_traffic_gen.instantiate(scenario_cfg, {}))
+ tg_rfc2544_trex.WAIT_TIME = 0
+ self.assertIn(trex_traffic_gen.instantiate(scenario_cfg, {}), {0, None})
def test_instantiate_error(self):
mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
mock.Mock(return_value=(1, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
scenario_cfg = {"tc": "tc_baremetal_rfc2544_ipv4_1flow_64B"}
+ tg_rfc2544_trex.WAIT_TIME = 0
self.assertRaises(RuntimeError,
trex_traffic_gen.instantiate, scenario_cfg, {})
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
self.assertEqual([0.8, 1.0],
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
self.assertEqual(None, trex_traffic_gen._start_server())
file_path = os.path.join(curr_path, filename)
return file_path
- def test__traffic_runner(self):
+ @mock.patch("yardstick.network_services.vnf_generic.vnf.tg_rfc2544_trex.time")
+ def test__traffic_runner(self, mock_time):
mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.execute.return_value = "64"
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
self.sut = TrexTrafficGenRFC(vnfd)
self.sut.connection = mock.Mock()
self._get_file_abspath(
"tc_baremetal_rfc2544_ipv4_1flow_64B.yaml")
tg_rfc2544_trex.DURATION = 1
- tg_rfc2544_trex.WAIT_TIME = 1
+ tg_rfc2544_trex.WAIT_TIME = 0
self.sut._traffic_runner(mock_traffic_profile, q, client_started,
self.sut._terminated)
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
result = ['0x00', '0x00', '0x00', '0x00', '0x00', '0x01']
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
- tg_rfc2544_trex.WAIT_TIME = 1
+ tg_rfc2544_trex.WAIT_TIME = 0
self.assertEqual(None, trex_traffic_gen._generate_trex_cfg(vnfd))
def test_run_traffic(self):
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
self.sut = TrexTrafficGenRFC(vnfd)
self.sut.connection = mock.Mock()
mock.Mock(return_value=(0, "", ""))
ssh_mock.run = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
flavor = ""
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
self.assertEqual(None, trex_traffic_gen.terminate())
ssh_mock = mock.Mock(autospec=ssh.SSH)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
- ssh.return_value = ssh_mock
+ ssh.from_node.return_value = ssh_mock
trex_traffic_gen = TrexTrafficGenRFC(vnfd)
client = mock.Mock(autospec=STLClient)
client.connect = mock.Mock(return_value=0)