tools/pkt_gen/xena/xena_json.py

   1 # Copyright 2016 Red Hat Inc & Xena Networks.
   2 #
   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
   6 #
   7 #   http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 # Contributors:
  16 #   Dan Amzulescu, Xena Networks
  17 #   Christian Trautman, Red Hat Inc.
  18 #
  19 # Usage can be seen below in unit test. This implementation is designed for one
  20 # module two port Xena chassis runs only.
  21
  22 """
  23 Xena JSON module
  24 """
  25
  26 import base64
  27 from collections import OrderedDict
  28 import json
  29 import locale
  30 import logging
  31 import uuid
  32
  33 import scapy.layers.inet as inet
  34
  35 _LOGGER = logging.getLogger(__name__)
  36 _LOCALE = locale.getlocale()[1]
  37
  38
  39 class XenaJSON(object):
  40     """
  41     Class to modify and read Xena JSON configuration files.
  42     """
  43     def __init__(self, json_path='./profiles/baseconfig.x2544'):
  44         """
  45         Constructor
  46         :param json_path: path to JSON file to read. Expected files must have
  47          two module ports with each port having its own stream config profile.
  48         :return: XenaJSON object
  49         """
  50         self.json_data = read_json_file(json_path)
  51
  52         self.packet_data = OrderedDict()
  53         self.packet_data['layer2'] = None
  54         self.packet_data['vlan'] = None
  55         self.packet_data['layer3'] = None
  56         self.packet_data['layer4'] = None
  57
  58     def _add_multistream_layer(self, entity, seg_uuid, stop_value, layer):
  59         """
  60         Add the multi stream layers to the json file based on the layer provided
  61         :param entity: Entity to append the segment to in entity list
  62         :param seg_uuid: The UUID to attach the multistream layer to
  63         :param stop_value: The number of flows to configure
  64         :param layer: the layer that the multistream will be attached to
  65         :return: None
  66         """
  67         field_name = {
  68             2: ('Dst MAC addr', 'Src MAC addr'),
  69             3: ('Dest IP Addr', 'Src IP Addr'),
  70             4: ('Dest Port', 'Src Port')
  71         }
  72         segments = [
  73             {
  74                 "Offset": 0,
  75                 "Mask": "//8=",  # mask of 255/255
  76                 "Action": "INC",
  77                 "StartValue": 0,
  78                 "StopValue": stop_value,
  79                 "StepValue": 1,
  80                 "RepeatCount": 1,
  81                 "SegmentId": seg_uuid,
  82                 "FieldName": field_name[int(layer)][0]
  83             },
  84             {
  85                 "Offset": 0,
  86                 "Mask": "//8=",  # mask of 255/255
  87                 "Action": "INC",
  88                 "StartValue": 0,
  89                 "StopValue": stop_value,
  90                 "StepValue": 1,
  91                 "RepeatCount": 1,
  92                 "SegmentId": seg_uuid,
  93                 "FieldName": field_name[int(layer)][1]
  94             }
  95         ]
  96
  97         self.json_data['StreamProfileHandler']['EntityList'][entity][
  98             'StreamConfig']['HwModifiers'] = (segments)
  99
 100     def _create_packet_header(self):
 101         """
 102         Create the scapy packet header based on what has been built in this
 103         instance using the set header methods. Return tuple of the two byte
 104         arrays, one for each port.
 105         :return: Scapy packet headers as bytearrays
 106         """
 107         if not self.packet_data['layer2']:
 108             _LOGGER.warning('Using dummy info for layer 2 in Xena JSON file')
 109             self.set_header_layer2()
 110         packet1, packet2 = (self.packet_data['layer2'][0],
 111                             self.packet_data['layer2'][1])
 112         for packet_header in list(self.packet_data.copy().values())[1:]:
 113             if packet_header:
 114                 packet1 /= packet_header[0]
 115                 packet2 /= packet_header[1]
 116         ret = (bytes(packet1), bytes(packet2))
 117         return ret
 118
 119     def add_header_segments(self, flows=0, multistream_layer=None):
 120         """
 121         Build the header segments to write to the JSON file.
 122         :param flows: Number of flows to configure for multistream if enabled
 123         :param multistream_layer: layer to set multistream flows as string.
 124         Acceptable values are L2, L3 or L4
 125         :return: None
 126         """
 127         packet = self._create_packet_header()
 128         segment1 = list()
 129         segment2 = list()
 130         header_pos = 0
 131         if self.packet_data['layer2']:
 132             # slice out the layer 2 bytes from the packet header byte array
 133             layer2 = packet[0][header_pos: len(self.packet_data['layer2'][0])]
 134             seg = create_segment(
 135                 "ETHERNET", encode_byte_array(layer2).decode(_LOCALE))
 136             if multistream_layer == 'L2' and flows > 0:
 137                 self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
 138                                             stop_value=flows, layer=2)
 139             segment1.append(seg)
 140             # now do the other port data with reversed src, dst info
 141             layer2 = packet[1][header_pos: len(self.packet_data['layer2'][1])]
 142             seg = create_segment(
 143                 "ETHERNET", encode_byte_array(layer2).decode(_LOCALE))
 144             segment2.append(seg)
 145             if multistream_layer == 'L2' and flows > 0:
 146                 self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
 147                                             stop_value=flows, layer=2)
 148             header_pos = len(layer2)
 149         if self.packet_data['vlan']:
 150             # slice out the vlan bytes from the packet header byte array
 151             vlan = packet[0][header_pos: len(
 152                 self.packet_data['vlan'][0]) + header_pos]
 153             segment1.append(create_segment(
 154                 "VLAN", encode_byte_array(vlan).decode(_LOCALE)))
 155             segment2.append(create_segment(
 156                 "VLAN", encode_byte_array(vlan).decode(_LOCALE)))
 157             header_pos += len(vlan)
 158         if self.packet_data['layer3']:
 159             # slice out the layer 3 bytes from the packet header byte array
 160             layer3 = packet[0][header_pos: len(
 161                 self.packet_data['layer3'][0]) + header_pos]
 162             seg = create_segment(
 163                 "IP", encode_byte_array(layer3).decode(_LOCALE))
 164             segment1.append(seg)
 165             if multistream_layer == 'L3' and flows > 0:
 166                 self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
 167                                             stop_value=flows, layer=3)
 168             # now do the other port data with reversed src, dst info
 169             layer3 = packet[1][header_pos: len(
 170                 self.packet_data['layer3'][1]) + header_pos]
 171             seg = create_segment(
 172                 "IP", encode_byte_array(layer3).decode(_LOCALE))
 173             segment2.append(seg)
 174             if multistream_layer == 'L3' and flows > 0:
 175                 self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
 176                                             stop_value=flows, layer=3)
 177             header_pos += len(layer3)
 178         if self.packet_data['layer4']:
 179             # slice out the layer 4 bytes from the packet header byte array
 180             layer4 = packet[0][header_pos: len(
 181                 self.packet_data['layer4'][0]) + header_pos]
 182             seg = create_segment(
 183                 "UDP", encode_byte_array(layer4).decode(_LOCALE))
 184             segment1.append(seg)
 185             if multistream_layer == 'L4' and flows > 0:
 186                 self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
 187                                             stop_value=flows, layer=4)
 188             # now do the other port data with reversed src, dst info
 189             layer4 = packet[1][header_pos: len(
 190                 self.packet_data['layer4'][1]) + header_pos]
 191             seg = create_segment(
 192                 "UDP", encode_byte_array(layer4).decode(_LOCALE))
 193             segment2.append(seg)
 194             if multistream_layer == 'L4' and flows > 0:
 195                 self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
 196                                             stop_value=flows, layer=4)
 197             header_pos += len(layer4)
 198
 199         self.json_data['StreamProfileHandler']['EntityList'][0][
 200             'StreamConfig']['HeaderSegments'] = segment1
 201         self.json_data['StreamProfileHandler']['EntityList'][1][
 202             'StreamConfig']['HeaderSegments'] = segment2
 203
 204     def disable_back2back_test(self):
 205         """
 206         Disable the rfc2544 back to back test
 207         :return: None
 208         """
 209         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 210             'Enabled'] = 'false'
 211
 212     def disable_throughput_test(self):
 213         """
 214         Disable the rfc2544 throughput test
 215         :return: None
 216         """
 217         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 218             'Enabled'] = 'false'
 219
 220     def enable_back2back_test(self):
 221         """
 222         Enable the rfc2544 back to back test
 223         :return: None
 224         """
 225         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 226             'Enabled'] = 'true'
 227
 228     def enable_throughput_test(self):
 229         """
 230         Enable the rfc2544 throughput test
 231         :return: None
 232         """
 233         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 234             'Enabled'] = 'true'
 235
 236     def modify_2544_tput_options(self, initial_value, minimum_value,
 237                                  maximum_value, value_resolution,
 238                                  use_pass_threshhold, pass_threshhold):
 239         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 240             'RateIterationOptions']['InitialValue'] = initial_value
 241         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 242             'RateIterationOptions']['MinimumValue'] = minimum_value
 243         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 244             'RateIterationOptions']['MaximumValue'] = maximum_value
 245         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 246             'RateIterationOptions']['ValueResolution'] = value_resolution
 247         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 248             'RateIterationOptions']['UsePassThreshold'] = use_pass_threshhold
 249         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 250             'RateIterationOptions']['PassThreshold'] = pass_threshhold
 251
 252     def set_chassis_info(self, hostname, pwd):
 253         """
 254         Set the chassis info
 255         :param hostname: hostname as string of ip
 256         :param pwd: password to chassis as string
 257         :return: None
 258         """
 259         self.json_data['ChassisManager']['ChassisList'][0][
 260             'HostName'] = hostname
 261         self.json_data['ChassisManager']['ChassisList'][0][
 262             'Password'] = pwd
 263
 264     def set_header_layer2(self, dst_mac='cc:cc:cc:cc:cc:cc',
 265                           src_mac='bb:bb:bb:bb:bb:bb', **kwargs):
 266         """
 267         Build a scapy Ethernet L2 objects inside instance packet_data structure
 268         :param dst_mac: destination mac as string. Example "aa:aa:aa:aa:aa:aa"
 269         :param src_mac: source mac as string. Example "bb:bb:bb:bb:bb:bb"
 270         :param kwargs: Extra params per scapy usage.
 271         :return: None
 272         """
 273         self.packet_data['layer2'] = [
 274             inet.Ether(dst=dst_mac, src=src_mac, **kwargs),
 275             inet.Ether(dst=src_mac, src=dst_mac, **kwargs)]
 276
 277     def set_header_layer3(self, src_ip='192.168.0.2', dst_ip='192.168.0.3',
 278                           protocol='UDP', **kwargs):
 279         """
 280         Build scapy IPV4 L3 objects inside instance packet_data structure
 281         :param src_ip: source IP as string in dot notation format
 282         :param dst_ip: destination IP as string in dot notation format
 283         :param protocol: protocol for l4
 284         :param kwargs: Extra params per scapy usage
 285         :return: None
 286         """
 287         self.packet_data['layer3'] = [
 288             inet.IP(src=src_ip, dst=dst_ip, proto=protocol.lower(), **kwargs),
 289             inet.IP(src=dst_ip, dst=src_ip, proto=protocol.lower(), **kwargs)]
 290
 291     def set_header_layer4_udp(self, source_port, destination_port, **kwargs):
 292         """
 293         Build scapy UDP L4 objects inside instance packet_data structure
 294         :param source_port: Source port as int
 295         :param destination_port: Destination port as int
 296         :param kwargs: Extra params per scapy usage
 297         :return: None
 298         """
 299         self.packet_data['layer4'] = [
 300             inet.UDP(sport=source_port, dport=destination_port, **kwargs),
 301             inet.UDP(sport=source_port, dport=destination_port, **kwargs)]
 302
 303     def set_header_vlan(self, vlan_id=1, **kwargs):
 304         """
 305         Build a Dot1Q scapy object inside instance packet_data structure
 306         :param vlan_id: The VLAN ID
 307         :param kwargs: Extra params per scapy usage
 308         :return: None
 309         """
 310         self.packet_data['vlan'] = [
 311             inet.Dot1Q(vlan=vlan_id, **kwargs),
 312             inet.Dot1Q(vlan=vlan_id, **kwargs)]
 313
 314     def set_port(self, index, module, port):
 315         """
 316         Set the module and port for the 0 index port to use with the test
 317         :param index: Index of port to set, 0 = port1, 1=port2, etc..
 318         :param module: module location as int
 319         :param port: port location in module as int
 320         :return: None
 321         """
 322         self.json_data['PortHandler']['EntityList'][index]['PortRef'][
 323             'ModuleIndex'] = module
 324         self.json_data['PortHandler']['EntityList'][index]['PortRef'][
 325             'PortIndex'] = port
 326
 327     def set_port_ip_v4(self, port, ip_addr, netmask, gateway):
 328         """
 329         Set the port IP info
 330         :param port: port number as int of port to set ip info
 331         :param ip_addr: ip address in dot notation format as string
 332         :param netmask: cidr number for netmask (ie 24/16/8) as int
 333         :param gateway: gateway address in dot notation format
 334         :return: None
 335         """
 336         available_ports = range(len(
 337             self.json_data['PortHandler']['EntityList']))
 338         if port not in available_ports:
 339             raise ValueError("{}{}{}".format(
 340                 'Port assignment must be an available port ',
 341                 'number in baseconfig file. Port=', port))
 342         self.json_data['PortHandler']['EntityList'][
 343             port]["IpV4Address"] = ip_addr
 344         self.json_data['PortHandler']['EntityList'][
 345             port]["IpV4Gateway"] = gateway
 346         self.json_data['PortHandler']['EntityList'][
 347             port]["IpV4RoutingPrefix"] = int(netmask)
 348
 349     def set_port_ip_v6(self, port, ip_addr, netmask, gateway):
 350         """
 351         Set the port IP info
 352         :param port: port number as int of port to set ip info
 353         :param ip_addr: ip address as 8 groups of 4 hexadecimal groups separated
 354          by a colon.
 355         :param netmask: cidr number for netmask (ie 24/16/8) as int
 356         :param gateway: gateway address as string in 8 group of 4 hexadecimal
 357                         groups separated by a colon.
 358         :return: None
 359         """
 360         available_ports = range(len(
 361             self.json_data['PortHandler']['EntityList']))
 362         if port not in available_ports:
 363             raise ValueError("{}{}{}".format(
 364                 'Port assignment must be an available port ',
 365                 'number in baseconfig file. Port=', port))
 366         self.json_data['PortHandler']['EntityList'][
 367             port]["IpV6Address"] = ip_addr
 368         self.json_data['PortHandler']['EntityList'][
 369             port]["IpV6Gateway"] = gateway
 370         self.json_data['PortHandler']['EntityList'][
 371             port]["IpV6RoutingPrefix"] = int(netmask)
 372
 373     def set_test_options_tput(self, packet_sizes, duration, iterations,
 374                               loss_rate, micro_tpld=False):
 375         """
 376         Set the tput test options
 377         :param packet_sizes: List of packet sizes to test, single int entry is
 378          acceptable for one packet size testing
 379         :param duration: time for each test in seconds as int
 380         :param iterations: number of iterations of testing as int
 381         :param loss_rate: acceptable loss rate as float
 382         :param micro_tpld: boolean if micro_tpld should be enabled or disabled
 383         :return: None
 384         """
 385         if isinstance(packet_sizes, int):
 386             packet_sizes = [packet_sizes]
 387         self.json_data['TestOptions']['PacketSizes'][
 388             'CustomPacketSizes'] = packet_sizes
 389         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 390             'Duration'] = duration
 391         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 392             'RateIterationOptions']['AcceptableLoss'] = loss_rate
 393         self.json_data['TestOptions']['FlowCreationOptions'][
 394             'UseMicroTpldOnDemand'] = 'true' if micro_tpld else 'false'
 395         self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
 396             'Iterations'] = iterations
 397
 398     def set_test_options_back2back(self, packet_sizes, duration,
 399                                    iterations, startvalue, endvalue,
 400                                    micro_tpld=False):
 401         """
 402         Set the back2back test options
 403         :param packet_sizes: List of packet sizes to test, single int entry is
 404          acceptable for one packet size testing
 405         :param duration: time for each test in seconds as int
 406         :param iterations: number of iterations of testing as int
 407         :param micro_tpld: boolean if micro_tpld should be enabled or disabled
 408         :param StartValue: start value
 409         :param EndValue: end value
 410         :return: None
 411         """
 412         if isinstance(packet_sizes, int):
 413             packet_sizes = [packet_sizes]
 414         self.json_data['TestOptions']['PacketSizes'][
 415             'CustomPacketSizes'] = packet_sizes
 416         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 417             'Duration'] = duration
 418         self.json_data['TestOptions']['FlowCreationOptions'][
 419             'UseMicroTpldOnDemand'] = 'true' if micro_tpld else 'false'
 420         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 421             'Iterations'] = iterations
 422         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 423             'RateSweepOptions']['StartValue'] = startvalue
 424         self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
 425             'RateSweepOptions']['EndValue'] = endvalue
 426
 427     def set_topology_blocks(self):
 428         """
 429         Set the test topology to a West to East config for half duplex flow with
 430         port 0 as the sender and port 1 as the receiver.
 431         :return: None
 432         """
 433         self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'BLOCKS'
 434         self.json_data['TestOptions']['TopologyConfig'][
 435             'Direction'] = 'WEST_EAST'
 436         self.json_data['PortHandler']['EntityList'][0][
 437             'PortGroup'] = "WEST"
 438         self.json_data['PortHandler']['EntityList'][1][
 439             'PortGroup'] = "EAST"
 440
 441     def set_topology_mesh(self):
 442         """
 443         Set the test topology to Mesh for bi directional full duplex flow
 444         :return: None
 445         """
 446         self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'MESH'
 447         self.json_data['TestOptions']['TopologyConfig']['Direction'] = 'BIDIR'
 448         self.json_data['PortHandler']['EntityList'][0][
 449             'PortGroup'] = "UNDEFINED"
 450         self.json_data['PortHandler']['EntityList'][1][
 451             'PortGroup'] = "UNDEFINED"
 452
 453     def write_config(self, path='./2bUsed.x2544'):
 454         """
 455         Write the config to out as file
 456         :param path: Output file to export the json data to
 457         :return: None
 458         """
 459         if not write_json_file(self.json_data, path):
 460             raise RuntimeError("Could not write out file, please check config")
 461
 462
 463 def create_segment(header_type, encode_64_string):
 464     """
 465     Create segment for JSON file
 466     :param header_type: Type of header as string
 467     :param encode_64_string: 64 byte encoded string value of the hex bytes
 468     :return: segment as dictionary
 469     """
 470     return {
 471         "SegmentType": header_type.upper(),
 472         "SegmentValue": encode_64_string,
 473         "ItemID": str(uuid.uuid4()),
 474         "ParentID": "",
 475         "Label": ""}
 476
 477
 478 def decode_byte_array(enc_str):
 479     """ Decodes the base64-encoded string to a byte array
 480     :param enc_str: The base64-encoded string representing a byte array
 481     :return: The decoded byte array
 482     """
 483     dec_string = base64.b64decode(enc_str)
 484     barray = bytearray()
 485     barray.extend(dec_string)
 486     return barray
 487
 488
 489 def encode_byte_array(byte_arr):
 490     """ Encodes the byte array as a base64-encoded string
 491     :param byte_arr: A bytearray containing the bytes to convert
 492     :return: A base64 encoded string
 493     """
 494     enc_string = base64.b64encode(bytes(byte_arr))
 495     return enc_string
 496
 497
 498 def print_json_report(json_data):
 499     """
 500     Print out info from the json data for testing purposes only.
 501     :param json_data: json loaded data from json.loads
 502     :return: None
 503     """
 504     print("<<Xena JSON Config Report>>\n")
 505     try:
 506         print("### Chassis Info ###")
 507         print("Chassis IP: {}".format(json_data['ChassisManager'][
 508             'ChassisList'][0]['HostName']))
 509         print("Chassis Password: {}".format(json_data['ChassisManager'][
 510             'ChassisList'][0]['Password']))
 511         print("### Port Configuration ###")
 512         print("Port 1 IPv4:{}/{} gateway:{}".format(
 513             json_data['PortHandler']['EntityList'][0]["IpV4Address"],
 514             json_data['PortHandler']['EntityList'][0]["IpV4RoutingPrefix"],
 515             json_data['PortHandler']['EntityList'][0]["IpV4Gateway"]))
 516         print("Port 1 IPv6:{}/{} gateway:{}".format(
 517             json_data['PortHandler']['EntityList'][0]["IpV6Address"],
 518             json_data['PortHandler']['EntityList'][0]["IpV6RoutingPrefix"],
 519             json_data['PortHandler']['EntityList'][0]["IpV6Gateway"]))
 520         print("Port 2 IPv4:{}/{} gateway:{}".format(
 521             json_data['PortHandler']['EntityList'][1]["IpV4Address"],
 522             json_data['PortHandler']['EntityList'][1]["IpV4RoutingPrefix"],
 523             json_data['PortHandler']['EntityList'][1]["IpV4Gateway"]))
 524         print("Port 2 IPv6:{}/{} gateway:{}".format(
 525             json_data['PortHandler']['EntityList'][1]["IpV6Address"],
 526             json_data['PortHandler']['EntityList'][1]["IpV6RoutingPrefix"],
 527             json_data['PortHandler']['EntityList'][1]["IpV6Gateway"]))
 528         print("Port 1: {}/{} group: {}".format(
 529             json_data['PortHandler']['EntityList'][0]['PortRef']['ModuleIndex'],
 530             json_data['PortHandler']['EntityList'][0]['PortRef']['PortIndex'],
 531             json_data['PortHandler']['EntityList'][0]['PortGroup']))
 532         print("Port 2: {}/{} group: {}".format(
 533             json_data['PortHandler']['EntityList'][1]['PortRef']['ModuleIndex'],
 534             json_data['PortHandler']['EntityList'][1]['PortRef']['PortIndex'],
 535             json_data['PortHandler']['EntityList'][1]['PortGroup']))
 536         print("### Tests Enabled ###")
 537         print("Back2Back Enabled: {}".format(json_data['TestOptions'][
 538             'TestTypeOptionMap']['Back2Back']['Enabled']))
 539         print("Throughput Enabled: {}".format(json_data['TestOptions'][
 540             'TestTypeOptionMap']['Throughput']['Enabled']))
 541         print("### Test Options ###")
 542         print("Test topology: {}/{}".format(
 543             json_data['TestOptions']['TopologyConfig']['Topology'],
 544             json_data['TestOptions']['TopologyConfig']['Direction']))
 545         print("Packet Sizes: {}".format(json_data['TestOptions'][
 546             'PacketSizes']['CustomPacketSizes']))
 547         print("Test duration: {}".format(json_data['TestOptions'][
 548             'TestTypeOptionMap']['Throughput']['Duration']))
 549         print("Acceptable loss rate: {}".format(json_data['TestOptions'][
 550             'TestTypeOptionMap']['Throughput']['RateIterationOptions'][
 551                 'AcceptableLoss']))
 552         print("Micro TPLD enabled: {}".format(json_data['TestOptions'][
 553             'FlowCreationOptions']['UseMicroTpldOnDemand']))
 554         print("Test iterations: {}".format(json_data['TestOptions'][
 555             'TestTypeOptionMap']['Throughput']['Iterations']))
 556         if 'StreamConfig' in json_data['StreamProfileHandler']['EntityList'][0]:
 557             print("### Header segments ###")
 558             for seg in json_data['StreamProfileHandler']['EntityList']:
 559                 for header in seg['StreamConfig']['HeaderSegments']:
 560                     print("Type: {}".format(
 561                         header['SegmentType']))
 562                     print("Value: {}".format(decode_byte_array(
 563                         header['SegmentValue'])))
 564             print("### Multi Stream config ###")
 565             for seg in json_data['StreamProfileHandler']['EntityList']:
 566                 for header in seg['StreamConfig']['HwModifiers']:
 567                     print(header)
 568     except KeyError as exc:
 569         print("Error setting not found in JSON data: {}".format(exc))
 570
 571
 572 def read_json_file(json_file):
 573     """
 574     Read the json file path and return a dictionary of the data
 575     :param json_file: path to json file
 576     :return: dictionary of json data
 577     """
 578     try:
 579         with open(json_file, 'r', encoding=_LOCALE) as data_file:
 580             file_data = json.loads(data_file.read())
 581     except ValueError as exc:
 582         # general json exception, Python 3.5 adds new exception type
 583         _LOGGER.exception("Exception with json read: %s", exc)
 584         raise
 585     except IOError as exc:
 586         _LOGGER.exception(
 587             'Exception during file open: %s file=%s', exc, json_file)
 588         raise
 589     return file_data
 590
 591
 592 def write_json_file(json_data, output_path):
 593     """
 594     Write out the dictionary of data to a json file
 595     :param json_data: dictionary of json data
 596     :param output_path: file path to write output
 597     :return: Boolean if success
 598     """
 599     try:
 600         with open(output_path, 'w', encoding=_LOCALE) as fileh:
 601             json.dump(json_data, fileh, indent=2, sort_keys=True,
 602                       ensure_ascii=True)
 603         return True
 604     except ValueError as exc:
 605         # general json exception, Python 3.5 adds new exception type
 606         _LOGGER.exception(
 607             "Exception with json write: %s", exc)
 608         return False
 609     except IOError as exc:
 610         _LOGGER.exception(
 611             'Exception during file write: %s file=%s', exc, output_path)
 612         return False
 613
 614
 615 if __name__ == "__main__":
 616     print("Running UnitTest for XenaJSON")
 617     JSON = XenaJSON()
 618     print_json_report(JSON.json_data)
 619     JSON.set_chassis_info('192.168.0.5', 'vsperf')
 620     JSON.set_port(0, 1, 0)
 621     JSON.set_port(1, 1, 1)
 622     JSON.set_port_ip_v4(0, '192.168.240.10', 32, '192.168.240.1')
 623     JSON.set_port_ip_v4(1, '192.168.240.11', 32, '192.168.240.1')
 624     JSON.set_port_ip_v6(0, 'a1a1:a2a2:a3a3:a4a4:a5a5:a6a6:a7a7:a8a8', 128,
 625                         'a1a1:a2a2:a3a3:a4a4:a5a5:a6a6:a7a7:1111')
 626     JSON.set_port_ip_v6(1, 'b1b1:b2b2:b3b3:b4b4:b5b5:b6b6:b7b7:b8b8', 128,
 627                         'b1b1:b2b2:b3b3:b4b4:b5b5:b6b6:b7b7:1111')
 628     JSON.set_header_layer2(dst_mac='dd:dd:dd:dd:dd:dd',
 629                            src_mac='ee:ee:ee:ee:ee:ee')
 630     JSON.set_header_vlan(vlan_id=5)
 631     JSON.set_header_layer3(src_ip='192.168.100.2', dst_ip='192.168.100.3',
 632                            protocol='udp')
 633     JSON.set_header_layer4_udp(source_port=3000, destination_port=3001)
 634     JSON.set_test_options_tput(packet_sizes=[64], duration=10, iterations=1,
 635                                loss_rate=0.0, micro_tpld=True)
 636     JSON.add_header_segments(flows=4000, multistream_layer='L4')
 637     JSON.set_topology_blocks()
 638     write_json_file(JSON.json_data, './testthis.x2544')
 639     JSON = XenaJSON('./testthis.x2544')
 640     print_json_report(JSON.json_data)
 641