summaryrefslogtreecommitdiffstats
path: root/scripts
diff options
context:
space:
mode:
authorimarom <imarom@cisco.com>2017-02-06 13:20:41 +0200
committerimarom <imarom@cisco.com>2017-02-06 13:21:44 +0200
commitbcea02625fb2ce43ee81e56320941dfd8ff10327 (patch)
tree1f237b927b4a3a75a53befca0e68ae765cd05b04 /scripts
parent1570aea16299122399e14c7c281fe3d4259e63a7 (diff)
added example for functional tests using start_capture/stop_capture APIs
Signed-off-by: imarom <imarom@cisco.com>
Diffstat (limited to 'scripts')
-rw-r--r--scripts/automation/regression/stateless_tests/stl_general_test.py1
-rw-r--r--scripts/automation/trex_control_plane/stl/examples/stl_functional.py78
-rwxr-xr-xscripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py108
-rw-r--r--scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_exceptions.py17
-rwxr-xr-xscripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py7
5 files changed, 197 insertions, 14 deletions
diff --git a/scripts/automation/regression/stateless_tests/stl_general_test.py b/scripts/automation/regression/stateless_tests/stl_general_test.py
index 7d44d04a..6470d8c2 100644
--- a/scripts/automation/regression/stateless_tests/stl_general_test.py
+++ b/scripts/automation/regression/stateless_tests/stl_general_test.py
@@ -37,6 +37,7 @@ class CStlGeneral_Test(CTRexGeneral_Test):
sys.stdout.write('.')
sys.stdout.flush()
try:
+ CTRexScenario.stl_trex.remove_all_captures()
CTRexScenario.stl_ports_map = stl_map_ports(self.stl_trex)
if self.verify_bidirectional(CTRexScenario.stl_ports_map):
print('')
diff --git a/scripts/automation/trex_control_plane/stl/examples/stl_functional.py b/scripts/automation/trex_control_plane/stl/examples/stl_functional.py
new file mode 100644
index 00000000..6322ce88
--- /dev/null
+++ b/scripts/automation/trex_control_plane/stl/examples/stl_functional.py
@@ -0,0 +1,78 @@
+import stl_path
+from trex_stl_lib.api import *
+
+"""
+An example on how to use TRex for functional tests
+
+It can be used for various tasks and can replace simple Pagent/Scapy
+low rate tests
+"""
+
+def test_dot1q (c, rx_port, tx_port):
+
+ # activate service mode on RX code
+ c.set_service_mode(ports = rx_port)
+
+ # generate a simple Dot1Q
+ pkt = Ether() / Dot1Q(vlan = 100) / IP()
+
+ # start a capture
+ capture = c.start_capture(rx_ports = rx_port)
+
+ # push the Dot1Q packet to TX port... we need 'force' because this is under service mode
+ print('\nSending 1 Dot1Q packet(s) on port {}'.format(tx_port))
+
+ c.push_packets(ports = tx_port, pkts = pkt, force = True)
+ c.wait_on_traffic(ports = tx_port)
+
+ rx_pkts = []
+ c.stop_capture(capture_id = capture['id'], output = rx_pkts)
+
+ print('\nRecived {} packets on port {}:\n'.format(len(rx_pkts), rx_port))
+
+ c.set_service_mode(ports = rx_port, enabled = False)
+
+ # got back one packet
+ assert(len(rx_pkts) == 1)
+ rx_scapy_pkt = Ether(rx_pkts[0]['binary'])
+
+ # it's a Dot1Q with the same VLAN
+ assert('Dot1Q' in rx_scapy_pkt)
+ assert(rx_scapy_pkt.vlan == 100)
+
+
+ rx_scapy_pkt.show2()
+
+
+
+def main ():
+
+ # create a client
+ c = STLClient()
+
+ try:
+ # connect to the server
+ c.connect()
+
+ # there should be at least two ports connected
+ tx_port, rx_port = stl_map_ports(c)['bi'][0]
+ c.reset(ports = [tx_port, rx_port])
+
+ # call the test
+ test_dot1q(c, tx_port, rx_port)
+
+
+ except STLError as e:
+ print(e)
+
+ finally:
+ c.stop()
+ c.remove_all_captures()
+ c.set_service_mode(enabled = False)
+ c.disconnect()
+
+
+
+if __name__ == '__main__':
+ main()
+
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
index b95c190b..a1290a37 100755
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
@@ -20,12 +20,14 @@ from texttable import ansi_len
from collections import namedtuple, defaultdict
from yaml import YAMLError
+from contextlib import contextmanager
import time
import datetime
import re
import random
import json
import traceback
+import tempfile
import os.path
############################ logger #############################
@@ -2691,8 +2693,8 @@ class STLClient(object):
self.remove_all_streams(ports = ports)
id_list = self.add_streams(profile.get_streams(), ports)
-
- return self.start(ports = ports, duration = duration)
+
+ return self.start(ports = ports, duration = duration, force = force)
else:
@@ -2728,11 +2730,91 @@ class STLClient(object):
if profile_b:
self.add_streams(profile_b.get_streams(), slave)
- return self.start(ports = all_ports, duration = duration)
+ return self.start(ports = all_ports, duration = duration, force = force)
+
+
+
+
+ @__api_check(True)
+ def push_packets (self,
+ pkts,
+ ports = None,
+ ipg_usec = 100,
+ count = 1,
+ duration = -1,
+ force = False,
+ vm = None):
+
+ """
+ Pushes a list of packets to the server
+ a 'packet' can be anything with a bytes representation
+ such as Scapy object, a simple string, a byte array and etc.
+
+ Total size, as for PCAP pushing is limited to 1MB
+ unless 'force' is specified
+
+ the list of packets will be saved to a temporary file
+ which will be deleted when the function exists
+
+ :parameters:
+ pkts : list or
+ PCAP filename (accessible locally)
+
+ ports : list
+ Ports on which to execute the command
+
+ ipg_usec : float
+ Inter-packet gap in microseconds.
+
+ count: int
+ How many times to transmit the list
+ duration: float
+ Limit runtime by duration in seconds
+ force: bool
+ Ignore size limit - push any size to the server
+ vm: list of VM instructions
+ VM instructions to apply for every packet
+ :raises:
+ + :exc:`STLError`
+ """
+ validate_type('ipg_usec', ipg_usec, (float, int, type(None)))
+ if ipg_usec < 0:
+ raise STLError("'ipg_usec' should not be negative")
+
+ # create a temporary file while will be deleted when leaving scope
+ with tempfile.NamedTemporaryFile(delete = True) as f:
+
+ # write packets to file
+ writer = RawPcapWriter(f.name, linktype = 1)
+ writer._write_header(None)
+
+ # write to the file
+ for i, pkt in enumerate(listify(pkts)):
+ ts = (ipg_usec * i) / 1.0e6
+ ts_sec, ts_usec = sec_split_usec(ts)
+ writer._write_packet(bytes(pkt), sec = ts_sec, usec = ts_usec)
+
+ # close the writer
+ writer.close()
+
+ # now inject the file
+ self.push_pcap(f.name,
+ ports = ports,
+ ipg_usec = ipg_usec,
+ speedup = 1,
+ count = count,
+ duration = duration,
+ force = force,
+ vm = vm,
+ packet_hook = None,
+ is_dual = False,
+ min_ipg_usec = None)
+
+
@__api_check(True)
def validate (self, ports = None, mult = "1", duration = -1, total = False):
@@ -3011,7 +3093,15 @@ class STLClient(object):
if not rc:
raise STLError(rc)
-
+ @contextmanager
+ def service_mode (self, ports):
+ self.set_service_mode(ports = ports)
+ try:
+ yield
+ finally:
+ self.set_service_mode(ports = ports, enabled = False)
+
+
@__api_check(True)
def resolve (self, ports = None, retries = 0, verbose = True):
"""
@@ -3206,11 +3296,19 @@ class STLClient(object):
capture_id: int
an active capture ID to stop
- output: None/ str / list
+ output: None / str / list
if output is None - all the packets will be discarded
if output is a 'str' - it will be interpeted as output filename
if it is a list, the API will populate the list with packet objects
+ in case 'output' is a list, each element in the list is an object
+ containing:
+ 'binary' - binary bytes of the packet
+ 'origin' - RX or TX origin
+ 'ts' - timestamp relative to the start of the capture
+ 'index' - order index in the capture
+ 'port' - on which port did the packet arrive or was transmitted from
+
:raises:
+ :exe:'STLError'
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_exceptions.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_exceptions.py
index 2ca92cb8..fa8163b8 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_exceptions.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_exceptions.py
@@ -17,14 +17,15 @@ class STLError(Exception):
def __str__ (self):
- fname = os.path.split(self.tb[-2][0])[1]
- lineno = self.tb[-2][1]
- func = self.tb[-2][2]
- src = self.tb[-2][3]
-
- s = "\n******\n"
- s += "Error at {0}:{1} - '{2}'\n\n".format(format_text(fname, 'bold'), format_text(lineno, 'bold'), format_text(src.strip(), 'bold'))
- s += "specific error:\n\n{0}\n".format(format_text(self.msg, 'bold'))
+ s = format_text("\n******\n", 'bold')
+ s += format_text('\nSummary error report:\n\n', 'underline')
+ s += format_text(self.msg + '\n', 'bold')
+
+ s += format_text("\nFull error report:\n\n", 'underline')
+
+ for line in reversed(self.tb):
+ fname, lineno, func, src = os.path.split(line[0])[1], line[1], line[2], line[3]
+ s += " {:}:{:<20} - '{}'\n".format(format_text(fname, 'bold'), format_text(lineno, 'bold'), format_text(src.strip(), 'bold'))
return s
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
index efa450e1..8d727d6d 100755
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py
@@ -1047,6 +1047,9 @@ class STLProfile(object):
if split_mode is None:
pkts = PCAPReader(pcap_file).read_all()
+ if len(pkts) == 0:
+ raise STLError("'{0}' does not contain any packets".format(pcap_file))
+
return STLProfile.__pkts_to_streams(pkts,
ipg_usec,
min_ipg_usec,
@@ -1056,7 +1059,9 @@ class STLProfile(object):
packet_hook)
else:
pkts_a, pkts_b = PCAPReader(pcap_file).read_all(split_mode = split_mode)
-
+ if (len(pkts_a) + len(pkts_b)) == 0:
+ raise STLError("'{0}' does not contain any packets".format(pcap_file))
+
# swap the packets if a is empty, or the ts of first packet in b is earlier
if not pkts_a:
pkts_a, pkts_b = pkts_b, pkts_a