fi
}
+get_external_libs() {
+ cd $(dirname ${BASH_SOURCE[0]})
+ TREX_DOWNLOAD="https://trex-tgn.cisco.com/trex/release/v2.05.tar.gz"
+ TREX_DIR=$PWD/trex/scripts
+ if [ ! -d "$TREX_DIR" ]; then
+ rm -rf ${TREX_DOWNLOAD##*/}
+ if [ ! -e ${TREX_DOWNLOAD##*/} ] ; then
+ wget $TREX_DOWNLOAD
+ fi
+ tar zxvf ${TREX_DOWNLOAD##*/}
+ pushd .
+ rm -rf trex && mkdir -p trex
+ mv v2.05 trex/scripts
+ rm -rf v2.05.tar.gz
+ touch "$PWD/trex/scripts/automation/trex_control_plane/stl/__init__.py"
+ popd
+ fi
+ echo "Done."
+ export PYTHONPATH=$PYTHONPATH:"$PWD/trex/scripts/automation/trex_control_plane"
+ export PYTHONPATH=$PYTHONPATH:"$PWD/trex/scripts/automation/trex_control_plane/stl"
+ echo $PYTHONPATH
+}
+
run_tests() {
+ echo "Get external libs needed for unit test"
+ get_external_libs
+
echo "Running unittest ... "
if [ $FILE_OPTION == "f" ]; then
python -m unittest discover -v -s tests/unit > $logfile 2>&1
self._get_file_abspath("ipv4_1flow_Packets_vpe.yaml")
self.scenario_cfg["traffic_options"]["imix"] = \
self._get_file_abspath("imix_voice.yaml")
- self.assertRaises(RuntimeError, self.s._fill_traffic_profile,
- self.scenario_cfg, self.context_cfg)
+ self.assertIsNotNone(self.s._fill_traffic_profile(self.scenario_cfg,
+ self.context_cfg))
def test_teardown(self):
vnf = mock.Mock(autospec=GenericVNF)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2016-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.
+#
+
+from __future__ import absolute_import
+import unittest
+
+from yardstick.network_services.traffic_profile.base import TrafficProfile
+from yardstick.network_services.traffic_profile.http import \
+ TrafficProfileGenericHTTP
+
+
+class TestTrafficProfileGenericHTTP(unittest.TestCase):
+ def test___init__(self):
+ traffic_profile_generic_htt_p = \
+ TrafficProfileGenericHTTP(TrafficProfile)
+ self.assertIsNotNone(traffic_profile_generic_htt_p)
+
+ def test_execute(self):
+ traffic_profile_generic_htt_p = \
+ TrafficProfileGenericHTTP(TrafficProfile)
+ traffic_generator = {}
+ self.assertEqual(None,
+ traffic_profile_generic_htt_p.execute(
+ traffic_generator))
+
+ def test__send_http_request(self):
+ traffic_profile_generic_htt_p = \
+ TrafficProfileGenericHTTP(TrafficProfile)
+ self.assertEqual(None,
+ traffic_profile_generic_htt_p._send_http_request(
+ "10.1.1.1", "250", "/req"))
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2016-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.
+#
+
+from __future__ import absolute_import
+import unittest
+import mock
+
+from yardstick.network_services.traffic_profile.traffic_profile \
+ import TrexProfile
+from yardstick.network_services.traffic_profile.rfc2544 import \
+ RFC2544Profile
+
+
+class TestRFC2544Profile(unittest.TestCase):
+ TRAFFIC_PROFILE = {
+ "schema": "isb:traffic_profile:0.1",
+ "name": "fixed",
+ "description": "Fixed traffic profile to run UDP traffic",
+ "traffic_profile": {
+ "traffic_type": "FixedTraffic",
+ "frame_rate": 100, # pps
+ "flow_number": 10,
+ "frame_size": 64}}
+
+ PROFILE = {'description': 'Traffic profile to run RFC2544 latency',
+ 'name': 'rfc2544',
+ 'traffic_profile': {'traffic_type': 'RFC2544Profile',
+ 'frame_rate': 100},
+ 'public': {'ipv4':
+ {'outer_l2': {'framesize':
+ {'64B': '100', '1518B': '0',
+ '128B': '0', '1400B': '0',
+ '256B': '0', '373b': '0',
+ '570B': '0'}},
+ 'outer_l3v4': {'dstip4': '1.1.1.1-1.15.255.255',
+ 'proto': 'udp',
+ 'srcip4': '90.90.1.1-90.105.255.255',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'srcport': '2001',
+ 'dsrport': '1234'}}},
+ 'private': {'ipv4':
+ {'outer_l2': {'framesize':
+ {'64B': '100', '1518B': '0',
+ '128B': '0', '1400B': '0',
+ '256B': '0', '373b': '0',
+ '570B': '0'}},
+ 'outer_l3v4': {'dstip4': '9.9.1.1-90.105.255.255',
+ 'proto': 'udp',
+ 'srcip4': '1.1.1.1-1.15.255.255',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'dstport': '2001',
+ 'srcport': '1234'}}},
+ 'schema': 'isb:traffic_profile:0.1'}
+
+ def test___init__(self):
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ assert r_f_c2544_profile.rate
+
+ def test_execute(self):
+ traffic_generator = mock.Mock(autospec=TrexProfile)
+ traffic_generator.my_ports = [0, 1]
+ traffic_generator.client = \
+ mock.Mock(return_value=True)
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ r_f_c2544_profile.params = self.PROFILE
+ r_f_c2544_profile.first_run = True
+ self.assertEqual(None, r_f_c2544_profile.execute(traffic_generator))
+
+ def test_get_drop_percentage(self):
+ traffic_generator = mock.Mock(autospec=TrexProfile)
+ traffic_generator.my_ports = [0, 1]
+ traffic_generator.client = \
+ mock.Mock(return_value=True)
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ r_f_c2544_profile.params = self.PROFILE
+ self.assertEqual(None, r_f_c2544_profile.execute(traffic_generator))
+ samples = {}
+ for ifname in range(1):
+ name = "xe{}".format(ifname)
+ samples[name] = {"rx_throughput_fps": 20,
+ "tx_throughput_fps": 20,
+ "rx_throughput_mbps": 10,
+ "tx_throughput_mbps": 10,
+ "in_packets": 1000,
+ "out_packets": 1000}
+ tol_min = 100.0
+ tolerance = 0.0
+ expected = {'DropPercentage': 0.0, 'RxThroughput': 33,
+ 'TxThroughput': 33, 'CurrentDropPercentage': 0.0,
+ 'Throughput': 33,
+ 'xe0': {'tx_throughput_fps': 20, 'in_packets': 1000,
+ 'out_packets': 1000, 'rx_throughput_mbps': 10,
+ 'tx_throughput_mbps': 10, 'rx_throughput_fps': 20}}
+ self.assertDictEqual(expected,
+ r_f_c2544_profile.get_drop_percentage(
+ traffic_generator, samples,
+ tol_min, tolerance))
+
+ def test_get_drop_percentage_update(self):
+ traffic_generator = mock.Mock(autospec=TrexProfile)
+ traffic_generator.my_ports = [0, 1]
+ traffic_generator.client = \
+ mock.Mock(return_value=True)
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ r_f_c2544_profile.params = self.PROFILE
+ self.assertEqual(None, r_f_c2544_profile.execute(traffic_generator))
+ samples = {}
+ for ifname in range(1):
+ name = "xe{}".format(ifname)
+ samples[name] = {"rx_throughput_fps": 20,
+ "tx_throughput_fps": 20,
+ "rx_throughput_mbps": 10,
+ "tx_throughput_mbps": 10,
+ "in_packets": 1000,
+ "out_packets": 1002}
+ tol_min = 0.0
+ tolerance = 1.0
+ expected = {'DropPercentage': 0.2, 'RxThroughput': 33,
+ 'TxThroughput': 33, 'CurrentDropPercentage': 0.2,
+ 'Throughput': 33,
+ 'xe0': {'tx_throughput_fps': 20, 'in_packets': 1000,
+ 'out_packets': 1002, 'rx_throughput_mbps': 10,
+ 'tx_throughput_mbps': 10, 'rx_throughput_fps': 20}}
+ self.assertDictEqual(expected,
+ r_f_c2544_profile.get_drop_percentage(
+ traffic_generator, samples,
+ tol_min, tolerance))
+
+ def test_get_drop_percentage_div_zero(self):
+ traffic_generator = mock.Mock(autospec=TrexProfile)
+ traffic_generator.my_ports = [0, 1]
+ traffic_generator.client = \
+ mock.Mock(return_value=True)
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ r_f_c2544_profile.params = self.PROFILE
+ self.assertEqual(None, r_f_c2544_profile.execute(traffic_generator))
+ samples = {}
+ for ifname in range(1):
+ name = "xe{}".format(ifname)
+ samples[name] = {"rx_throughput_fps": 20,
+ "tx_throughput_fps": 20,
+ "rx_throughput_mbps": 10,
+ "tx_throughput_mbps": 10,
+ "in_packets": 1000,
+ "out_packets": 0}
+ tol_min = 0.0
+ tolerance = 0.0
+ r_f_c2544_profile.tmp_throughput = 0
+ expected = {'DropPercentage': 100.0, 'RxThroughput': 33,
+ 'TxThroughput': 0, 'CurrentDropPercentage': 100.0,
+ 'Throughput': 33,
+ 'xe0': {'tx_throughput_fps': 20, 'in_packets': 1000,
+ 'out_packets': 0, 'rx_throughput_mbps': 10,
+ 'tx_throughput_mbps': 10, 'rx_throughput_fps': 20}}
+ self.assertDictEqual(expected,
+ r_f_c2544_profile.get_drop_percentage(
+ traffic_generator, samples,
+ tol_min, tolerance))
+
+ def test_get_multiplier(self):
+ r_f_c2544_profile = RFC2544Profile(self.TRAFFIC_PROFILE)
+ r_f_c2544_profile.max_rate = 100
+ r_f_c2544_profile.min_rate = 100
+ self.assertEqual("1.0", r_f_c2544_profile.get_multiplier())
+
+if __name__ == '__main__':
+ unittest.main()
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2016-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.
+#
+
+from __future__ import absolute_import
+import unittest
+
+from yardstick.network_services.traffic_profile.base import TrafficProfile
+from yardstick.network_services.traffic_profile.traffic_profile import \
+ TrexProfile
+
+
+class TestTrexProfile(unittest.TestCase):
+ TRAFFIC_PROFILE = {
+ "schema": "isb:traffic_profile:0.1",
+ "name": "fixed",
+ "description": "Fixed traffic profile to run UDP traffic",
+ "traffic_profile": {
+ "traffic_type": "FixedTraffic",
+ "frame_rate": 100, # pps
+ "flow_number": 10,
+ "frame_size": 64}}
+
+ PROFILE = {'description': 'Traffic profile to run RFC2544 latency',
+ 'name': 'rfc2544',
+ 'traffic_profile': {'traffic_type': 'RFC2544Profile',
+ 'frame_rate': 100},
+ 'public': {'ipv4': {'outer_l2': {'framesize': {'64B': '100',
+ '1518B': '0',
+ '128B': '0',
+ '1400B': '0',
+ '256B': '0',
+ '373b': '0',
+ '570B': '0'},
+ "srcmac": "00:00:00:00:00:02",
+ "dstmac": "00:00:00:00:00:01"},
+ 'outer_l3v4': {'dstip4': '1.1.1.1-1.1.2.2',
+ 'proto': 'udp',
+ 'srcip4': '9.9.1.1-90.1.2.2',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'srcport': '2001',
+ 'dsrport': '1234'}}},
+ 'private': {'ipv4':
+ {'outer_l2': {'framesize':
+ {'64B': '100', '1518B': '0',
+ '128B': '0', '1400B': '0',
+ '256B': '0', '373b': '0',
+ '570B': '0'},
+ "srcmac": "00:00:00:00:00:01",
+ "dstmac": "00:00:00:00:00:02"},
+ 'outer_l3v4': {'dstip4': '9.9.1.1-90.105.255.255',
+ 'proto': 'udp',
+ 'srcip4': '1.1.1.1-1.15.255.255',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'dstport': '2001',
+ 'srcport': '1234'}}},
+ 'schema': 'isb:traffic_profile:0.1'}
+ PROFILE_v6 = {'description': 'Traffic profile to run RFC2544 latency',
+ 'name': 'rfc2544',
+ 'traffic_profile': {'traffic_type': 'RFC2544Profile',
+ 'frame_rate': 100},
+ 'public': {'ipv6': {'outer_l2': {'framesize':
+ {'64B': '100', '1518B': '0',
+ '128B': '0', '1400B': '0',
+ '256B': '0', '373b': '0',
+ '570B': '0'},
+ "srcmac": "00:00:00:00:00:02",
+ "dstmac": "00:00:00:00:00:01"},
+ 'outer_l3v4': {'dstip6': '0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420',
+ 'proto': 'udp',
+ 'srcip6': '0064:ff9b:0:0:0:0:9810:2814-0064:ff9b:0:0:0:0:9810:2820',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'srcport': '2001',
+ 'dsrport': '1234'}}},
+ 'private':
+ {'ipv6': {'outer_l2': {'framesize':
+ {'64B': '100', '1518B': '0',
+ '128B': '0', '1400B': '0',
+ '256B': '0', '373b': '0',
+ '570B': '0'},
+ "srcmac": "00:00:00:00:00:01",
+ "dstmac": "00:00:00:00:00:02"},
+ 'outer_l3v4': {'dstip6': '0064:ff9b:0:0:0:0:9810:2814-0064:ff9b:0:0:0:0:9810:2820',
+ 'proto': 'udp',
+ 'srcip6': '0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420',
+ 'dscp': 0, 'ttl': 32},
+ 'outer_l4': {'dstport': '2001',
+ 'srcport': '1234'}}},
+ 'schema': 'isb:traffic_profile:0.1'}
+
+ def test___init__(self):
+ TrafficProfile.params = self.PROFILE
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(trex_profile.pps, 100)
+
+ def test_execute(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.execute({}))
+
+ def test_set_src_mac(self):
+ src_mac = "00:00:00:00:00:01"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_src_mac(src_mac))
+
+ src_mac = "00:00:00:00:00:01-00:00:00:00:00:02"
+ self.assertEqual(None, trex_profile.set_src_mac(src_mac))
+
+ def test_set_dst_mac(self):
+ dst_mac = "00:00:00:00:00:03"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_dst_mac(dst_mac))
+
+ dst_mac = "00:00:00:00:00:03-00:00:00:00:00:04"
+ self.assertEqual(None, trex_profile.set_dst_mac(dst_mac))
+
+ def test_set_src_ip4(self):
+ src_ipv4 = "152.16.100.20"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_src_ip4(src_ipv4))
+
+ src_ipv4 = "152.16.100.20-152.16.100.30"
+ self.assertEqual(None, trex_profile.set_src_ip4(src_ipv4))
+
+ def test_set_dst_ip4(self):
+ dst_ipv4 = "152.16.100.20"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_dst_ip4(dst_ipv4))
+
+ dst_ipv4 = "152.16.100.20-152.16.100.30"
+ self.assertEqual(None, trex_profile.set_dst_ip4(dst_ipv4))
+
+ def test_set_src_ip6(self):
+ src_ipv6 = "0064:ff9b:0:0:0:0:9810:6414"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_src_ip6(src_ipv6))
+
+ src_ipv6 = "0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420"
+ self.assertEqual(None, trex_profile.set_src_ip6(src_ipv6))
+
+ def test_set_dst_ip6(self):
+ dst_ipv6 = "0064:ff9b:0:0:0:0:9810:6414"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_dst_ip6(dst_ipv6))
+
+ dst_ipv6 = "0064:ff9b:0:0:0:0:9810:6414-0064:ff9b:0:0:0:0:9810:6420"
+ self.assertEqual(None, trex_profile.set_dst_ip6(dst_ipv6))
+
+ def test_dscp(self):
+ dscp = "0"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_dscp(dscp))
+
+ dscp = "0-1"
+ self.assertEqual(None, trex_profile.set_dscp(dscp))
+
+ def test_src_port(self):
+ port = "1234"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_src_port(port))
+
+ port = "1234-5678"
+ self.assertEqual(None, trex_profile.set_src_port(port))
+
+ def test_dst_port(self):
+ port = "1234"
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_dst_port(port))
+
+ port = "1234-5678"
+ self.assertEqual(None, trex_profile.set_dst_port(port))
+
+ def test_qinq(self):
+ qinq = {"S-VLAN": {"id": 128, "priority": 0, "cfi": 0},
+ "C-VLAN": {"id": 512, "priority": 0, "cfi": 0}}
+
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual(None, trex_profile.set_qinq(qinq))
+
+ qinq = {"S-VLAN": {"id": "128-130", "priority": 0, "cfi": 0},
+ "C-VLAN": {"id": "512-515", "priority": 0, "cfi": 0}}
+ self.assertEqual(None, trex_profile.set_qinq(qinq))
+
+ def test_set_outer_l2_fields(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ qinq = {"S-VLAN": {"id": 128, "priority": 0, "cfi": 0},
+ "C-VLAN": {"id": 512, "priority": 0, "cfi": 0}}
+ outer_l2 = self.PROFILE['private']['ipv4']['outer_l2']
+ outer_l2['QinQ'] = qinq
+ self.assertEqual(None, trex_profile.set_outer_l2_fields(outer_l2))
+
+ def test_set_outer_l3v4_fields(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ outer_l3v4 = self.PROFILE['private']['ipv4']['outer_l3v4']
+ outer_l3v4['proto'] = 'tcp'
+ self.assertEqual(None, trex_profile.set_outer_l3v4_fields(outer_l3v4))
+
+ def test_set_outer_l3v6_fields(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ outer_l3v6 = self.PROFILE_v6['private']['ipv6']['outer_l3v4']
+ outer_l3v6['proto'] = 'tcp'
+ outer_l3v6['tc'] = 1
+ outer_l3v6['hlim'] = 10
+ self.assertEqual(None, trex_profile.set_outer_l3v6_fields(outer_l3v6))
+
+ def test_set_outer_l4_fields(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ outer_l4 = self.PROFILE['private']['ipv4']['outer_l4']
+ self.assertEqual(None, trex_profile.set_outer_l4_fields(outer_l4))
+
+ def test_get_streams(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ trex_profile.params = self.PROFILE
+ trex_profile.profile_data = self.PROFILE["private"]
+ self.assertIsNotNone(trex_profile.get_streams())
+ trex_profile.pg_id = 1
+ self.assertIsNotNone(trex_profile.get_streams())
+ trex_profile.params = self.PROFILE_v6
+ trex_profile.profile_data = self.PROFILE_v6["private"]
+ self.assertIsNotNone(trex_profile.get_streams())
+ trex_profile.pg_id = 1
+ self.assertIsNotNone(trex_profile.get_streams())
+
+ def test_generate_packets(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ trex_profile.fsize = 10
+ trex_profile.base_pkt = [10]
+ self.assertIsNone(trex_profile.generate_packets())
+
+ def test_generate_imix_data_error(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertEqual({}, trex_profile.generate_imix_data(False))
+
+ def test__get_start_end_ipv6(self):
+ trex_profile = \
+ TrexProfile(TrafficProfile)
+ self.assertRaises(SystemExit, trex_profile._get_start_end_ipv6,
+ "1.1.1.3", "1.1.1.1")
--- /dev/null
+# Copyright (c) 2016-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.
+""" Generic HTTP profile used by different Traffic generators """
+
+from __future__ import absolute_import
+
+from yardstick.network_services.traffic_profile.base import TrafficProfile
+
+
+class TrafficProfileGenericHTTP(TrafficProfile):
+ """ This Class handles setup of generic http traffic profile """
+
+ def __init__(self, TrafficProfile):
+ super(TrafficProfileGenericHTTP, self).__init__(TrafficProfile)
+
+ def execute(self, traffic_generator):
+ ''' send run traffic for a selected traffic generator'''
+ pass
+
+ def _send_http_request(self, server, port, locator, **kwargs):
+ ''' send http request for a given server, port '''
+ pass
--- /dev/null
+# Copyright (c) 2016-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.
+""" RFC2544 Throughput implemenation """
+
+from __future__ import absolute_import
+import logging
+
+from yardstick.network_services.traffic_profile.traffic_profile \
+ import TrexProfile
+
+LOGGING = logging.getLogger(__name__)
+
+
+class RFC2544Profile(TrexProfile):
+ """ This class handles rfc2544 implemenation. """
+
+ def __init__(self, traffic_generator):
+ super(RFC2544Profile, self).__init__(traffic_generator)
+ self.max_rate = None
+ self.min_rate = None
+ self.rate = 100
+ self.tmp_drop = None
+ self.tmp_throughput = None
+ self.profile_data = None
+
+ def execute(self, traffic_generator):
+ ''' Generate the stream and run traffic on the given ports '''
+ if self.first_run:
+ self.profile_data = self.params.get('private', '')
+ ports = [traffic_generator.my_ports[0]]
+ traffic_generator.client.add_streams(self.get_streams(),
+ ports=ports[0])
+ profile_data = self.params.get('public', '')
+ if profile_data:
+ self.profile_data = profile_data
+ ports.append(traffic_generator.my_ports[1])
+ traffic_generator.client.add_streams(self.get_streams(),
+ ports=ports[1])
+
+ self.max_rate = self.rate
+ self.min_rate = 0
+ traffic_generator.client.start(ports=ports,
+ mult=self.get_multiplier(),
+ duration=30, force=True)
+ self.tmp_drop = 0
+ self.tmp_throughput = 0
+
+ def get_multiplier(self):
+ ''' Get the rate at which next iternation to run '''
+ self.rate = round((self.max_rate + self.min_rate) / 2.0, 2)
+ multiplier = round(self.rate / self.pps, 2)
+ return str(multiplier)
+
+ def get_drop_percentage(self, traffic_generator,
+ samples, tol_min, tolerance):
+ ''' Calculate the drop percentage and run the traffic '''
+ in_packets = sum([samples[iface]['in_packets'] for iface in samples])
+ out_packets = sum([samples[iface]['out_packets'] for iface in samples])
+ packet_drop = abs(out_packets - in_packets)
+ drop_percent = 100.0
+ try:
+ drop_percent = round((packet_drop / float(out_packets)) * 100, 2)
+ except ZeroDivisionError:
+ LOGGING.info('No traffic is flowing')
+ samples['TxThroughput'] = out_packets / 30
+ samples['RxThroughput'] = in_packets / 30
+ samples['CurrentDropPercentage'] = drop_percent
+ samples['Throughput'] = self.tmp_throughput
+ samples['DropPercentage'] = self.tmp_drop
+ if drop_percent > tolerance and self.tmp_throughput == 0:
+ samples['Throughput'] = (in_packets / 30)
+ samples['DropPercentage'] = drop_percent
+ if self.first_run:
+ max_supported_rate = out_packets / 30
+ self.rate = max_supported_rate
+ self.first_run = False
+ if drop_percent > tolerance:
+ self.max_rate = self.rate
+ elif drop_percent < tol_min:
+ self.min_rate = self.rate
+ if drop_percent >= self.tmp_drop:
+ self.tmp_drop = drop_percent
+ self.tmp_throughput = (in_packets / 30)
+ samples['Throughput'] = (in_packets / 30)
+ samples['DropPercentage'] = drop_percent
+ else:
+ samples['Throughput'] = (in_packets / 30)
+ samples['DropPercentage'] = drop_percent
+
+ traffic_generator.client.clear_stats(ports=traffic_generator.my_ports)
+ traffic_generator.client.start(ports=traffic_generator.my_ports,
+ mult=self.get_multiplier(),
+ duration=30, force=True)
+ return samples
--- /dev/null
+# Copyright (c) 2016-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.
+""" Trex Traffic Profile definitions """
+
+from __future__ import absolute_import
+import struct
+import socket
+import logging
+from random import SystemRandom
+import six
+
+from yardstick.network_services.traffic_profile.base import TrafficProfile
+from stl.trex_stl_lib.trex_stl_client import STLStream
+from stl.trex_stl_lib.trex_stl_streams import STLFlowLatencyStats
+from stl.trex_stl_lib.trex_stl_streams import STLTXCont
+from stl.trex_stl_lib.trex_stl_streams import STLProfile
+from stl.trex_stl_lib.trex_stl_packet_builder_scapy import STLVmWrFlowVar
+from stl.trex_stl_lib.trex_stl_packet_builder_scapy import STLVmFlowVar
+from stl.trex_stl_lib.trex_stl_packet_builder_scapy import STLPktBuilder
+from stl.trex_stl_lib.trex_stl_packet_builder_scapy import STLScVmRaw
+from stl.trex_stl_lib.trex_stl_packet_builder_scapy import STLVmFixIpv4
+from stl.trex_stl_lib import api as Pkt
+
+
+class TrexProfile(TrafficProfile):
+ """ This class handles Trex Traffic profile generation and execution """
+
+ def __init__(self, yaml_data):
+ super(TrexProfile, self).__init__(yaml_data)
+ self.flows = 100
+ self.pps = 100
+ self.pg_id = 0
+ self.first_run = True
+ self.streams = 1
+ self.profile_data = []
+ self.profile = None
+ self.base_pkt = None
+ self.fsize = None
+ self.trex_vm = None
+ self.vms = []
+ self.rate = None
+ self.ip_packet = None
+ self.ip6_packet = None
+ self.udp_packet = None
+ self.udp_dport = ''
+ self.udp_sport = ''
+ self.qinq_packet = None
+ self.qinq = False
+ self.vm_flow_vars = []
+ self.packets = []
+ self.ether_packet = []
+
+ def execute(self, traffic_generator):
+ """ Generate the stream and run traffic on the given ports """
+ pass
+
+ def _set_ether_fields(self, **kwargs):
+ """ set ethernet protocol fields """
+ if not self.ether_packet:
+ self.ether_packet = Pkt.Ether()
+ for key, value in six.iteritems(kwargs):
+ setattr(self.ether_packet, key, value)
+
+ def _set_ip_fields(self, **kwargs):
+ """ set l3 ipv4 protocol fields """
+
+ if not self.ip_packet:
+ self.ip_packet = Pkt.IP()
+ for key in kwargs:
+ setattr(self.ip_packet, key, kwargs[key])
+
+ def _set_ip6_fields(self, **kwargs):
+ """ set l3 ipv6 protocol fields """
+ if not self.ip6_packet:
+ self.ip6_packet = Pkt.IPv6()
+ for key in kwargs:
+ setattr(self.ip6_packet, key, kwargs[key])
+
+ def _set_udp_fields(self, **kwargs):
+ """ set l4 udp ports fields """
+ if not self.udp_packet:
+ self.udp_packet = Pkt.UDP()
+ for key in kwargs:
+ setattr(self.udp_packet, key, kwargs[key])
+
+ def set_src_mac(self, src_mac):
+ """ set source mac address fields """
+ src_macs = src_mac.split('-')
+ min_value = src_macs[0]
+ if len(src_macs) == 1:
+ src_mac = min_value
+ self._set_ether_fields(src=src_mac)
+ else:
+ stl_vm_flow_var = STLVmFlowVar(name="mac_src",
+ min_value=1,
+ max_value=30,
+ size=4,
+ op='inc',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='mac_src',
+ pkt_offset='Ether.src')
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_dst_mac(self, dst_mac):
+ """ set destination mac address fields """
+ dst_macs = dst_mac.split('-')
+ min_value = dst_macs[0]
+ if len(dst_macs) == 1:
+ dst_mac = min_value
+ self._set_ether_fields(dst=dst_mac)
+ else:
+ stl_vm_flow_var = STLVmFlowVar(name="mac_dst",
+ min_value=1,
+ max_value=30,
+ size=4,
+ op='inc',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='mac_dst',
+ pkt_offset='Ether.dst')
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_src_ip4(self, src_ip4):
+ """ set source ipv4 address fields """
+ src_ips = src_ip4.split('-')
+ min_value = src_ips[0]
+ max_value = src_ips[1] if len(src_ips) == 2 else src_ips[0]
+ if len(src_ips) == 1:
+ src_ip4 = min_value
+ self._set_ip_fields(src=src_ip4)
+ else:
+ stl_vm_flow_var = STLVmFlowVar(name="ip4_src",
+ min_value=min_value,
+ max_value=max_value,
+ size=4,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='ip4_src',
+ pkt_offset='IP.src')
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+ stl_vm_fix_ipv4 = STLVmFixIpv4(offset="IP")
+ self.vm_flow_vars.append(stl_vm_fix_ipv4)
+
+ def set_dst_ip4(self, dst_ip4):
+ """ set destination ipv4 address fields """
+ dst_ips = dst_ip4.split('-')
+ min_value = dst_ips[0]
+ max_value = dst_ips[1] if len(dst_ips) == 2 else dst_ips[0]
+ if len(dst_ips) == 1:
+ dst_ip4 = min_value
+ self._set_ip_fields(dst=dst_ip4)
+ else:
+ stl_vm_flow_var = STLVmFlowVar(name="dst_ip4",
+ min_value=min_value,
+ max_value=max_value,
+ size=4,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='dst_ip4',
+ pkt_offset='IP.dst')
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+ stl_vm_fix_ipv4 = STLVmFixIpv4(offset="IP")
+ self.vm_flow_vars.append(stl_vm_fix_ipv4)
+
+ def set_src_ip6(self, src_ip6):
+ """ set source ipv6 address fields """
+ src_ips = src_ip6.split('-')
+ min_value = src_ips[0]
+ max_value = src_ips[1] if len(src_ips) == 2 else src_ips[0]
+ src_ip6 = min_value
+ self._set_ip6_fields(src=src_ip6)
+ if len(src_ips) == 2:
+ min_value, max_value = \
+ self._get_start_end_ipv6(min_value, max_value)
+ stl_vm_flow_var = STLVmFlowVar(name="ip6_src",
+ min_value=min_value,
+ max_value=max_value,
+ size=8,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='ip6_src',
+ pkt_offset='IPv6.src',
+ offset_fixup=8)
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_dst_ip6(self, dst_ip6):
+ """ set destination ipv6 address fields """
+ dst_ips = dst_ip6.split('-')
+ min_value = dst_ips[0]
+ max_value = dst_ips[1] if len(dst_ips) == 2 else dst_ips[0]
+ dst_ip6 = min_value
+ self._set_ip6_fields(dst=dst_ip6)
+ if len(dst_ips) == 2:
+ min_value, max_value = \
+ self._get_start_end_ipv6(min_value, max_value)
+ stl_vm_flow_var = STLVmFlowVar(name="dst_ip6",
+ min_value=min_value,
+ max_value=max_value,
+ size=8,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='dst_ip6',
+ pkt_offset='IPv6.dst',
+ offset_fixup=8)
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_dscp(self, dscp):
+ """ set dscp for trex """
+ dscps = str(dscp).split('-')
+ min_value = int(dscps[0])
+ max_value = int(dscps[1]) if len(dscps) == 2 else int(dscps[0])
+ if len(dscps) == 1:
+ dscp = min_value
+ self._set_ip_fields(tos=dscp)
+ else:
+ stl_vm_flow_var = STLVmFlowVar(name="dscp",
+ min_value=min_value,
+ max_value=max_value,
+ size=2,
+ op='inc',
+ step=8)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='dscp',
+ pkt_offset='IP.tos')
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_src_port(self, src_port):
+ """ set packet source port """
+ src_ports = str(src_port).split('-')
+ min_value = int(src_ports[0])
+ if len(src_ports) == 1:
+ max_value = int(src_ports[0])
+ src_port = min_value
+ self._set_udp_fields(sport=src_port)
+ else:
+ max_value = int(src_ports[1])
+ stl_vm_flow_var = STLVmFlowVar(name="port_src",
+ min_value=min_value,
+ max_value=max_value,
+ size=2,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='port_src',
+ pkt_offset=self.udp_sport)
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_dst_port(self, dst_port):
+ """ set packet destnation port """
+ dst_ports = str(dst_port).split('-')
+ min_value = int(dst_ports[0])
+ if len(dst_ports) == 1:
+ max_value = int(dst_ports[0])
+ dst_port = min_value
+ self._set_udp_fields(dport=dst_port)
+ else:
+ max_value = int(dst_ports[1])
+ stl_vm_flow_var = STLVmFlowVar(name="port_dst",
+ min_value=min_value,
+ max_value=max_value,
+ size=2,
+ op='random',
+ step=1)
+ self.vm_flow_vars.append(stl_vm_flow_var)
+ stl_vm_wr_flow_var = STLVmWrFlowVar(fv_name='port_dst',
+ pkt_offset=self.udp_dport)
+ self.vm_flow_vars.append(stl_vm_wr_flow_var)
+
+ def set_svlan_cvlan(self, svlan, cvlan):
+ """ set svlan & cvlan """
+ self.qinq = True
+ ether_params = {'type': 0x8100}
+ self._set_ether_fields(**ether_params)
+ svlans = str(svlan['id']).split('-')
+ svlan_min = int(svlans[0])
+ svlan_max = int(svlans[1]) if len(svlans) == 2 else int(svlans[0])
+ if len(svlans) == 2:
+ svlan = self._get_random_value(svlan_min, svlan_max)
+ else:
+ svlan = svlan_min
+ cvlans = str(cvlan['id']).split('-')
+ cvlan_min = int(cvlans[0])
+ cvlan_max = int(cvlans[1]) if len(cvlans) == 2 else int(cvlans[0])
+ if len(cvlans) == 2:
+ cvlan = self._get_random_value(cvlan_min, cvlan_max)
+ else:
+ cvlan = cvlan_min
+ self.qinq_packet = Pkt.Dot1Q(vlan=svlan) / Pkt.Dot1Q(vlan=cvlan)
+
+ def set_qinq(self, qinq):
+ """ set qinq in packet """
+ self.set_svlan_cvlan(qinq['S-VLAN'], qinq['C-VLAN'])
+
+ def set_outer_l2_fields(self, outer_l2):
+ """ setup outer l2 fields from traffic profile """
+ ether_params = {'type': 0x800}
+ self._set_ether_fields(**ether_params)
+ if 'srcmac' in outer_l2:
+ self.set_src_mac(outer_l2['srcmac'])
+ if 'dstmac' in outer_l2:
+ self.set_dst_mac(outer_l2['dstmac'])
+ if 'QinQ' in outer_l2:
+ self.set_qinq(outer_l2['QinQ'])
+
+ def set_outer_l3v4_fields(self, outer_l3v4):
+ """ setup outer l3v4 fields from traffic profile """
+ ip_params = {}
+ if 'proto' in outer_l3v4:
+ ip_params['proto'] = outer_l3v4['proto']
+ if outer_l3v4['proto'] == 'tcp':
+ self.udp_packet = Pkt.TCP()
+ self.udp_dport = 'TCP.dport'
+ self.udp_sport = 'TCP.sport'
+ tcp_params = {'flags': '', 'window': 0}
+ self._set_udp_fields(**tcp_params)
+ if 'ttl' in outer_l3v4:
+ ip_params['ttl'] = outer_l3v4['ttl']
+ self._set_ip_fields(**ip_params)
+ if 'dscp' in outer_l3v4:
+ self.set_dscp(outer_l3v4['dscp'])
+ if 'srcip4' in outer_l3v4:
+ self.set_src_ip4(outer_l3v4['srcip4'])
+ if 'dstip4' in outer_l3v4:
+ self.set_dst_ip4(outer_l3v4['dstip4'])
+
+ def set_outer_l3v6_fields(self, outer_l3v6):
+ """ setup outer l3v6 fields from traffic profile """
+ ether_params = {'type': 0x86dd}
+ self._set_ether_fields(**ether_params)
+ ip6_params = {}
+ if 'proto' in outer_l3v6:
+ ip6_params['proto'] = outer_l3v6['proto']
+ if outer_l3v6['proto'] == 'tcp':
+ self.udp_packet = Pkt.TCP()
+ self.udp_dport = 'TCP.dport'
+ self.udp_sport = 'TCP.sport'
+ tcp_params = {'flags': '', 'window': 0}
+ self._set_udp_fields(**tcp_params)
+ if 'ttl' in outer_l3v6:
+ ip6_params['ttl'] = outer_l3v6['ttl']
+ if 'tc' in outer_l3v6:
+ ip6_params['tc'] = outer_l3v6['tc']
+ if 'hlim' in outer_l3v6:
+ ip6_params['hlim'] = outer_l3v6['hlim']
+ self._set_ip6_fields(**ip6_params)
+ if 'srcip6' in outer_l3v6:
+ self.set_src_ip6(outer_l3v6['srcip6'])
+ if 'dstip6' in outer_l3v6:
+ self.set_dst_ip6(outer_l3v6['dstip6'])
+
+ def set_outer_l4_fields(self, outer_l4):
+ """ setup outer l4 fields from traffic profile """
+ if 'srcport' in outer_l4:
+ self.set_src_port(outer_l4['srcport'])
+ if 'dstport' in outer_l4:
+ self.set_dst_port(outer_l4['dstport'])
+
+ def generate_imix_data(self, packet_definition):
+ """ generate packet size for a given traffic profile """
+ imix_count = {}
+ imix_data = {}
+ if not packet_definition:
+ return imix_count
+ imix = packet_definition.get('framesize')
+ if imix:
+ for size in imix:
+ data = imix[size]
+ imix_data[int(size[:-1])] = int(data)
+ imix_sum = sum(imix_data.values())
+ if imix_sum > 100:
+ raise SystemExit("Error in IMIX data")
+ elif imix_sum < 100:
+ imix_data[64] = imix_data.get(64, 0) + (100 - imix_sum)
+
+ avg_size = 0.0
+ for size in imix_data:
+ count = int(imix_data[size])
+ if count:
+ avg_size += round(size * count / 100, 2)
+ pps = round(self.pps * count / 100, 0)
+ imix_count[size] = pps
+ self.rate = round(1342177280 / avg_size, 0) * 2
+ logging.debug("Imax: %s rate: %s", imix_count, self.rate)
+ return imix_count
+
+ def get_streams(self):
+ """ generate trex stream """
+ self.streams = []
+ self.pps = self.params['traffic_profile'].get('frame_rate', 100)
+ for packet_name in self.profile_data:
+ outer_l2 = self.profile_data[packet_name].get('outer_l2')
+ imix_data = self.generate_imix_data(outer_l2)
+ if not imix_data:
+ imix_data = {64: self.pps}
+ self.generate_vm(self.profile_data[packet_name])
+ for size in imix_data:
+ self._generate_streams(size, imix_data[size])
+ self._generate_profile()
+ return self.profile
+
+ def generate_vm(self, packet_definition):
+ """ generate trex vm with flows setup """
+ self.ether_packet = Pkt.Ether()
+ self.ip_packet = Pkt.IP()
+ self.ip6_packet = None
+ self.udp_packet = Pkt.UDP()
+ self.udp_dport = 'UDP.dport'
+ self.udp_sport = 'UDP.sport'
+ self.qinq = False
+ self.vm_flow_vars = []
+ outer_l2 = packet_definition.get('outer_l2', None)
+ outer_l3v4 = packet_definition.get('outer_l3v4', None)
+ outer_l3v6 = packet_definition.get('outer_l3v6', None)
+ outer_l4 = packet_definition.get('outer_l4', None)
+ if outer_l2:
+ self.set_outer_l2_fields(outer_l2)
+ if outer_l3v4:
+ self.set_outer_l3v4_fields(outer_l3v4)
+ if outer_l3v6:
+ self.set_outer_l3v6_fields(outer_l3v6)
+ if outer_l4:
+ self.set_outer_l4_fields(outer_l4)
+ self.trex_vm = STLScVmRaw(self.vm_flow_vars)
+
+ def generate_packets(self):
+ """ generate packets from trex TG """
+ base_pkt = self.base_pkt
+ size = self.fsize - 4
+ pad = max(0, size - len(base_pkt)) * 'x'
+ self.packets = [STLPktBuilder(pkt=base_pkt / pad,
+ vm=vm) for vm in self.vms]
+
+ def _create_single_packet(self, size=64):
+ size = size - 4
+ ether_packet = self.ether_packet
+ ip_packet = self.ip6_packet if self.ip6_packet else self.ip_packet
+ udp_packet = self.udp_packet
+ if self.qinq:
+ qinq_packet = self.qinq_packet
+ base_pkt = ether_packet / qinq_packet / ip_packet / udp_packet
+ else:
+ base_pkt = ether_packet / ip_packet / udp_packet
+ pad = max(0, size - len(base_pkt)) * 'x'
+ packet = STLPktBuilder(pkt=base_pkt / pad, vm=self.trex_vm)
+ return packet
+
+ def _create_single_stream(self, packet_size, pps, isg=0):
+ packet = self._create_single_packet(packet_size)
+ if self.pg_id:
+ self.pg_id += 1
+ stl_flow = STLFlowLatencyStats(pg_id=self.pg_id)
+ stream = STLStream(isg=isg, packet=packet, mode=STLTXCont(pps=pps),
+ flow_stats=stl_flow)
+ else:
+ stream = STLStream(isg=isg, packet=packet, mode=STLTXCont(pps=pps))
+ return stream
+
+ def _generate_streams(self, packet_size, pps):
+ self.streams.append(self._create_single_stream(packet_size, pps))
+
+ def _generate_profile(self):
+ self.profile = STLProfile(self.streams)
+
+ @classmethod
+ def _get_start_end_ipv6(cls, start_ip, end_ip):
+ try:
+ ip1 = socket.inet_pton(socket.AF_INET6, start_ip)
+ ip2 = socket.inet_pton(socket.AF_INET6, end_ip)
+ hi1, lo1 = struct.unpack('!QQ', ip1)
+ hi2, lo2 = struct.unpack('!QQ', ip2)
+ if ((hi1 << 64) | lo1) > ((hi2 << 64) | lo2):
+ raise SystemExit("IPv6: start_ip is greater then end_ip")
+ max_p1 = abs(int(lo1) - int(lo2))
+ base_p1 = lo1
+ except Exception as ex_error:
+ raise SystemExit(ex_error)
+ else:
+ return base_p1, max_p1 + base_p1
+
+ @classmethod
+ def _get_random_value(cls, min_port, max_port):
+ cryptogen = SystemRandom()
+ return cryptogen.randrange(min_port, max_port)
Code Review / yardstick.git / commitdiff