aboutsummaryrefslogtreecommitdiffstats
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to 'tools')
-rw-r--r--tools/pkt_gen/xena/__init__.py13
-rw-r--r--tools/pkt_gen/xena/profiles/baseconfig.x2544373
-rwxr-xr-xtools/pkt_gen/xena/xena.py365
-rw-r--r--tools/pkt_gen/xena/xena_json.py528
4 files changed, 1279 insertions, 0 deletions
diff --git a/tools/pkt_gen/xena/__init__.py b/tools/pkt_gen/xena/__init__.py
new file mode 100644
index 00000000..8081be42
--- /dev/null
+++ b/tools/pkt_gen/xena/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2015-2016 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/profiles/baseconfig.x2544 b/tools/pkt_gen/xena/profiles/baseconfig.x2544
new file mode 100644
index 00000000..93c8fabd
--- /dev/null
+++ b/tools/pkt_gen/xena/profiles/baseconfig.x2544
@@ -0,0 +1,373 @@
+{
+ "copyright": [
+ "# Copyright 2015-2016 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\n",
+ "#",
+ "# 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\n",
+ "# limitations under the License."
+ ],
+ "PortHandler": {
+ "EntityList": [
+ {
+ "PortRef": {
+ "ChassisId": "4605b3c9-70cc-42d9-9d8c-16c34989a4c1",
+ "ModuleIndex": 3,
+ "PortIndex": 0
+ },
+ "PortGroup": "UNDEFINED",
+ "PairPeerRef": null,
+ "PairPeerId": "",
+ "MulticastRole": "Undefined",
+ "PortSpeed": "AUTO",
+ "InterFrameGap": 20,
+ "PauseModeOn": false,
+ "AutoNegEnabled": true,
+ "AdjustPpm": 0,
+ "LatencyOffset": 0,
+ "MdiMdixMode": "AUTO",
+ "EnableFec": true,
+ "ReplyArpRequests": true,
+ "ReplyPingRequests": true,
+ "IpV4Address": "192.168.199.10",
+ "IpV4RoutingPrefix": 24,
+ "IpV4Gateway": "192.168.199.1",
+ "IpV6Address": "::",
+ "IpV6RoutingPrefix": 64,
+ "IpV6Gateway": "::",
+ "IpGatewayMacAddress": "AAAAAAAA",
+ "PublicIpAddress": "",
+ "PublicIpRoutingPrefix": 24,
+ "PublicIpAddressV6": "",
+ "PublicIpRoutingPrefixV6": 64,
+ "RemoteLoopIpAddress": "",
+ "RemoteLoopIpAddressV6": "",
+ "RemoteLoopMacAddress": "AAAAAAAA",
+ "EnablePortRateCap": false,
+ "PortRateCapValue": 1000.0,
+ "PortRateCapProfile": "Physical Port Rate",
+ "PortRateCapUnit": "Mbps",
+ "MultiStreamMap": null,
+ "ItemID": "4faf0f0c-2fc6-44a7-87ea-5f47b02d4c1a",
+ "ParentID": "",
+ "Label": ""
+ },
+ {
+ "PortRef": {
+ "ChassisId": "4605b3c9-70cc-42d9-9d8c-16c34989a4c1",
+ "ModuleIndex": 3,
+ "PortIndex": 1
+ },
+ "PortGroup": "UNDEFINED",
+ "PairPeerRef": null,
+ "PairPeerId": "",
+ "MulticastRole": "Undefined",
+ "PortSpeed": "AUTO",
+ "InterFrameGap": 20,
+ "PauseModeOn": false,
+ "AutoNegEnabled": true,
+ "AdjustPpm": 0,
+ "LatencyOffset": 0,
+ "MdiMdixMode": "AUTO",
+ "EnableFec": true,
+ "ReplyArpRequests": true,
+ "ReplyPingRequests": true,
+ "IpV4Address": "192.168.199.11",
+ "IpV4RoutingPrefix": 24,
+ "IpV4Gateway": "192.168.199.1",
+ "IpV6Address": "::",
+ "IpV6RoutingPrefix": 64,
+ "IpV6Gateway": "::",
+ "IpGatewayMacAddress": "AAAAAAAA",
+ "PublicIpAddress": "",
+ "PublicIpRoutingPrefix": 24,
+ "PublicIpAddressV6": "",
+ "PublicIpRoutingPrefixV6": 64,
+ "RemoteLoopIpAddress": "",
+ "RemoteLoopIpAddressV6": "",
+ "RemoteLoopMacAddress": "AAAAAAAA",
+ "EnablePortRateCap": false,
+ "PortRateCapValue": 1000.0,
+ "PortRateCapProfile": "Physical Port Rate",
+ "PortRateCapUnit": "Mbps",
+ "MultiStreamMap": null,
+ "ItemID": "1b88dc59-1b1a-43f5-a314-673219f47545",
+ "ParentID": "",
+ "Label": ""
+ }
+ ]
+ },
+ "StreamHandler": {
+ "StreamConnectionList": [
+ {
+ "ConnectionId": 0,
+ "Port1Id": "4faf0f0c-2fc6-44a7-87ea-5f47b02d4c1a",
+ "Port2Id": "1b88dc59-1b1a-43f5-a314-673219f47545",
+ "AddressOffset1": 2,
+ "AddressOffset2": 3,
+ "ItemID": "244b9295-9a5a-4405-8404-a62074152783",
+ "ParentID": "",
+ "Label": ""
+ }
+ ]
+ },
+ "StreamProfileHandler": {
+ "ProfileAssignmentMap": {
+ "guid_1b88dc59-1b1a-43f5-a314-673219f47545": "033f23c9-3986-40c9-b7e4-9ac1176f3c0b",
+ "guid_4faf0f0c-2fc6-44a7-87ea-5f47b02d4c1a": "106a3aa6-ea43-4dd7-84b5-51424a52ac87"
+ },
+ "EntityList": [
+ {
+ "StreamConfig": {
+ "SwModifier": null,
+ "HwModifiers": [],
+ "FieldValueRanges": [],
+ "StreamDescrPrefix": "Stream",
+ "ResourceIndex": -1,
+ "TpldId": -1,
+ "EnableState": "OFF",
+ "RateType": "Fraction",
+ "PacketLimit": 0,
+ "RateFraction": 100.0,
+ "RatePps": 0.0,
+ "RateL2Mbps": 0.0,
+ "UseBurstValues": false,
+ "BurstSize": 0,
+ "BurstDensity": 100,
+ "HeaderSegments": [],
+ "PacketLengthType": "FIXED",
+ "PacketMinSize": 64,
+ "PacketMaxSize": 64,
+ "PayloadDefinition": {
+ "PayloadType": "Incrementing",
+ "PayloadPattern": "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
+ },
+ "ResourceUsed": false,
+ "ChildResourceUsed": false
+ },
+ "ItemID": "106a3aa6-ea43-4dd7-84b5-51424a52ac87",
+ "ParentID": "",
+ "Label": ""
+ },
+ {
+ "StreamConfig": {
+ "SwModifier": null,
+ "HwModifiers": [],
+ "FieldValueRanges": [],
+ "StreamDescrPrefix": "Stream",
+ "ResourceIndex": -1,
+ "TpldId": -1,
+ "EnableState": "OFF",
+ "RateType": "Fraction",
+ "PacketLimit": 0,
+ "RateFraction": 100.0,
+ "RatePps": 0.0,
+ "RateL2Mbps": 0.0,
+ "UseBurstValues": false,
+ "BurstSize": 0,
+ "BurstDensity": 100,
+ "HeaderSegments": [],
+ "PacketLengthType": "FIXED",
+ "PacketMinSize": 64,
+ "PacketMaxSize": 64,
+ "PayloadDefinition": {
+ "PayloadType": "Incrementing",
+ "PayloadPattern": "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
+ },
+ "ResourceUsed": false,
+ "ChildResourceUsed": false
+ },
+ "ItemID": "033f23c9-3986-40c9-b7e4-9ac1176f3c0b",
+ "ParentID": "",
+ "Label": ""
+ }
+ ]
+ },
+ "TestOptions": {
+ "TestTypeOptionMap": {
+ "Throughput": {
+ "$type": "XenaCommon.TestConfig.Xena2544.TestTypeOptions.ThroughputTestOptions, Xena2544",
+ "RateIterationOptions": {
+ "SearchType": "BinarySearch",
+ "AcceptableLoss": 0.0,
+ "ResultScope": "CommonResult",
+ "FastBinarySearch": false,
+ "InitialValue": 10.0,
+ "MinimumValue": 0.1,
+ "MaximumValue": 100.0,
+ "ValueResolution": 0.5,
+ "UsePassThreshold": false,
+ "PassThreshold": 0.0
+ },
+ "ReportPropertyOptions": [
+ "b"
+ ],
+ "TestType": "Throughput",
+ "Enabled": true,
+ "DurationType": "Seconds",
+ "Duration": 1.0,
+ "DurationFrames": 1,
+ "DurationFrameUnit": "Mframes",
+ "Iterations": 3,
+ "ItemID": "5ba8b4d4-9a52-4697-860a-4af1b97d2a5c",
+ "ParentID": "",
+ "Label": ""
+ },
+ "Latency": {
+ "$type": "XenaCommon.TestConfig.Xena2544.TestTypeOptions.LatencyTestOptions, Xena2544",
+ "RateSweepOptions": {
+ "StartValue": 50.0,
+ "EndValue": 100.0,
+ "StepValue": 50.0
+ },
+ "LatencyMode": "Last_To_Last",
+ "RateRelativeTputMaxRate": true,
+ "TestType": "Latency",
+ "Enabled": false,
+ "DurationType": "Seconds",
+ "Duration": 1.0,
+ "DurationFrames": 1,
+ "DurationFrameUnit": "Mframes",
+ "Iterations": 1,
+ "ItemID": "c63c0362-96a6-434b-9c67-6be518492a49",
+ "ParentID": "",
+ "Label": ""
+ },
+ "Loss": {
+ "$type": "XenaCommon.TestConfig.Xena2544.TestTypeOptions.LossTestOptions, Xena2544",
+ "RateSweepOptions": {
+ "StartValue": 50.0,
+ "EndValue": 100.0,
+ "StepValue": 50.0
+ },
+ "UsePassFailCriteria": false,
+ "AcceptableLoss": 0.0,
+ "AcceptableLossType": "Percent",
+ "TestType": "Loss",
+ "Enabled": false,
+ "DurationType": "Seconds",
+ "Duration": 1.0,
+ "DurationFrames": 1,
+ "DurationFrameUnit": "Mframes",
+ "Iterations": 1,
+ "ItemID": "f5cf336e-c983-4c48-a8cb-88447b3e2adb",
+ "ParentID": "",
+ "Label": ""
+ },
+ "Back2Back": {
+ "$type": "XenaCommon.TestConfig.Xena2544.TestTypeOptions.Back2BackTestOptions, Xena2544",
+ "RateSweepOptions": {
+ "StartValue": 50.0,
+ "EndValue": 100.0,
+ "StepValue": 50.0
+ },
+ "ResultScope": "CommonResult",
+ "BurstResolution": 100.0,
+ "TestType": "Back2Back",
+ "Enabled": false,
+ "DurationType": "Seconds",
+ "Duration": 1.0,
+ "DurationFrames": 1,
+ "DurationFrameUnit": "Mframes",
+ "Iterations": 1,
+ "ItemID": "2c494ee2-16f1-4a40-b28b-aff6ad7464e3",
+ "ParentID": "",
+ "Label": ""
+ }
+ },
+ "PacketSizes": {
+ "PacketSizeType": "CustomSizes",
+ "CustomPacketSizes": [
+ 512.0
+ ],
+ "SwPacketStartSize": 100,
+ "SwPacketEndSize": 1500,
+ "SwPacketStepSize": 100,
+ "HwPacketMinSize": 64,
+ "HwPacketMaxSize": 1500,
+ "MixedSizesWeights": []
+ },
+ "TopologyConfig": {
+ "Topology": "MESH",
+ "Direction": "BIDIR"
+ },
+ "FlowCreationOptions": {
+ "FlowCreationType": "StreamBased",
+ "MacBaseAddress": "4,244,188",
+ "UseGatewayMacAsDmac": true,
+ "EnableMultiStream": false,
+ "PerPortStreamCount": 1,
+ "MultiStreamAddressOffset": 2,
+ "MultiStreamAddressIncrement": 1,
+ "MultiStreamMacBaseAddress": "4,244,188",
+ "UseMicroTpldOnDemand": false
+ },
+ "LearningOptions": {
+ "MacLearningMode": "EveryTrial",
+ "MacLearningRetries": 1,
+ "ArpRefreshEnabled": true,
+ "ArpRefreshPeriod": 4000.0,
+ "UseFlowBasedLearningPreamble": false,
+ "FlowBasedLearningFrameCount": 1,
+ "FlowBasedLearningDelay": 500,
+ "LearningRatePercent": 1.0,
+ "LearningDuration": 5000.0
+ },
+ "ToggleSyncState": true,
+ "SyncOffDuration": 1,
+ "SyncOnDuration": 1,
+ "PayloadDefinition": {
+ "PayloadType": "Incrementing",
+ "PayloadPattern": "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
+ },
+ "EnableSpeedReductSweep": false,
+ "UsePortSyncStart": false,
+ "PortStaggerSteps": 0,
+ "ShouldStopOnLos": true,
+ "PortResetDelay": 5
+ },
+ "CreationDate": "2016-02-24 13:33:50Z",
+ "ChassisManager": {
+ "ChassisList": [
+ {
+ "ChassisID": "4605b3c9-70cc-42d9-9d8c-16c34989a4c1",
+ "HostName": "10.19.15.19",
+ "PortNumber": 22606,
+ "Password": "xena",
+ "ConnectionType": "Native",
+ "UsedModuleList": [],
+ "ResourceIndex": 0,
+ "ResourceUsed": false,
+ "ChildResourceUsed": false
+ }
+ ]
+ },
+ "ReportConfig": {
+ "CustomerName": "Xena Networks",
+ "CustomerServiceID": "",
+ "CustomerAccessID": "",
+ "Comments": "",
+ "RateUnitTerminology": "FPS",
+ "IncludeTestPairInfo": true,
+ "IncludePerStreamInfo": false,
+ "IncludeGraphs": true,
+ "PlotThroughputUnit": "Pps",
+ "GeneratePdf": false,
+ "GenerateHtml": false,
+ "GenerateXml": true,
+ "GenerateCsv": false,
+ "SaveIntermediateResults": false,
+ "ReportFilename": "xena2544-report",
+ "AppendTimestamp": false
+ },
+ "TidAllocationScope": "ConfigScope",
+ "FormatVersion": 10,
+ "ApplicationVersion": "2.39.5876.25884"
+} \ No newline at end of file
diff --git a/tools/pkt_gen/xena/xena.py b/tools/pkt_gen/xena/xena.py
new file mode 100755
index 00000000..88dd3700
--- /dev/null
+++ b/tools/pkt_gen/xena/xena.py
@@ -0,0 +1,365 @@
+# Copyright 2016 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:
+# Rick Alongi, Red Hat Inc.
+# Amit Supugade, Red Hat Inc.
+# Dan Amzulescu, Xena Networks
+# Christian Trautman, Red Hat Inc.
+
+"""
+Xena Traffic Generator Model
+"""
+
+# python imports
+import inspect
+import logging
+import subprocess
+import sys
+from time import sleep
+import xml.etree.ElementTree as ET
+from collections import OrderedDict
+
+# VSPerf imports
+from conf import settings
+from core.results.results_constants import ResultsConstants
+from tools.pkt_gen.trafficgen.trafficgenhelper import (
+ TRAFFIC_DEFAULTS,
+ merge_spec)
+from tools.pkt_gen.trafficgen.trafficgen import ITrafficGenerator
+
+# Xena module imports
+from tools.pkt_gen.xena.xena_json import XenaJSON
+
+
+class Xena(ITrafficGenerator):
+ """
+ Xena Traffic generator wrapper class
+ """
+ _traffic_defaults = TRAFFIC_DEFAULTS.copy()
+ _logger = logging.getLogger(__name__)
+
+ def __init__(self):
+ self.mono_pipe = None
+ self._params = {}
+ self._duration = None
+
+ @property
+ def traffic_defaults(self):
+ """Default traffic values.
+
+ These can be expected to be constant across traffic generators,
+ so no setter is provided. Changes to the structure or contents
+ will likely break traffic generator implementations or tests
+ respectively.
+ """
+ return self._traffic_defaults
+
+ @staticmethod
+ def _create_throughput_result(root):
+ """
+ Create the results based off the output xml file from the Xena2544.exe
+ execution
+ :param root: root dictionary from xml import
+ :return: Results Ordered dictionary based off ResultsConstants
+ """
+ throughput_test = False
+ back2back_test = False
+ # get the calling method so we know how to return the stats
+ caller = inspect.stack()[1][3]
+ if 'throughput' in caller:
+ throughput_test = True
+ elif 'back2back' in caller:
+ back2back_test = True
+ else:
+ raise NotImplementedError(
+ "Unknown implementation for result return")
+
+ if throughput_test:
+ results = OrderedDict()
+ results[ResultsConstants.THROUGHPUT_RX_FPS] = int(
+ root[0][1][0][1].get('PortRxPps'))
+ results[ResultsConstants.THROUGHPUT_RX_MBPS] = int(
+ root[0][1][0][1].get('PortRxBpsL1')) / 1000000
+ results[ResultsConstants.THROUGHPUT_RX_PERCENT] = (
+ 100 - int(root[0][1][0].get('TotalLossRatioPcnt'))) * float(
+ root[0][1][0].get('TotalTxRatePcnt'))/100
+ results[ResultsConstants.TX_RATE_FPS] = root[0][1][0].get(
+ 'TotalTxRateFps')
+ results[ResultsConstants.TX_RATE_MBPS] = float(
+ root[0][1][0].get('TotalTxRateBpsL1')) / 1000000
+ results[ResultsConstants.TX_RATE_PERCENT] = root[0][1][0].get(
+ 'TotalTxRatePcnt')
+ try:
+ results[ResultsConstants.MIN_LATENCY_NS] = float(
+ root[0][1][0][0].get('MinLatency')) * 1000
+ except ValueError:
+ # Stats for latency returned as N/A so just post them
+ results[ResultsConstants.MIN_LATENCY_NS] = root[0][1][0][0].get(
+ 'MinLatency')
+ try:
+ results[ResultsConstants.MAX_LATENCY_NS] = float(
+ root[0][1][0][0].get('MaxLatency')) * 1000
+ except ValueError:
+ # Stats for latency returned as N/A so just post them
+ results[ResultsConstants.MAX_LATENCY_NS] = root[0][1][0][0].get(
+ 'MaxLatency')
+ try:
+ results[ResultsConstants.AVG_LATENCY_NS] = float(
+ root[0][1][0][0].get('AvgLatency')) * 1000
+ except ValueError:
+ # Stats for latency returned as N/A so just post them
+ results[ResultsConstants.AVG_LATENCY_NS] = root[0][1][0][0].get(
+ 'AvgLatency')
+ elif back2back_test:
+ raise NotImplementedError('Back to back results not implemented')
+
+ return results
+
+ def _setup_json_config(self, trials, loss_rate, testtype=None):
+ """
+ Create a 2bUsed json file that will be used for xena2544.exe execution.
+ :param trials: Number of trials
+ :param loss_rate: The acceptable loss rate as float
+ :param testtype: Either '2544_b2b' or '2544_throughput' as string
+ :return: None
+ """
+ try:
+ j_file = XenaJSON('./tools/pkt_gen/xena/profiles/baseconfig.x2544')
+ j_file.set_chassis_info(
+ settings.getValue('TRAFFICGEN_XENA_IP'),
+ settings.getValue('TRAFFICGEN_XENA_PASSWORD')
+ )
+ j_file.set_port(0, settings.getValue('TRAFFICGEN_XENA_MODULE1'),
+ settings.getValue('TRAFFICGEN_XENA_PORT1')
+ )
+ j_file.set_port(1, settings.getValue('TRAFFICGEN_XENA_MODULE2'),
+ settings.getValue('TRAFFICGEN_XENA_PORT2')
+ )
+ 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.enable_throughput_test()
+ elif testtype == '2544_b2b':
+ j_file.enable_back2back_test()
+
+ j_file.set_header_layer2(
+ dst_mac=self._params['traffic']['l2']['dstmac'],
+ src_mac=self._params['traffic']['l2']['srcmac'])
+ j_file.set_header_layer3(
+ src_ip=self._params['traffic']['l3']['srcip'],
+ dst_ip=self._params['traffic']['l3']['dstip'],
+ protocol=self._params['traffic']['l3']['proto'])
+ j_file.set_header_layer4_udp(
+ source_port=self._params['traffic']['l4']['srcport'],
+ destination_port=self._params['traffic']['l4']['dstport'])
+ if self._params['traffic']['vlan']['enabled']:
+ j_file.set_header_vlan(
+ vlan_id=self._params['traffic']['vlan']['id'],
+ id=self._params['traffic']['vlan']['cfi'],
+ prio=self._params['traffic']['vlan']['priority'])
+ 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']:
+ j_file.set_topology_mesh()
+ else:
+ j_file.set_topology_blocks()
+
+ j_file.write_config('./tools/pkt_gen/xena/profiles/2bUsed.x2544')
+ except Exception as exc:
+ self._logger.exception("Error during Xena JSON setup: %s", exc)
+ raise
+
+ def connect(self):
+ """Connect to the traffic generator.
+
+ This is an optional function, designed for traffic generators
+ which must be "connected to" (i.e. via SSH or an API) before
+ they can be used. If not required, simply do nothing here.
+
+ Where implemented, this function should raise an exception on
+ failure.
+
+ :returns: None
+ """
+ pass
+
+ def disconnect(self):
+ """Disconnect from the traffic generator.
+
+ As with :func:`connect`, this function is optional.
+
+ Where implemented, this function should raise an exception on
+ failure.
+
+ :returns: None
+ """
+ pass
+
+ 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``.
+
+ 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
+
+ :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')
+
+ def send_cont_traffic(self, traffic=None, duration=20):
+ """Send a continuous flow of traffic.r
+
+ Send packets at ``framerate``, using ``traffic`` configuration,
+ until timeout ``time`` occurs.
+
+ :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
+ :param duration: Time to wait to receive packets (secs)
+ :returns: dictionary of strings with following data:
+ - Tx Throughput (fps),
+ - Rx Throughput (fps),
+ - Tx Throughput (mbps),
+ - Rx Throughput (mbps),
+ - Tx Throughput (% linerate),
+ - Rx Throughput (% linerate),
+ - Min Latency (ns),
+ - Max Latency (ns),
+ - Avg Latency (ns)
+ """
+ raise NotImplementedError('Xena continuous traffic not implemented')
+
+ def start_cont_traffic(self, traffic=None, duration=20):
+ """Non-blocking version of 'send_cont_traffic'.
+
+ Start transmission and immediately return. Do not wait for
+ results.
+ :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
+ :param duration: Time to wait to receive packets (secs)
+ """
+ raise NotImplementedError('Xena continuous traffic not implemented')
+
+ def stop_cont_traffic(self):
+ """Stop continuous transmission and return results.
+ """
+ raise NotImplementedError('Xena continuous traffic not implemented')
+
+ def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
+ lossrate=0.0):
+ """Send traffic per RFC2544 throughput test specifications.
+
+ See ITrafficGenerator for description
+ """
+ self._duration = duration
+
+ self._params.clear()
+ self._params['traffic'] = self.traffic_defaults.copy()
+ if traffic:
+ self._params['traffic'] = merge_spec(self._params['traffic'],
+ traffic)
+
+ self._setup_json_config(trials, lossrate, '2544_throughput')
+
+ args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
+ "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
+ "./tools/pkt_gen/xena", "-u",
+ settings.getValue('TRAFFICGEN_XENA_USER')]
+ self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
+ self.mono_pipe.communicate()
+ root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ return Xena._create_throughput_result(root)
+
+ def start_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
+ lossrate=0.0):
+ """Non-blocking version of 'send_rfc2544_throughput'.
+
+ See ITrafficGenerator for description
+ """
+ self._duration = duration
+ self._params.clear()
+ self._params['traffic'] = self.traffic_defaults.copy()
+ if traffic:
+ self._params['traffic'] = merge_spec(self._params['traffic'],
+ traffic)
+
+ self._setup_json_config(trials, lossrate, '2544_throughput')
+
+ args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
+ "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
+ "./tools/pkt_gen/xena", "-u",
+ settings.getValue('TRAFFICGEN_XENA_USER')]
+ self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
+
+ def wait_rfc2544_throughput(self):
+ """Wait for and return results of RFC2544 test.
+
+ See ITrafficGenerator for description
+ """
+ self.mono_pipe.communicate()
+ sleep(2)
+ root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
+ return Xena._create_throughput_result(root)
+
+ def send_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
+ lossrate=0.0):
+ """Send traffic per RFC2544 back2back test specifications.
+
+ Send packets at a fixed rate, using ``traffic``
+ configuration, until minimum time at which no packet loss is
+ detected is found.
+
+ :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN
+ tags
+ :param trials: Number of trials to execute
+ :param duration: Per iteration duration
+ :param lossrate: Acceptable loss percentage
+
+ :returns: Named tuple of Rx Throughput (fps), Rx Throughput (mbps),
+ Tx Rate (% linerate), Rx Rate (% linerate), Tx Count (frames),
+ Back to Back Count (frames), Frame Loss (frames), Frame Loss (%)
+ :rtype: :class:`Back2BackResult`
+ """
+ raise NotImplementedError('Xena back2back not implemented')
+
+ def start_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
+ lossrate=0.0):
+ """Non-blocking version of 'send_rfc2544_back2back'.
+
+ Start transmission and immediately return. Do not wait for
+ results.
+ """
+ raise NotImplementedError('Xena back2back not implemented')
+
+ def wait_rfc2544_back2back(self):
+ """Wait and set results of RFC2544 test.
+ """
+ raise NotImplementedError('Xena back2back not implemented')
+
+
+if __name__ == "__main__":
+ pass
+
diff --git a/tools/pkt_gen/xena/xena_json.py b/tools/pkt_gen/xena/xena_json.py
new file mode 100644
index 00000000..39cc56c8
--- /dev/null
+++ b/tools/pkt_gen/xena/xena_json.py
@@ -0,0 +1,528 @@
+# Copyright 2016 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
+"""
+
+import base64
+from collections import OrderedDict
+import json
+import locale
+import logging
+import uuid
+
+import scapy.layers.inet as inet
+
+_LOGGER = logging.getLogger(__name__)
+_LOCALE = locale.getlocale()[1]
+
+
+class XenaJSON(object):
+ """
+ Class to modify and read Xena JSON configuration files.
+ """
+ def __init__(self, json_path='./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.packet_data = OrderedDict()
+ self.packet_data['layer2'] = None
+ self.packet_data['vlan'] = None
+ self.packet_data['layer3'] = None
+ self.packet_data['layer4'] = None
+
+ def _add_multistream_layer(self, entity, seg_uuid, stop_value, layer):
+ """
+ Add the multi stream layers to the json file based on the layer provided
+ :param entity: Entity to append the segment to in entity list
+ :param seg_uuid: The UUID to attach the multistream layer to
+ :param stop_value: The number of flows to configure
+ :param layer: the layer that the multistream will be attached to
+ :return: None
+ """
+ field_name = {
+ 2: ('Dst MAC addr', 'Src MAC addr'),
+ 3: ('Dest IP Addr', 'Src IP Addr'),
+ 4: ('Dest Port', 'Src Port')
+ }
+ segments = [
+ {
+ "Offset": 0,
+ "Mask": "//8=", # mask of 255/255
+ "Action": "INC",
+ "StartValue": 0,
+ "StopValue": stop_value,
+ "StepValue": 1,
+ "RepeatCount": 1,
+ "SegmentId": seg_uuid,
+ "FieldName": field_name[int(layer)][0]
+ },
+ {
+ "Offset": 0,
+ "Mask": "//8=", # mask of 255/255
+ "Action": "INC",
+ "StartValue": 0,
+ "StopValue": stop_value,
+ "StepValue": 1,
+ "RepeatCount": 1,
+ "SegmentId": seg_uuid,
+ "FieldName": field_name[int(layer)][1]
+ }
+ ]
+
+ self.json_data['StreamProfileHandler']['EntityList'][entity][
+ 'StreamConfig']['HwModifiers'].append(segments)
+
+ def _create_packet_header(self):
+ """
+ Create the scapy packet header based on what has been built in this
+ instance using the set header methods. Return tuple of the two byte
+ arrays, one for each port.
+ :return: Scapy packet headers as bytearrays
+ """
+ if not self.packet_data['layer2']:
+ _LOGGER.warning('Using dummy info for layer 2 in Xena JSON file')
+ self.set_header_layer2()
+ packet1, packet2 = (self.packet_data['layer2'][0],
+ self.packet_data['layer2'][1])
+ for packet_header in list(self.packet_data.copy().values())[1:]:
+ if packet_header:
+ packet1 /= packet_header[0]
+ packet2 /= packet_header[1]
+ ret = (bytes(packet1), bytes(packet2))
+ return ret
+
+ def add_header_segments(self, flows=0, multistream_layer=None):
+ """
+ Build the header segments to write to the JSON file.
+ :param flows: Number of flows to configure for multistream if enabled
+ :param multistream_layer: layer to set multistream flows as string.
+ Acceptable values are L2, L3 or L4
+ :return: None
+ """
+ packet = self._create_packet_header()
+ segment1 = list()
+ segment2 = list()
+ header_pos = 0
+ 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))
+ 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))
+ segment2.append(seg)
+ if multistream_layer == 'L2' and flows > 0:
+ self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
+ stop_value=flows, layer=2)
+ header_pos = len(layer2)
+ if self.packet_data['vlan']:
+ # 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)))
+ 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))
+ segment1.append(seg)
+ if multistream_layer == 'L3' and flows > 0:
+ self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
+ stop_value=flows, layer=3)
+ # 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))
+ segment2.append(seg)
+ if multistream_layer == 'L3' and flows > 0:
+ self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
+ stop_value=flows, layer=3)
+ header_pos += len(layer3)
+ if self.packet_data['layer4']:
+ # 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))
+ segment1.append(seg)
+ if multistream_layer == 'L4' and flows > 0:
+ self._add_multistream_layer(entity=0, seg_uuid=seg['ItemID'],
+ stop_value=flows, layer=4)
+ # 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))
+ segment2.append(seg)
+ if multistream_layer == 'L4' and flows > 0:
+ self._add_multistream_layer(entity=1, seg_uuid=seg['ItemID'],
+ stop_value=flows, layer=4)
+ header_pos += len(layer4)
+
+ self.json_data['StreamProfileHandler']['EntityList'][0][
+ 'StreamConfig']['HeaderSegments'] = segment1
+ self.json_data['StreamProfileHandler']['EntityList'][1][
+ 'StreamConfig']['HeaderSegments'] = segment2
+
+ def disable_back2back_test(self):
+ """
+ Disable the rfc2544 back to back test
+ :return: None
+ """
+ self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
+ 'Enabled'] = 'false'
+
+ def disable_throughput_test(self):
+ """
+ Disable the rfc2544 throughput test
+ :return: None
+ """
+ self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
+ 'Enabled'] = 'false'
+
+ def enable_back2back_test(self):
+ """
+ Enable the rfc2544 back to back test
+ :return: None
+ """
+ self.json_data['TestOptions']['TestTypeOptionMap']['Back2Back'][
+ 'Enabled'] = 'true'
+
+ def enable_throughput_test(self):
+ """
+ Enable the rfc2544 throughput test
+ :return: None
+ """
+ self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
+ 'Enabled'] = 'true'
+
+ def set_chassis_info(self, hostname, pwd):
+ """
+ Set the chassis info
+ :param hostname: hostname as string of ip
+ :param pwd: password to chassis as string
+ :return: None
+ """
+ self.json_data['ChassisManager']['ChassisList'][0][
+ 'HostName'] = hostname
+ self.json_data['ChassisManager']['ChassisList'][0][
+ 'Password'] = pwd
+
+ def set_header_layer2(self, dst_mac='cc:cc:cc:cc:cc:cc',
+ src_mac='bb:bb:bb:bb:bb:bb', **kwargs):
+ """
+ Build a scapy Ethernet L2 objects inside instance packet_data structure
+ :param dst_mac: destination mac as string. Example "aa:aa:aa:aa:aa:aa"
+ :param src_mac: source mac as string. Example "bb:bb:bb:bb:bb:bb"
+ :param kwargs: Extra params per scapy usage.
+ :return: None
+ """
+ self.packet_data['layer2'] = [
+ inet.Ether(dst=dst_mac, src=src_mac, **kwargs),
+ inet.Ether(dst=src_mac, src=dst_mac, **kwargs)]
+
+ def set_header_layer3(self, src_ip='192.168.0.2', dst_ip='192.168.0.3',
+ protocol='UDP', **kwargs):
+ """
+ Build scapy IPV4 L3 objects inside instance packet_data structure
+ :param src_ip: source IP as string in dot notation format
+ :param dst_ip: destination IP as string in dot notation format
+ :param protocol: protocol for l4
+ :param kwargs: Extra params per scapy usage
+ :return: None
+ """
+ self.packet_data['layer3'] = [
+ inet.IP(src=src_ip, dst=dst_ip, proto=protocol.lower(), **kwargs),
+ inet.IP(src=dst_ip, dst=src_ip, proto=protocol.lower(), **kwargs)]
+
+ def set_header_layer4_udp(self, source_port, destination_port, **kwargs):
+ """
+ Build scapy UDP L4 objects inside instance packet_data structure
+ :param source_port: Source port as int
+ :param destination_port: Destination port as int
+ :param kwargs: Extra params per scapy usage
+ :return: None
+ """
+ self.packet_data['layer4'] = [
+ inet.UDP(sport=source_port, dport=destination_port, **kwargs),
+ inet.UDP(sport=source_port, dport=destination_port, **kwargs)]
+
+ def set_header_vlan(self, vlan_id=1, **kwargs):
+ """
+ Build a Dot1Q scapy object inside instance packet_data structure
+ :param vlan_id: The VLAN ID
+ :param kwargs: Extra params per scapy usage
+ :return: None
+ """
+ self.packet_data['vlan'] = [
+ inet.Dot1Q(vlan=vlan_id, **kwargs),
+ inet.Dot1Q(vlan=vlan_id, **kwargs)]
+
+ def set_port(self, index, module, port):
+ """
+ Set the module and port for the 0 index port to use with the test
+ :param index: Index of port to set, 0 = port1, 1=port2, etc..
+ :param module: module location as int
+ :param port: port location in module as int
+ :return: None
+ """
+ self.json_data['PortHandler']['EntityList'][index]['PortRef'][
+ 'ModuleIndex'] = module
+ self.json_data['PortHandler']['EntityList'][index]['PortRef'][
+ 'PortIndex'] = port
+
+ def set_test_options(self, packet_sizes, duration, iterations, loss_rate,
+ micro_tpld=False):
+ """
+ Set the 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 loss_rate: acceptable loss rate as float
+ :param micro_tpld: boolean if micro_tpld should be enabled or disabled
+ :return: None
+ """
+ if isinstance(packet_sizes, int):
+ packet_sizes = [packet_sizes]
+ self.json_data['TestOptions']['PacketSizes'][
+ 'CustomPacketSizes'] = packet_sizes
+ self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
+ 'Duration'] = duration
+ self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
+ 'RateIterationOptions']['AcceptableLoss'] = loss_rate
+ self.json_data['TestOptions']['FlowCreationOptions'][
+ 'UseMicroTpldOnDemand'] = 'true' if micro_tpld else 'false'
+ self.json_data['TestOptions']['TestTypeOptionMap']['Throughput'][
+ 'Iterations'] = iterations
+
+ 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):
+ 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: {}/{} 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_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(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)
+