- def test_get_drop_percentage_div_zero(self):
- traffic_generator = mock.Mock(autospec=TrexProfile)
- traffic_generator.networks = {
- "uplink_0": ["xe0"],
- "downlink_0": ["xe1"],
- }
- traffic_generator.client = \
- mock.Mock(return_value=True)
- r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
- r_f_c2544_profile.params = self.PROFILE
- self.assertIsNone(
- r_f_c2544_profile.execute_traffic(traffic_generator))
- samples = {}
- for ifname in range(1):
- name = "xe{}".format(ifname)
- samples[name] = {"rx_throughput_fps": 20,
- "tx_throughput_fps": 20,
- "rx_throughput_mbps": 10,
- "tx_throughput_mbps": 10,
- "in_packets": 1000,
- "out_packets": 0}
- r_f_c2544_profile.throughput_max = 0
- expected = {
- 'DropPercentage': 100.0, 'RxThroughput': 100 / 3.0,
- 'TxThroughput': 0.0, 'CurrentDropPercentage': 100.0,
- 'Throughput': 66.66666666666667,
- 'xe0': {
- 'tx_throughput_fps': 20, 'in_packets': 1000,
- 'out_packets': 0, 'rx_throughput_mbps': 10,
- 'tx_throughput_mbps': 10, 'rx_throughput_fps': 20
- }
- }
- traffic_generator.generate_samples = mock.Mock(return_value=samples)
- traffic_generator.RUN_DURATION = 30
- traffic_generator.rfc2544_helper.tolerance_low = 0.0001
- traffic_generator.rfc2544_helper.tolerance_high = 0.0001
- self.assertDictEqual(expected,
- r_f_c2544_profile.get_drop_percentage(traffic_generator))
+ @mock.patch.object(trex_stl_streams, 'STLFlowLatencyStats')
+ @mock.patch.object(trex_stl_streams, 'STLTXCont')
+ @mock.patch.object(trex_stl_client, 'STLStream')
+ def test__create_streams(self, mock_stream, mock_txcont, mock_latency):
+ imix_data = {'64': 25, '512': 75}
+ rate = 35
+ port_pg_id = rfc2544.PortPgIDMap()
+ port_pg_id.add_port(10)
+ mock_stream.side_effect = ['stream1', 'stream2']
+ mock_txcont.side_effect = ['txcont1', 'txcont2']
+ mock_latency.side_effect = ['latency1', 'latency2']
+ rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
+ with mock.patch.object(rfc2544_profile, '_create_single_packet'):
+ output = rfc2544_profile._create_streams(imix_data, rate,
+ port_pg_id, True)
+ self.assertEqual(['stream1', 'stream2'], output)
+ mock_latency.assert_has_calls([
+ mock.call(pg_id=1), mock.call(pg_id=2)])
+ mock_txcont.assert_has_calls([
+ mock.call(percentage=float(25 * 35) / 100),
+ mock.call(percentage=float(75 * 35) / 100)], any_order=True)
+
+ def test_get_drop_percentage(self):
+ rfc2544_profile = rfc2544.RFC2544Profile(self.TRAFFIC_PROFILE)
+ samples = [
+ {'xe1': {'tx_throughput_fps': 110,
+ 'rx_throughput_fps': 101,
+ 'out_packets': 2100,
+ 'in_packets': 2010,
+ 'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 1)},
+ 'xe2': {'tx_throughput_fps': 210,
+ 'rx_throughput_fps': 201,
+ 'out_packets': 4100,
+ 'in_packets': 4010,
+ 'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 1)}},
+ {'xe1': {'tx_throughput_fps': 156,
+ 'rx_throughput_fps': 108,
+ 'out_packets': 2110,
+ 'in_packets': 2040,
+ 'latency': 'Latency1',
+ 'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 31)},
+ 'xe2': {'tx_throughput_fps': 253,
+ 'rx_throughput_fps': 215,
+ 'out_packets': 4150,
+ 'in_packets': 4010,
+ 'latency': 'Latency2',
+ 'timestamp': datetime.datetime(2000, 1, 1, 1, 1, 1, 31)}}
+ ]
+ completed, output = rfc2544_profile.get_drop_percentage(
+ samples, 0, 0, False)
+ expected = {'DropPercentage': 50.0,
+ 'Latency': {'xe1': 'Latency1', 'xe2': 'Latency2'},
+ 'RxThroughput': 1000000.0,
+ 'TxThroughput': 2000000.0,
+ 'CurrentDropPercentage': 50.0,
+ 'Rate': 100.0,
+ 'Throughput': 1000000.0}
+ self.assertEqual(expected, output)
+ self.assertFalse(completed)