1 # Copyright 2016 Red Hat Inc & Xena Networks.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 # Rick Alongi, Red Hat Inc.
17 # Amit Supugade, Red Hat Inc.
18 # Dan Amzulescu, Xena Networks
19 # Christian Trautman, Red Hat Inc.
22 Xena Traffic Generator Model
30 from time import sleep
31 import xml.etree.ElementTree as ET
32 from collections import OrderedDict
34 import scapy.layers.inet as inet
37 from conf import settings
38 from core.results.results_constants import ResultsConstants
39 from tools.pkt_gen.trafficgen.trafficgenhelper import (
42 from tools.pkt_gen.trafficgen.trafficgen import ITrafficGenerator
45 from tools.pkt_gen.xena.xena_json import XenaJSON
46 from tools.pkt_gen.xena.XenaDriver import (
53 class Xena(ITrafficGenerator):
55 Xena Traffic generator wrapper class
57 _traffic_defaults = TRAFFIC_DEFAULTS.copy()
58 _logger = logging.getLogger(__name__)
70 def traffic_defaults(self):
71 """Default traffic values.
73 These can be expected to be constant across traffic generators,
74 so no setter is provided. Changes to the structure or contents
75 will likely break traffic generator implementations or tests
78 return self._traffic_defaults
81 def _create_throughput_result(root):
83 Create the results based off the output xml file from the Xena2544.exe
85 :param root: root dictionary from xml import
86 :return: Results Ordered dictionary based off ResultsConstants
88 # get the test type from the report file
89 test_type = root[0][1].get('TestType')
90 # set the version from the report file
91 settings.setValue('XENA_VERSION', root[0][0][1].get('GeneratedBy'))
93 if test_type == 'Throughput':
94 results = OrderedDict()
95 results[ResultsConstants.THROUGHPUT_RX_FPS] = int(
96 root[0][1][0][0].get('PortRxPps')) + int(
97 root[0][1][0][1].get('PortRxPps'))
98 results[ResultsConstants.THROUGHPUT_RX_MBPS] = (int(
99 root[0][1][0][0].get('PortRxBpsL1')) + int(
100 root[0][1][0][1].get('PortRxBpsL1')))/ 1000000
101 results[ResultsConstants.THROUGHPUT_RX_PERCENT] = (
102 100 - int(root[0][1][0].get('TotalLossRatioPcnt'))) * float(
103 root[0][1][0].get('TotalTxRatePcnt'))/100
104 results[ResultsConstants.TX_RATE_FPS] = root[0][1][0].get(
106 results[ResultsConstants.TX_RATE_MBPS] = float(
107 root[0][1][0].get('TotalTxRateBpsL1')) / 1000000
108 results[ResultsConstants.TX_RATE_PERCENT] = root[0][1][0].get(
111 results[ResultsConstants.MIN_LATENCY_NS] = float(
112 root[0][1][0][0].get('MinLatency')) * 1000
114 # Stats for latency returned as N/A so just post them
115 results[ResultsConstants.MIN_LATENCY_NS] = root[0][1][0][0].get(
118 results[ResultsConstants.MAX_LATENCY_NS] = float(
119 root[0][1][0][0].get('MaxLatency')) * 1000
121 # Stats for latency returned as N/A so just post them
122 results[ResultsConstants.MAX_LATENCY_NS] = root[0][1][0][0].get(
125 results[ResultsConstants.AVG_LATENCY_NS] = float(
126 root[0][1][0][0].get('AvgLatency')) * 1000
128 # Stats for latency returned as N/A so just post them
129 results[ResultsConstants.AVG_LATENCY_NS] = root[0][1][0][0].get(
131 elif test_type == 'Back2Back':
132 results = OrderedDict()
134 # Just mimic what Ixia does and only return the b2b frame count.
135 # This may change later once its decided the common results stats
136 # to be returned should be.
137 results[ResultsConstants.B2B_FRAMES] = root[0][1][0][0].get(
138 'TotalTxBurstFrames')
140 raise NotImplementedError('Unknown test type in report file.')
144 def _build_packet_header(self, reverse=False):
146 Build a packet header based on traffic profile using scapy external
148 :param reverse: Swap source and destination info when building header
149 :return: packet header in hex
151 srcmac = self._params['traffic']['l2'][
152 'srcmac'] if not reverse else self._params['traffic']['l2'][
154 dstmac = self._params['traffic']['l2'][
155 'dstmac'] if not reverse else self._params['traffic']['l2'][
157 srcip = self._params['traffic']['l3'][
158 'srcip'] if not reverse else self._params['traffic']['l3']['dstip']
159 dstip = self._params['traffic']['l3'][
160 'dstip'] if not reverse else self._params['traffic']['l3']['srcip']
161 layer2 = inet.Ether(src=srcmac, dst=dstmac)
162 layer3 = inet.IP(src=srcip, dst=dstip,
163 proto=self._params['traffic']['l3']['proto'])
164 layer4 = inet.UDP(sport=self._params['traffic']['l4']['srcport'],
165 dport=self._params['traffic']['l4']['dstport'])
166 if self._params['traffic']['vlan']['enabled']:
167 vlan = inet.Dot1Q(vlan=self._params['traffic']['vlan']['id'],
168 prio=self._params['traffic']['vlan']['priority'],
169 id=self._params['traffic']['vlan']['cfi'])
172 packet = layer2/vlan/layer3/layer4 if vlan else layer2/layer3/layer4
173 packet_bytes = bytes(packet)
174 packet_hex = '0x' + binascii.hexlify(packet_bytes).decode('utf-8')
177 def _create_api_result(self):
179 Create result dictionary per trafficgen specifications from socket API
180 stats. If stats are not available return values of 0.
181 :return: ResultsConstants as dictionary
183 # Handle each case of statistics based on if the data is available.
184 # This prevents uncaught exceptions when the stats aren't available.
185 result_dict = OrderedDict()
186 if self.tx_stats.data.get(self.tx_stats.pt_stream_keys[0]):
187 result_dict[ResultsConstants.TX_FRAMES] = self.tx_stats.data[
188 self.tx_stats.pt_stream_keys[0]]['packets']
189 result_dict[ResultsConstants.TX_RATE_FPS] = self.tx_stats.data[
190 self.tx_stats.pt_stream_keys[0]]['pps']
191 result_dict[ResultsConstants.TX_RATE_MBPS] = self.tx_stats.data[
192 self.tx_stats.pt_stream_keys[0]]['bps'] / 1000000
193 result_dict[ResultsConstants.TX_BYTES] = self.tx_stats.data[
194 self.tx_stats.pt_stream_keys[0]]['bytes']
195 # tx rate percent may need to be halved if bi directional
196 result_dict[ResultsConstants.TX_RATE_PERCENT] = line_percentage(
197 self.xmanager.ports[0], self.tx_stats, self._duration,
198 self._params['traffic']['l2']['framesize']) if \
199 self._params['traffic']['bidir'] == 'False' else\
201 self.xmanager.ports[0], self.tx_stats, self._duration,
202 self._params['traffic']['l2']['framesize']) / 2
204 self._logger.error('Transmit stats not available.')
205 result_dict[ResultsConstants.TX_FRAMES] = 0
206 result_dict[ResultsConstants.TX_RATE_FPS] = 0
207 result_dict[ResultsConstants.TX_RATE_MBPS] = 0
208 result_dict[ResultsConstants.TX_BYTES] = 0
209 result_dict[ResultsConstants.TX_RATE_PERCENT] = 0
211 if self.rx_stats.data.get('pr_tpldstraffic'):
212 result_dict[ResultsConstants.RX_FRAMES] = self.rx_stats.data[
213 'pr_tpldstraffic']['0']['packets']
215 ResultsConstants.THROUGHPUT_RX_FPS] = self.rx_stats.data[
216 'pr_tpldstraffic']['0']['pps']
218 ResultsConstants.THROUGHPUT_RX_MBPS] = self.rx_stats.data[
219 'pr_tpldstraffic']['0']['bps'] / 1000000
220 result_dict[ResultsConstants.RX_BYTES] = self.rx_stats.data[
221 'pr_tpldstraffic']['0']['bytes']
222 # throughput percent may need to be halved if bi directional
224 ResultsConstants.THROUGHPUT_RX_PERCENT] = line_percentage(
225 self.xmanager.ports[1], self.rx_stats, self._duration,
226 self._params['traffic']['l2']['framesize']) if \
227 self._params['traffic']['bidir'] == 'False' else \
229 self.xmanager.ports[1], self.rx_stats, self._duration,
230 self._params['traffic']['l2']['framesize']) / 2
233 self._logger.error('Receive stats not available.')
234 result_dict[ResultsConstants.RX_FRAMES] = 0
235 result_dict[ResultsConstants.THROUGHPUT_RX_FPS] = 0
236 result_dict[ResultsConstants.THROUGHPUT_RX_MBPS] = 0
237 result_dict[ResultsConstants.RX_BYTES] = 0
238 result_dict[ResultsConstants.THROUGHPUT_RX_PERCENT] = 0
240 if self.rx_stats.data.get('pr_tplderrors'):
241 result_dict[ResultsConstants.PAYLOAD_ERR] = self.rx_stats.data[
242 'pr_tplderrors']['0']['pld']
243 result_dict[ResultsConstants.SEQ_ERR] = self.rx_stats.data[
244 'pr_tplderrors']['0']['seq']
246 result_dict[ResultsConstants.PAYLOAD_ERR] = 0
247 result_dict[ResultsConstants.SEQ_ERR] = 0
249 if self.rx_stats.data.get('pr_tpldlatency'):
250 result_dict[ResultsConstants.MIN_LATENCY_NS] = self.rx_stats.data[
251 'pr_tpldlatency']['0']['min']
252 result_dict[ResultsConstants.MAX_LATENCY_NS] = self.rx_stats.data[
253 'pr_tpldlatency']['0']['max']
254 result_dict[ResultsConstants.AVG_LATENCY_NS] = self.rx_stats.data[
255 'pr_tpldlatency']['0']['avg']
257 result_dict[ResultsConstants.MIN_LATENCY_NS] = 0
258 result_dict[ResultsConstants.MAX_LATENCY_NS] = 0
259 result_dict[ResultsConstants.AVG_LATENCY_NS] = 0
263 def _setup_json_config(self, trials, loss_rate, testtype=None):
265 Create a 2bUsed json file that will be used for xena2544.exe execution.
266 :param trials: Number of trials
267 :param loss_rate: The acceptable loss rate as float
268 :param testtype: Either '2544_b2b' or '2544_throughput' as string
272 j_file = XenaJSON('./tools/pkt_gen/xena/profiles/baseconfig.x2544')
273 j_file.set_chassis_info(
274 settings.getValue('TRAFFICGEN_XENA_IP'),
275 settings.getValue('TRAFFICGEN_XENA_PASSWORD')
277 j_file.set_port(0, settings.getValue('TRAFFICGEN_XENA_MODULE1'),
278 settings.getValue('TRAFFICGEN_XENA_PORT1'))
279 j_file.set_port(1, settings.getValue('TRAFFICGEN_XENA_MODULE2'),
280 settings.getValue('TRAFFICGEN_XENA_PORT2'))
281 j_file.set_port_ip_v4(
282 0, settings.getValue("TRAFFICGEN_XENA_PORT0_IP"),
283 settings.getValue("TRAFFICGEN_XENA_PORT0_CIDR"),
284 settings.getValue("TRAFFICGEN_XENA_PORT0_GATEWAY"))
285 j_file.set_port_ip_v4(
286 1, settings.getValue("TRAFFICGEN_XENA_PORT1_IP"),
287 settings.getValue("TRAFFICGEN_XENA_PORT1_CIDR"),
288 settings.getValue("TRAFFICGEN_XENA_PORT1_GATEWAY"))
290 if testtype == '2544_throughput':
291 j_file.set_test_options_tput(
292 packet_sizes=self._params['traffic']['l2']['framesize'],
293 iterations=trials, loss_rate=loss_rate,
294 duration=self._duration, micro_tpld=True if self._params[
295 'traffic']['l2']['framesize'] == 64 else False)
296 j_file.enable_throughput_test()
298 elif testtype == '2544_b2b':
299 j_file.set_test_options_back2back(
300 packet_sizes=self._params['traffic']['l2']['framesize'],
301 iterations=trials, duration=self._duration,
302 startvalue=self._params['traffic']['frame_rate'],
303 endvalue=self._params['traffic']['frame_rate'],
304 micro_tpld=True if self._params[
305 'traffic']['l2']['framesize'] == 64 else False)
306 j_file.enable_back2back_test()
308 j_file.set_header_layer2(
309 dst_mac=self._params['traffic']['l2']['dstmac'],
310 src_mac=self._params['traffic']['l2']['srcmac'])
311 j_file.set_header_layer3(
312 src_ip=self._params['traffic']['l3']['srcip'],
313 dst_ip=self._params['traffic']['l3']['dstip'],
314 protocol=self._params['traffic']['l3']['proto'])
315 j_file.set_header_layer4_udp(
316 source_port=self._params['traffic']['l4']['srcport'],
317 destination_port=self._params['traffic']['l4']['dstport'])
318 if self._params['traffic']['vlan']['enabled']:
319 j_file.set_header_vlan(
320 vlan_id=self._params['traffic']['vlan']['id'],
321 id=self._params['traffic']['vlan']['cfi'],
322 prio=self._params['traffic']['vlan']['priority'])
323 j_file.add_header_segments(
324 flows=self._params['traffic']['multistream'],
325 multistream_layer=self._params['traffic']['stream_type'])
327 if self._params['traffic']['bidir'] == "True":
328 j_file.set_topology_mesh()
330 j_file.set_topology_blocks()
332 j_file.write_config('./tools/pkt_gen/xena/profiles/2bUsed.x2544')
333 except Exception as exc:
334 self._logger.exception("Error during Xena JSON setup: %s", exc)
337 def _start_traffic_api(self, packet_limit):
339 Start the Xena traffic using the socket API driver
340 :param packet_limit: packet limit for stream, set to -1 for no limit
343 if not self.xmanager:
344 self._xsocket = XenaSocketDriver(
345 settings.getValue('TRAFFICGEN_XENA_IP'))
346 self.xmanager = XenaManager(
347 self._xsocket, settings.getValue('TRAFFICGEN_XENA_USER'),
348 settings.getValue('TRAFFICGEN_XENA_PASSWORD'))
350 # for the report file version info ask the chassis directly for its
352 settings.setValue('XENA_VERSION', 'XENA Socket API - {}'.format(
353 self.xmanager.get_version()))
355 if not len(self.xmanager.ports):
356 self.xmanager.ports[0] = self.xmanager.add_module_port(
357 settings.getValue('TRAFFICGEN_XENA_MODULE1'),
358 settings.getValue('TRAFFICGEN_XENA_PORT1'))
359 if not self.xmanager.ports[0].reserve_port():
361 'Unable to reserve port 0. Please release Xena Port')
363 if len(self.xmanager.ports) < 2:
364 self.xmanager.ports[1] = self.xmanager.add_module_port(
365 settings.getValue('TRAFFICGEN_XENA_MODULE2'),
366 settings.getValue('TRAFFICGEN_XENA_PORT2'))
367 if not self.xmanager.ports[1].reserve_port():
369 'Unable to reserve port 1. Please release Xena Port')
371 # Clear port configuration for a clean start
372 self.xmanager.ports[0].reset_port()
373 self.xmanager.ports[1].reset_port()
374 self.xmanager.ports[0].clear_stats()
375 self.xmanager.ports[1].clear_stats()
377 # set the port IP from the conf file
378 self.xmanager.ports[0].set_port_ip(
379 settings.getValue('TRAFFICGEN_XENA_PORT0_IP'),
380 settings.getValue('TRAFFICGEN_XENA_PORT0_CIDR'),
381 settings.getValue('TRAFFICGEN_XENA_PORT0_GATEWAY'))
382 self.xmanager.ports[1].set_port_ip(
383 settings.getValue('TRAFFICGEN_XENA_PORT1_IP'),
384 settings.getValue('TRAFFICGEN_XENA_PORT1_CIDR'),
385 settings.getValue('TRAFFICGEN_XENA_PORT1_GATEWAY'))
387 def setup_stream(stream, port, payload_id, flip_addr=False):
389 Helper function to configure streams.
390 :param stream: Stream object from XenaDriver module
391 :param port: Port object from XenaDriver module
392 :param payload_id: payload ID as int
393 :param flip_addr: Boolean if the source and destination addresses
398 stream.set_packet_limit(packet_limit)
400 stream.set_rate_fraction(
401 10000 * self._params['traffic']['frame_rate'])
402 stream.set_packet_header(self._build_packet_header(
404 stream.set_header_protocol(
405 'ETHERNET VLAN IP UDP' if self._params['traffic']['vlan'][
406 'enabled'] else 'ETHERNET IP UDP')
407 stream.set_packet_length(
408 'fixed', self._params['traffic']['l2']['framesize'], 16383)
409 stream.set_packet_payload('incrementing', '0x00')
410 stream.set_payload_id(payload_id)
411 port.set_port_time_limit(self._duration * 1000000)
413 if self._params['traffic']['l2']['framesize'] == 64:
415 port.micro_tpld_enable()
417 if self._params['traffic']['multistream']:
418 stream.enable_multistream(
419 flows=self._params['traffic']['multistream'],
420 layer=self._params['traffic']['stream_type'])
422 s1_p0 = self.xmanager.ports[0].add_stream()
423 setup_stream(s1_p0, self.xmanager.ports[0], 0)
425 if self._params['traffic']['bidir'] == 'True':
426 s1_p1 = self.xmanager.ports[1].add_stream()
427 setup_stream(s1_p1, self.xmanager.ports[1], 1, flip_addr=True)
429 if not self.xmanager.ports[0].traffic_on():
431 "Failure to start port 0. Check settings and retry.")
432 if self._params['traffic']['bidir'] == 'True':
433 if not self.xmanager.ports[1].traffic_on():
435 "Failure to start port 1. Check settings and retry.")
436 sleep(self._duration)
438 if self._params['traffic']['bidir'] == 'True':
439 # need to aggregate out both ports stats and assign that data
440 self.rx_stats = self.xmanager.ports[1].get_rx_stats()
441 self.tx_stats = self.xmanager.ports[0].get_tx_stats()
442 self.tx_stats.data = aggregate_stats(
444 self.xmanager.ports[1].get_tx_stats().data)
445 self.rx_stats.data = aggregate_stats(
447 self.xmanager.ports[0].get_rx_stats().data)
449 # no need to aggregate, just grab the appropriate port stats
450 self.tx_stats = self.xmanager.ports[0].get_tx_stats()
451 self.rx_stats = self.xmanager.ports[1].get_rx_stats()
454 def _stop_api_traffic(self):
456 Stop traffic through the socket API
457 :return: Return results from _create_api_result method
459 self.xmanager.ports[0].traffic_off()
460 if self._params['traffic']['bidir'] == 'True':
461 self.xmanager.ports[1].traffic_off()
464 stat = self._create_api_result()
469 self._logger.debug('Connect')
472 def disconnect(self):
473 """Disconnect from the traffic generator.
475 As with :func:`connect`, this function is optional.
478 Where implemented, this function should raise an exception on
483 self._logger.debug('disconnect')
485 self.xmanager.disconnect()
489 self._xsocket.disconnect()
492 def send_burst_traffic(self, traffic=None, numpkts=100, duration=20):
493 """Send a burst of traffic.
495 Send a ``numpkts`` packets of traffic, using ``traffic``
496 configuration, with a timeout of ``time``.
499 :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
500 :param numpkts: Number of packets to send
501 :param duration: Time to wait to receive packets
503 :returns: dictionary of strings with following data:
507 - List of List of Rx Bytes,
508 - Payload Errors and Sequence Errors.
510 raise NotImplementedError('Xena burst traffic not implemented')
512 def send_cont_traffic(self, traffic=None, duration=20):
513 """Send a continuous flow of traffic.
515 See ITrafficGenerator for description
517 self._duration = duration
520 self._params['traffic'] = self.traffic_defaults.copy()
522 self._params['traffic'] = merge_spec(self._params['traffic'],
525 self._start_traffic_api(-1)
526 return self._stop_api_traffic()
528 def start_cont_traffic(self, traffic=None, duration=20):
529 """Non-blocking version of 'send_cont_traffic'.
531 See ITrafficGenerator for description
533 self._duration = duration
536 self._params['traffic'] = self.traffic_defaults.copy()
538 self._params['traffic'] = merge_spec(self._params['traffic'],
541 self._start_traffic_api(-1)
543 def stop_cont_traffic(self):
544 """Stop continuous transmission and return results.
546 return self._stop_api_traffic()
548 def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
550 """Send traffic per RFC2544 throughput test specifications.
552 See ITrafficGenerator for description
554 self._duration = duration
557 self._params['traffic'] = self.traffic_defaults.copy()
559 self._params['traffic'] = merge_spec(self._params['traffic'],
562 self._setup_json_config(trials, lossrate, '2544_throughput')
564 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
565 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
566 "./tools/pkt_gen/xena", "-u",
567 settings.getValue('TRAFFICGEN_XENA_USER')]
568 self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
569 self.mono_pipe.communicate()
570 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
571 return Xena._create_throughput_result(root)
573 def start_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
575 """Non-blocking version of 'send_rfc2544_throughput'.
577 See ITrafficGenerator for description
579 self._duration = duration
581 self._params['traffic'] = self.traffic_defaults.copy()
583 self._params['traffic'] = merge_spec(self._params['traffic'],
586 self._setup_json_config(trials, lossrate, '2544_throughput')
588 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
589 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
590 "./tools/pkt_gen/xena", "-u",
591 settings.getValue('TRAFFICGEN_XENA_USER')]
592 self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
594 def wait_rfc2544_throughput(self):
595 """Wait for and return results of RFC2544 test.
597 See ITrafficGenerator for description
599 self.mono_pipe.communicate()
601 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
602 return Xena._create_throughput_result(root)
604 def send_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
606 """Send traffic per RFC2544 back2back test specifications.
608 See ITrafficGenerator for description
610 self._duration = duration
613 self._params['traffic'] = self.traffic_defaults.copy()
615 self._params['traffic'] = merge_spec(self._params['traffic'],
618 self._setup_json_config(trials, lossrate, '2544_b2b')
620 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
621 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
622 "./tools/pkt_gen/xena", "-u",
623 settings.getValue('TRAFFICGEN_XENA_USER')]
624 self.mono_pipe = subprocess.Popen(
625 args, stdout=sys.stdout)
626 self.mono_pipe.communicate()
627 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
628 return Xena._create_throughput_result(root)
630 def start_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
632 """Non-blocking version of 'send_rfc2544_back2back'.
634 See ITrafficGenerator for description
636 self._duration = duration
639 self._params['traffic'] = self.traffic_defaults.copy()
641 self._params['traffic'] = merge_spec(self._params['traffic'],
644 self._setup_json_config(trials, lossrate, '2544_b2b')
646 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
647 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
648 "./tools/pkt_gen/xena", "-u",
649 settings.getValue('TRAFFICGEN_XENA_USER')]
650 self.mono_pipe = subprocess.Popen(
651 args, stdout=sys.stdout)
653 def wait_rfc2544_back2back(self):
654 """Wait and set results of RFC2544 test.
656 self.mono_pipe.communicate()
658 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
659 return Xena._create_throughput_result(root)
662 if __name__ == "__main__":