diff options
Diffstat (limited to 'tools/pkt_gen')
| -rwxr-xr-x | tools/pkt_gen/xena/xena.py | 49 | ||||
| -rw-r--r-- | tools/pkt_gen/xena/xena_json.py | 39 |
2 files changed, 60 insertions, 28 deletions
diff --git a/tools/pkt_gen/xena/xena.py b/tools/pkt_gen/xena/xena.py index 194de343..e76ebcbf 100755 --- a/tools/pkt_gen/xena/xena.py +++ b/tools/pkt_gen/xena/xena.py @@ -30,6 +30,8 @@ import sys from time import sleep import xml.etree.ElementTree as ET from collections import OrderedDict +# scapy imports +import scapy.layers.inet as inet # VSPerf imports from conf import settings @@ -48,10 +50,6 @@ from tools.pkt_gen.xena.XenaDriver import ( XenaManager, ) -# scapy imports -import scapy.layers.inet as inet - - class Xena(ITrafficGenerator): """ Xena Traffic generator wrapper class @@ -288,14 +286,23 @@ class Xena(ITrafficGenerator): 1, settings.getValue("TRAFFICGEN_XENA_PORT1_IP"), settings.getValue("TRAFFICGEN_XENA_PORT1_CIDR"), settings.getValue("TRAFFICGEN_XENA_PORT1_GATEWAY")) - j_file.set_test_options( - packet_sizes=self._params['traffic']['l2']['framesize'], - iterations=trials, loss_rate=loss_rate, - duration=self._duration, micro_tpld=True if self._params[ - 'traffic']['l2']['framesize'] == 64 else False) + if testtype == '2544_throughput': + j_file.set_test_options_tput( + packet_sizes=self._params['traffic']['l2']['framesize'], + iterations=trials, loss_rate=loss_rate, + duration=self._duration, micro_tpld=True if self._params[ + 'traffic']['l2']['framesize'] == 64 else False) j_file.enable_throughput_test() + elif testtype == '2544_b2b': + j_file.set_test_options_back2back( + packet_sizes=self._params['traffic']['l2']['framesize'], + iterations=trials, duration=self._duration, + startvalue=self._params['traffic']['frame_rate'], + endvalue=self._params['traffic']['frame_rate'], + micro_tpld=True if self._params[ + 'traffic']['l2']['framesize'] == 64 else False) j_file.enable_back2back_test() j_file.set_header_layer2( @@ -485,22 +492,18 @@ class Xena(ITrafficGenerator): def send_burst_traffic(self, traffic=None, numpkts=100, duration=20): """Send a burst of traffic. - Send a ``numpkts`` packets of traffic, using ``traffic`` - configuration, with a timeout of ``time``. + See ITrafficGenerator for description + """ + self._duration = duration - Attributes: - :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags - :param numpkts: Number of packets to send - :param duration: Time to wait to receive packets + self._params.clear() + self._params['traffic'] = self.traffic_defaults.copy() + if traffic: + self._params['traffic'] = merge_spec(self._params['traffic'], + traffic) - :returns: dictionary of strings with following data: - - List of Tx Frames, - - List of Rx Frames, - - List of Tx Bytes, - - List of List of Rx Bytes, - - Payload Errors and Sequence Errors. - """ - raise NotImplementedError('Xena burst traffic not implemented') + self._start_traffic_api(numpkts) + return self._stop_api_traffic() def send_cont_traffic(self, traffic=None, duration=20): """Send a continuous flow of traffic. diff --git a/tools/pkt_gen/xena/xena_json.py b/tools/pkt_gen/xena/xena_json.py index 23b56084..2a15a932 100644 --- a/tools/pkt_gen/xena/xena_json.py +++ b/tools/pkt_gen/xena/xena_json.py @@ -354,10 +354,10 @@ class XenaJSON(object): self.json_data['PortHandler']['EntityList'][ port]["IpV6RoutingPrefix"] = int(netmask) - def set_test_options(self, packet_sizes, duration, iterations, loss_rate, - micro_tpld=False): + def set_test_options_tput(self, packet_sizes, duration, iterations, + loss_rate, micro_tpld=False): """ - Set the test options + Set the tput test options :param packet_sizes: List of packet sizes to test, single int entry is acceptable for one packet size testing :param duration: time for each test in seconds as int @@ -379,6 +379,35 @@ class XenaJSON(object): self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][ 'Iterations'] = iterations + def set_test_options_back2back(self, packet_sizes, duration, + iterations, startvalue, endvalue, + micro_tpld=False): + """ + Set the back2back test options + :param packet_sizes: List of packet sizes to test, single int entry is + acceptable for one packet size testing + :param duration: time for each test in seconds as int + :param iterations: number of iterations of testing as int + :param micro_tpld: boolean if micro_tpld should be enabled or disabled + :param StartValue: start value + :param EndValue: end value + :return: None + """ + if isinstance(packet_sizes, int): + packet_sizes = [packet_sizes] + self.json_data['TestOptions']['PacketSizes'][ + 'CustomPacketSizes'] = packet_sizes + self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][ + 'Duration'] = duration + self.json_data['TestOptions']['FlowCreationOptions'][ + 'UseMicroTpldOnDemand'] = 'true' if micro_tpld else 'false' + self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][ + 'Iterations'] = iterations + self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][ + 'RateSweepOptions']['StartValue'] = startvalue + self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][ + 'RateSweepOptions']['EndValue'] = endvalue + def set_topology_blocks(self): """ Set the test topology to a West to East config for half duplex flow with @@ -586,8 +615,8 @@ if __name__ == "__main__": JSON.set_header_layer3(src_ip='192.168.100.2', dst_ip='192.168.100.3', protocol='udp') JSON.set_header_layer4_udp(source_port=3000, destination_port=3001) - JSON.set_test_options(packet_sizes=[64], duration=10, iterations=1, - loss_rate=0.0, micro_tpld=True) + JSON.set_test_options_tput(packet_sizes=[64], duration=10, iterations=1, + loss_rate=0.0, micro_tpld=True) JSON.add_header_segments(flows=4000, multistream_layer='L4') JSON.set_topology_blocks() write_json_file(JSON.json_data, './testthis.x2544') |