summaryrefslogtreecommitdiffstats
path: root/tools/pkt_gen/xena
diff options
context:
space:
mode:
Diffstat (limited to 'tools/pkt_gen/xena')
-rw-r--r--tools/pkt_gen/xena/json/__init__.py13
-rw-r--r--tools/pkt_gen/xena/json/json_utilities.py105
-rw-r--r--tools/pkt_gen/xena/json/xena_json.py (renamed from tools/pkt_gen/xena/xena_json.py)258
-rw-r--r--tools/pkt_gen/xena/json/xena_json_blocks.py50
-rw-r--r--tools/pkt_gen/xena/json/xena_json_mesh.py46
-rw-r--r--tools/pkt_gen/xena/json/xena_json_pairs.py60
-rwxr-xr-xtools/pkt_gen/xena/xena.py56
7 files changed, 335 insertions, 253 deletions
diff --git a/tools/pkt_gen/xena/json/__init__.py b/tools/pkt_gen/xena/json/__init__.py
new file mode 100644
index 00000000..54e0301a
--- /dev/null
+++ b/tools/pkt_gen/xena/json/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2015-2017 Intel Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
diff --git a/tools/pkt_gen/xena/json/json_utilities.py b/tools/pkt_gen/xena/json/json_utilities.py
new file mode 100644
index 00000000..857aa102
--- /dev/null
+++ b/tools/pkt_gen/xena/json/json_utilities.py
@@ -0,0 +1,105 @@
+# Copyright 2017 Red Hat Inc & Xena Networks.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Contributors:
+# Dan Amzulescu, Xena Networks
+# Christian Trautman, Red Hat Inc.
+
+"""JSON utility module"""
+
+import base64
+import json
+import locale
+import logging
+import uuid
+
+_LOGGER = logging.getLogger(__name__)
+_LOCALE = locale.getlocale()[1]
+
+def create_segment(header_type, encode_64_string):
+ """
+ Create segment for JSON file
+ :param header_type: Type of header as string
+ :param encode_64_string: 64 byte encoded string value of the hex bytes
+ :return: segment as dictionary
+ """
+ return {
+ "SegmentType": header_type.upper(),
+ "SegmentValue": encode_64_string,
+ "ItemID": str(uuid.uuid4()),
+ "ParentID": "",
+ "Label": ""}
+
+
+def decode_byte_array(enc_str):
+ """ Decodes the base64-encoded string to a byte array
+ :param enc_str: The base64-encoded string representing a byte array
+ :return: The decoded byte array
+ """
+ dec_string = base64.b64decode(enc_str)
+ barray = bytearray()
+ barray.extend(dec_string)
+ return barray
+
+
+def encode_byte_array(byte_arr):
+ """ Encodes the byte array as a base64-encoded string
+ :param byte_arr: A bytearray containing the bytes to convert
+ :return: A base64 encoded string
+ """
+ enc_string = base64.b64encode(bytes(byte_arr))
+ return enc_string
+
+
+def read_json_file(json_file):
+ """
+ Read the json file path and return a dictionary of the data
+ :param json_file: path to json file
+ :return: dictionary of json data
+ """
+ try:
+ with open(json_file, 'r', encoding=_LOCALE) as data_file:
+ file_data = json.loads(data_file.read())
+ except ValueError as exc:
+ # general json exception, Python 3.5 adds new exception type
+ _LOGGER.exception("Exception with json read: %s", exc)
+ raise
+ except IOError as exc:
+ _LOGGER.exception(
+ 'Exception during file open: %s file=%s', exc, json_file)
+ raise
+ return file_data
+
+
+def write_json_file(json_data, output_path):
+ """
+ Write out the dictionary of data to a json file
+ :param json_data: dictionary of json data
+ :param output_path: file path to write output
+ :return: Boolean if success
+ """
+ try:
+ with open(output_path, 'w', encoding=_LOCALE) as fileh:
+ json.dump(json_data, fileh, indent=2, sort_keys=True,
+ ensure_ascii=True)
+ return True
+ except ValueError as exc:
+ # general json exception, Python 3.5 adds new exception type
+ _LOGGER.exception(
+ "Exception with json write: %s", exc)
+ return False
+ except IOError as exc:
+ _LOGGER.exception(
+ 'Exception during file write: %s file=%s', exc, output_path)
+ return False
diff --git a/tools/pkt_gen/xena/xena_json.py b/tools/pkt_gen/xena/json/xena_json.py
index 50d0e2fe..b1eed720 100644
--- a/tools/pkt_gen/xena/xena_json.py
+++ b/tools/pkt_gen/xena/json/xena_json.py
@@ -23,31 +23,35 @@
Xena JSON module
"""
-import base64
from collections import OrderedDict
-import json
import locale
import logging
-import uuid
+import os
import scapy.layers.inet as inet
+from tools.pkt_gen.xena.json import json_utilities
+
_LOGGER = logging.getLogger(__name__)
_LOCALE = locale.getlocale()[1]
+_CURR_DIR = os.path.dirname(os.path.realpath(__file__))
+
class XenaJSON(object):
"""
- Class to modify and read Xena JSON configuration files.
+ Parent Class to modify and read Xena JSON configuration files.
"""
- def __init__(self, json_path='./profiles/baseconfig.x2544'):
+ def __init__(self,
+ json_path=os.path.join(
+ _CURR_DIR, '../profiles/baseconfig.x2544')):
"""
Constructor
:param json_path: path to JSON file to read. Expected files must have
two module ports with each port having its own stream config profile.
:return: XenaJSON object
"""
- self.json_data = read_json_file(json_path)
+ self.json_data = json_utilities.read_json_file(json_path)
self.packet_data = OrderedDict()
self.packet_data['layer2'] = None
@@ -131,16 +135,18 @@ class XenaJSON(object):
if self.packet_data['layer2']:
# slice out the layer 2 bytes from the packet header byte array
layer2 = packet[0][header_pos: len(self.packet_data['layer2'][0])]
- seg = create_segment(
- "ETHERNET", encode_byte_array(layer2).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "ETHERNET", json_utilities.encode_byte_array(layer2).decode(
+ _LOCALE))
if multistream_layer == 'L2' and flows > 0:
self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
stop_value=flows, layer=2)
segment1.append(seg)
# now do the other port data with reversed src, dst info
layer2 = packet[1][header_pos: len(self.packet_data['layer2'][1])]
- seg = create_segment(
- "ETHERNET", encode_byte_array(layer2).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "ETHERNET", json_utilities.encode_byte_array(layer2).decode(
+ _LOCALE))
segment2.append(seg)
if multistream_layer == 'L2' and flows > 0:
self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
@@ -150,17 +156,17 @@ class XenaJSON(object):
# slice out the vlan bytes from the packet header byte array
vlan = packet[0][header_pos: len(
self.packet_data['vlan'][0]) + header_pos]
- segment1.append(create_segment(
- "VLAN", encode_byte_array(vlan).decode(_LOCALE)))
- segment2.append(create_segment(
- "VLAN", encode_byte_array(vlan).decode(_LOCALE)))
+ segment1.append(json_utilities.create_segment(
+ "VLAN", json_utilities.encode_byte_array(vlan).decode(_LOCALE)))
+ segment2.append(json_utilities.create_segment(
+ "VLAN", json_utilities.encode_byte_array(vlan).decode(_LOCALE)))
header_pos += len(vlan)
if self.packet_data['layer3']:
# slice out the layer 3 bytes from the packet header byte array
layer3 = packet[0][header_pos: len(
self.packet_data['layer3'][0]) + header_pos]
- seg = create_segment(
- "IP", encode_byte_array(layer3).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "IP", json_utilities.encode_byte_array(layer3).decode(_LOCALE))
segment1.append(seg)
if multistream_layer == 'L3' and flows > 0:
self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
@@ -168,8 +174,8 @@ class XenaJSON(object):
# now do the other port data with reversed src, dst info
layer3 = packet[1][header_pos: len(
self.packet_data['layer3'][1]) + header_pos]
- seg = create_segment(
- "IP", encode_byte_array(layer3).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "IP", json_utilities.encode_byte_array(layer3).decode(_LOCALE))
segment2.append(seg)
if multistream_layer == 'L3' and flows > 0:
self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
@@ -179,8 +185,8 @@ class XenaJSON(object):
# slice out the layer 4 bytes from the packet header byte array
layer4 = packet[0][header_pos: len(
self.packet_data['layer4'][0]) + header_pos]
- seg = create_segment(
- "UDP", encode_byte_array(layer4).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "UDP", json_utilities.encode_byte_array(layer4).decode(_LOCALE))
segment1.append(seg)
if multistream_layer == 'L4' and flows > 0:
self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
@@ -188,8 +194,8 @@ class XenaJSON(object):
# now do the other port data with reversed src, dst info
layer4 = packet[1][header_pos: len(
self.packet_data['layer4'][1]) + header_pos]
- seg = create_segment(
- "UDP", encode_byte_array(layer4).decode(_LOCALE))
+ seg = json_utilities.create_segment(
+ "UDP", json_utilities.encode_byte_array(layer4).decode(_LOCALE))
segment2.append(seg)
if multistream_layer == 'L4' and flows > 0:
self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
@@ -427,217 +433,11 @@ class XenaJSON(object):
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
- port 0 as the sender and port 1 as the receiver.
- :return: None
- """
- self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'BLOCKS'
- self.json_data['TestOptions']['TopologyConfig'][
- 'Direction'] = 'WEST_EAST'
- self.json_data['PortHandler']['EntityList'][0][
- 'PortGroup'] = "WEST"
- self.json_data['PortHandler']['EntityList'][1][
- 'PortGroup'] = "EAST"
-
- def set_topology_mesh(self):
- """
- Set the test topology to Mesh for bi directional full duplex flow
- :return: None
- """
- self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'MESH'
- self.json_data['TestOptions']['TopologyConfig']['Direction'] = 'BIDIR'
- self.json_data['PortHandler']['EntityList'][0][
- 'PortGroup'] = "UNDEFINED"
- self.json_data['PortHandler']['EntityList'][1][
- 'PortGroup'] = "UNDEFINED"
-
def write_config(self, path='./2bUsed.x2544'):
"""
Write the config to out as file
:param path: Output file to export the json data to
:return: None
"""
- if not write_json_file(self.json_data, path):
+ if not json_utilities.write_json_file(self.json_data, path):
raise RuntimeError("Could not write out file, please check config")
-
-
-def create_segment(header_type, encode_64_string):
- """
- Create segment for JSON file
- :param header_type: Type of header as string
- :param encode_64_string: 64 byte encoded string value of the hex bytes
- :return: segment as dictionary
- """
- return {
- "SegmentType": header_type.upper(),
- "SegmentValue": encode_64_string,
- "ItemID": str(uuid.uuid4()),
- "ParentID": "",
- "Label": ""}
-
-
-def decode_byte_array(enc_str):
- """ Decodes the base64-encoded string to a byte array
- :param enc_str: The base64-encoded string representing a byte array
- :return: The decoded byte array
- """
- dec_string = base64.b64decode(enc_str)
- barray = bytearray()
- barray.extend(dec_string)
- return barray
-
-
-def encode_byte_array(byte_arr):
- """ Encodes the byte array as a base64-encoded string
- :param byte_arr: A bytearray containing the bytes to convert
- :return: A base64 encoded string
- """
- enc_string = base64.b64encode(bytes(byte_arr))
- return enc_string
-
-
-def print_json_report(json_data):
- """
- Print out info from the json data for testing purposes only.
- :param json_data: json loaded data from json.loads
- :return: None
- """
- print("<<Xena JSON Config Report>>\n")
- try:
- print("### Chassis Info ###")
- print("Chassis IP: {}".format(json_data['ChassisManager'][
- 'ChassisList'][0]['HostName']))
- print("Chassis Password: {}".format(json_data['ChassisManager'][
- 'ChassisList'][0]['Password']))
- print("### Port Configuration ###")
- print("Port 1 IPv4:{}/{} gateway:{}".format(
- json_data['PortHandler']['EntityList'][0]["IpV4Address"],
- json_data['PortHandler']['EntityList'][0]["IpV4RoutingPrefix"],
- json_data['PortHandler']['EntityList'][0]["IpV4Gateway"]))
- print("Port 1 IPv6:{}/{} gateway:{}".format(
- json_data['PortHandler']['EntityList'][0]["IpV6Address"],
- json_data['PortHandler']['EntityList'][0]["IpV6RoutingPrefix"],
- json_data['PortHandler']['EntityList'][0]["IpV6Gateway"]))
- print("Port 2 IPv4:{}/{} gateway:{}".format(
- json_data['PortHandler']['EntityList'][1]["IpV4Address"],
- json_data['PortHandler']['EntityList'][1]["IpV4RoutingPrefix"],
- json_data['PortHandler']['EntityList'][1]["IpV4Gateway"]))
- print("Port 2 IPv6:{}/{} gateway:{}".format(
- json_data['PortHandler']['EntityList'][1]["IpV6Address"],
- json_data['PortHandler']['EntityList'][1]["IpV6RoutingPrefix"],
- json_data['PortHandler']['EntityList'][1]["IpV6Gateway"]))
- print("Port 1: {}/{} group: {}".format(
- json_data['PortHandler']['EntityList'][0]['PortRef']['ModuleIndex'],
- json_data['PortHandler']['EntityList'][0]['PortRef']['PortIndex'],
- json_data['PortHandler']['EntityList'][0]['PortGroup']))
- print("Port 2: {}/{} group: {}".format(
- json_data['PortHandler']['EntityList'][1]['PortRef']['ModuleIndex'],
- json_data['PortHandler']['EntityList'][1]['PortRef']['PortIndex'],
- json_data['PortHandler']['EntityList'][1]['PortGroup']))
- print("### Tests Enabled ###")
- print("Back2Back Enabled: {}".format(json_data['TestOptions'][
- 'TestTypeOptionMap']['Back2Back']['Enabled']))
- print("Throughput Enabled: {}".format(json_data['TestOptions'][
- 'TestTypeOptionMap']['Throughput']['Enabled']))
- print("### Test Options ###")
- print("Test topology: {}/{}".format(
- json_data['TestOptions']['TopologyConfig']['Topology'],
- json_data['TestOptions']['TopologyConfig']['Direction']))
- print("Packet Sizes: {}".format(json_data['TestOptions'][
- 'PacketSizes']['CustomPacketSizes']))
- print("Test duration: {}".format(json_data['TestOptions'][
- 'TestTypeOptionMap']['Throughput']['Duration']))
- print("Acceptable loss rate: {}".format(json_data['TestOptions'][
- 'TestTypeOptionMap']['Throughput']['RateIterationOptions'][
- 'AcceptableLoss']))
- print("Micro TPLD enabled: {}".format(json_data['TestOptions'][
- 'FlowCreationOptions']['UseMicroTpldOnDemand']))
- print("Test iterations: {}".format(json_data['TestOptions'][
- 'TestTypeOptionMap']['Throughput']['Iterations']))
- if 'StreamConfig' in json_data['StreamProfileHandler']['EntityList'][0]:
- print("### Header segments ###")
- for seg in json_data['StreamProfileHandler']['EntityList']:
- for header in seg['StreamConfig']['HeaderSegments']:
- print("Type: {}".format(
- header['SegmentType']))
- print("Value: {}".format(decode_byte_array(
- header['SegmentValue'])))
- print("### Multi Stream config ###")
- for seg in json_data['StreamProfileHandler']['EntityList']:
- for header in seg['StreamConfig']['HwModifiers']:
- print(header)
- except KeyError as exc:
- print("Error setting not found in JSON data: {}".format(exc))
-
-
-def read_json_file(json_file):
- """
- Read the json file path and return a dictionary of the data
- :param json_file: path to json file
- :return: dictionary of json data
- """
- try:
- with open(json_file, 'r', encoding=_LOCALE) as data_file:
- file_data = json.loads(data_file.read())
- except ValueError as exc:
- # general json exception, Python 3.5 adds new exception type
- _LOGGER.exception("Exception with json read: %s", exc)
- raise
- except IOError as exc:
- _LOGGER.exception(
- 'Exception during file open: %s file=%s', exc, json_file)
- raise
- return file_data
-
-
-def write_json_file(json_data, output_path):
- """
- Write out the dictionary of data to a json file
- :param json_data: dictionary of json data
- :param output_path: file path to write output
- :return: Boolean if success
- """
- try:
- with open(output_path, 'w', encoding=_LOCALE) as fileh:
- json.dump(json_data, fileh, indent=2, sort_keys=True,
- ensure_ascii=True)
- return True
- except ValueError as exc:
- # general json exception, Python 3.5 adds new exception type
- _LOGGER.exception(
- "Exception with json write: %s", exc)
- return False
- except IOError as exc:
- _LOGGER.exception(
- 'Exception during file write: %s file=%s', exc, output_path)
- return False
-
-
-if __name__ == "__main__":
- print("Running UnitTest for XenaJSON")
- JSON = XenaJSON()
- print_json_report(JSON.json_data)
- JSON.set_chassis_info('192.168.0.5', 'vsperf')
- JSON.set_port(0, 1, 0)
- JSON.set_port(1, 1, 1)
- JSON.set_port_ip_v4(0, '192.168.240.10', 32, '192.168.240.1')
- JSON.set_port_ip_v4(1, '192.168.240.11', 32, '192.168.240.1')
- JSON.set_port_ip_v6(0, 'a1a1:a2a2:a3a3:a4a4:a5a5:a6a6:a7a7:a8a8', 128,
- 'a1a1:a2a2:a3a3:a4a4:a5a5:a6a6:a7a7:1111')
- JSON.set_port_ip_v6(1, 'b1b1:b2b2:b3b3:b4b4:b5b5:b6b6:b7b7:b8b8', 128,
- 'b1b1:b2b2:b3b3:b4b4:b5b5:b6b6:b7b7:1111')
- JSON.set_header_layer2(dst_mac='dd:dd:dd:dd:dd:dd',
- src_mac='ee:ee:ee:ee:ee:ee')
- JSON.set_header_vlan(vlan_id=5)
- 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_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')
- JSON = XenaJSON('./testthis.x2544')
- print_json_report(JSON.json_data)
diff --git a/tools/pkt_gen/xena/json/xena_json_blocks.py b/tools/pkt_gen/xena/json/xena_json_blocks.py
new file mode 100644
index 00000000..d0a0864c
--- /dev/null
+++ b/tools/pkt_gen/xena/json/xena_json_blocks.py
@@ -0,0 +1,50 @@
+# Copyright 2017 Red Hat Inc & Xena Networks.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Contributors:
+# Dan Amzulescu, Xena Networks
+# Christian Trautman, Red Hat Inc.
+#
+# Usage can be seen below in unit test. This implementation is designed for one
+# module two port Xena chassis runs only.
+
+"""
+Xena JSON module for blocks topology
+"""
+
+from tools.pkt_gen.xena.json.xena_json import XenaJSON
+
+
+class XenaJSONBlocks(XenaJSON):
+ """
+ Class to modify and read Xena JSON configuration files. Topology will be set
+ as Blocks
+ """
+ def __init__(self):
+ super(XenaJSONBlocks, self).__init__()
+ self.set_topology_blocks()
+
+ def set_topology_blocks(self):
+ """
+ Set the test topology to a West to East config for half duplex flow with
+ port 0 as the sender and port 1 as the receiver.
+ :return: None
+ """
+ self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'BLOCKS'
+ self.json_data['TestOptions']['TopologyConfig'][
+ 'Direction'] = 'WEST_EAST'
+ self.json_data['PortHandler']['EntityList'][0][
+ 'PortGroup'] = "WEST"
+ self.json_data['PortHandler']['EntityList'][1][
+ 'PortGroup'] = "EAST"
diff --git a/tools/pkt_gen/xena/json/xena_json_mesh.py b/tools/pkt_gen/xena/json/xena_json_mesh.py
new file mode 100644
index 00000000..5d6a1f13
--- /dev/null
+++ b/tools/pkt_gen/xena/json/xena_json_mesh.py
@@ -0,0 +1,46 @@
+# Copyright 2017 Red Hat Inc & Xena Networks.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Contributors:
+# Dan Amzulescu, Xena Networks
+# Christian Trautman, Red Hat Inc.
+#
+# Usage can be seen below in unit test. This implementation is designed for one
+# module two port Xena chassis runs only.
+
+"""
+Xena JSON module for mesh topology.
+"""
+
+from tools.pkt_gen.xena.json.xena_json import XenaJSON
+
+
+class XenaJSONMesh(XenaJSON):
+ """
+ Class to modify and read Xena JSON configuration files. Topology will be set
+ as Mesh.
+ """
+ def __init__(self):
+ super(XenaJSONMesh, self).__init__()
+ self.set_topology_mesh()
+
+ def set_topology_mesh(self):
+ """
+ Set the test topology to Mesh for bi directional full duplex flow
+ :return: None
+ """
+ self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'MESH'
+ self.json_data['TestOptions']['TopologyConfig']['Direction'] = 'BIDIR'
+ self.json_data['PortHandler']['EntityList'][0]['PortGroup'] = "UNDEFINED"
+ self.json_data['PortHandler']['EntityList'][1]['PortGroup'] = "UNDEFINED"
diff --git a/tools/pkt_gen/xena/json/xena_json_pairs.py b/tools/pkt_gen/xena/json/xena_json_pairs.py
new file mode 100644
index 00000000..4f97f0ba
--- /dev/null
+++ b/tools/pkt_gen/xena/json/xena_json_pairs.py
@@ -0,0 +1,60 @@
+# Copyright 2017 Red Hat Inc & Xena Networks.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Contributors:
+# Dan Amzulescu, Xena Networks
+# Christian Trautman, Red Hat Inc.
+#
+# Usage can be seen below in unit test. This implementation is designed for one
+# module two port Xena chassis runs only.
+
+"""
+Xena JSON module for pairs topology
+"""
+
+from tools.pkt_gen.xena.json.xena_json import XenaJSON
+
+
+class XenaJSONPairs(XenaJSON):
+ """
+ Class to modify and read Xena JSON configuration files. Topology will be set
+ as Blocks
+ """
+ def __init__(self):
+ super(XenaJSONPairs, self).__init__()
+ self.set_topology_pairs()
+
+ def set_topology_pairs(self):
+ """
+ Set the test topology to loopback pairs for packets that need to return
+ to the same port they are sent from
+ :return: None
+ """
+ self.json_data['TestOptions']['TopologyConfig']['Topology'] = 'PAIRS'
+ self.json_data['TestOptions']['TopologyConfig'][
+ 'Direction'] = 'WEST_EAST'
+ self.json_data['PortHandler']['EntityList'][0][
+ 'PortGroup'] = "WEST"
+ self.json_data['PortHandler']['EntityList'][1][
+ 'PortGroup'] = "WEST"
+
+ def remove_port(self, port_number):
+ """
+ Remove a port from the config.
+ :param port_number: port to remove from json config as int
+ :return: None
+ """
+ del self.json_data['PortHandler']['EntityList'][port_number]
+ del self.json_data['StreamHandler']['StreamConnectionList'][0][
+ 'Port{}Id'.format(port_number + 1)]
diff --git a/tools/pkt_gen/xena/xena.py b/tools/pkt_gen/xena/xena.py
index 26d26542..31347c5f 100755
--- a/tools/pkt_gen/xena/xena.py
+++ b/tools/pkt_gen/xena/xena.py
@@ -28,27 +28,27 @@ import logging
import os
import subprocess
import sys
-from time import sleep
import xml.etree.ElementTree as ET
from collections import OrderedDict
-# scapy imports
+from time import sleep
+
import scapy.layers.inet as inet
-# VSPerf imports
-from conf import settings
from conf import merge_spec
+from conf import settings
from core.results.results_constants import ResultsConstants
from tools.pkt_gen.trafficgen.trafficgen import ITrafficGenerator
-
-# Xena module imports
-from tools.pkt_gen.xena.xena_json import XenaJSON
from tools.pkt_gen.xena.XenaDriver import (
aggregate_stats,
line_percentage,
XenaSocketDriver,
XenaManager,
)
+from tools.pkt_gen.xena.json.xena_json_mesh import XenaJSONMesh
+from tools.pkt_gen.xena.json.xena_json_blocks import XenaJSONBlocks
+from tools.pkt_gen.xena.json.xena_json_pairs import XenaJSONPairs
+_CURR_DIR = os.path.dirname(os.path.realpath(__file__))
class Xena(ITrafficGenerator):
"""
@@ -263,16 +263,28 @@ class Xena(ITrafficGenerator):
return result_dict
- def _setup_json_config(self, tests, loss_rate, testtype=None):
+ def _setup_json_config(self, tests, loss_rate, testtype=None,
+ bonding_test=False):
"""
Create a 2bUsed json file that will be used for xena2544.exe execution.
:param tests: Number of tests
:param loss_rate: The acceptable loss rate as float
:param testtype: Either '2544_b2b' or '2544_throughput' as string
+ :param bonding_test: Specify if the test is a bonding test which will
+ enable the pairs topology
:return: None
"""
try:
- j_file = XenaJSON('./tools/pkt_gen/xena/profiles/baseconfig.x2544')
+ # set duplex mode, this code is valid, pylint complaining with a
+ # warning that many have complained about online.
+ # pylint: disable=redefined-variable-type
+ if self._params['traffic']['bidir'] == "True":
+ j_file = XenaJSONMesh()
+ elif self._params['traffic']['bidir'] == "False":
+ j_file = XenaJSONBlocks()
+ elif bonding_test:
+ j_file = XenaJSONPairs()
+
j_file.set_chassis_info(
settings.getValue('TRAFFICGEN_XENA_IP'),
settings.getValue('TRAFFICGEN_XENA_PASSWORD')
@@ -337,13 +349,9 @@ class Xena(ITrafficGenerator):
j_file.add_header_segments(
flows=self._params['traffic']['multistream'],
multistream_layer=self._params['traffic']['stream_type'])
- # set duplex mode
- if self._params['traffic']['bidir'] == "True":
- j_file.set_topology_mesh()
- else:
- j_file.set_topology_blocks()
- j_file.write_config('./tools/pkt_gen/xena/profiles/2bUsed.x2544')
+ j_file.write_config(os.path.join(
+ _CURR_DIR, 'profiles/2bUsed.x2544'))
except Exception as exc:
self._logger.exception("Error during Xena JSON setup: %s", exc)
raise
@@ -488,9 +496,9 @@ class Xena(ITrafficGenerator):
Start the xena2544 exe.
:return: None
"""
- args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
- "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
- "./tools/pkt_gen/xena", "-u",
+ args = ["mono", os.path.join(_CURR_DIR, "Xena2544.exe"), "-c",
+ os.path.join(_CURR_DIR, "profiles/2bUsed.x2544"), "-e", "-r",
+ _CURR_DIR, "-u",
settings.getValue('TRAFFICGEN_XENA_USER')]
# Sometimes Xena2544.exe completes, but mono holds the process without
# releasing it, this can cause a deadlock of the main thread. Use the
@@ -624,7 +632,7 @@ class Xena(ITrafficGenerator):
self._start_xena_2544()
self._wait_xena_2544_complete()
- root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ root = ET.parse(os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
if settings.getValue('TRAFFICGEN_XENA_RFC2544_VERIFY'):
# record the previous settings so we can revert to them if needed to
@@ -652,7 +660,7 @@ class Xena(ITrafficGenerator):
'TRAFFICGEN_XENA_RFC2544_VERIFY_DURATION'), lossrate)
self.wait_rfc2544_throughput()
root = ET.parse(
- r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
# If it passed, report the number of lost frames and exit the
# loop
@@ -695,7 +703,7 @@ class Xena(ITrafficGenerator):
traffic, old_tests, self._duration, lossrate)
self.wait_rfc2544_throughput()
root = ET.parse(
- r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
else:
self._logger.error(
'Maximum number of verify attempts reached. Reporting last result')
@@ -723,7 +731,7 @@ class Xena(ITrafficGenerator):
See ITrafficGenerator for description
"""
self._wait_xena_2544_complete()
- root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ root = ET.parse(os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
return Xena._create_throughput_result(root)
def send_rfc2544_back2back(self, traffic=None, tests=1, duration=20,
@@ -742,7 +750,7 @@ class Xena(ITrafficGenerator):
self._setup_json_config(tests, lossrate, '2544_b2b')
self._start_xena_2544()
self._wait_xena_2544_complete()
- root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ root = ET.parse(os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
return Xena._create_throughput_result(root)
def start_rfc2544_back2back(self, traffic=None, tests=1, duration=20,
@@ -764,7 +772,7 @@ class Xena(ITrafficGenerator):
"""Wait and set results of RFC2544 test.
"""
self._wait_xena_2544_complete()
- root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ root = ET.parse(os.path.join(_CURR_DIR, "xena2544-report.xml")).getroot()
return Xena._create_throughput_result(root)