#!/router/bin/python
-from client.trex_hltapi import CTRexHltApiBuilder
import os
import unittest
-
-gen_stream = CTRexHltApiBuilder.generate_stream
+from nose.plugins.attrib import attr
def compare_yamls(yaml1, yaml2):
- if type(yaml1) is not str:
- raise Exception("yaml1 is '%s', expected str" % type(yaml1))
- if type(yaml2) is not str:
- raise Exception("yaml2 is '%s', expected str" % type(yaml2))
+ from trex_stl_lib.trex_stl_types import validate_type
+ validate_type('yaml1', yaml1, str)
+ validate_type('yaml2', yaml2, str)
i = 0
for line1, line2 in zip(yaml1.strip().split('\n'), yaml2.strip().split('\n')):
i += 1
if line1 != line2:
raise Exception('yamls are not equal starting from line %s:\n%s\n\t<->\n%s' % (i, line1.strip(), line2.strip()))
+@attr('run_on_trex')
class CTRexHltApi_Test(unittest.TestCase):
''' Checks correct HLTAPI creation of packet/VM '''
def setUp(self):
+ from trex_stl_lib.trex_stl_hltapi import CTRexHltApiBuilder
+ self.gen_stream = CTRexHltApiBuilder.generate_stream
+
self.golden_yaml = None
self.test_yaml = None
# Eth/IP/TCP, all values default, no VM instructions
def test_default(self):
- test_stream = gen_stream(name = 'stream-0')
+ test_stream = self.gen_stream(name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
# Eth/IP/TCP, ip src and dest is changed by VM
def test_ip_ranges(self):
- test_stream = gen_stream(ip_src_addr = '192.168.1.1',
+ test_stream = self.gen_stream(ip_src_addr = '192.168.1.1',
ip_src_mode = 'increment',
ip_src_count = 5,
ip_dst_addr = '5.5.5.5',
# Eth / IP / TCP, tcp ports are changed by VM
def test_tcp_ranges(self):
- test_stream = gen_stream(tcp_src_port_mode = 'decrement',
+ test_stream = self.gen_stream(tcp_src_port_mode = 'decrement',
tcp_src_port_count = 10,
tcp_dst_port_mode = 'random',
tcp_dst_port_count = 10,
# Eth / IP / UDP, udp ports are changed by VM
def test_udp_ranges(self):
# UDP is not set, expecting ignore of wrong UDP arguments
- gen_stream(udp_src_port_mode = 'qwerqwer',
+ self.gen_stream(udp_src_port_mode = 'qwerqwer',
udp_src_port_count = 'weqwer',
udp_src_port = 'qwerqwer',
udp_dst_port_mode = 'qwerqwe',
udp_dst_port = 'sdfgsdfg')
# UDP is set, expecting fail due to wrong UDP arguments
with self.assertRaises(Exception):
- gen_stream(l4_protocol = 'udp',
+ self.gen_stream(l4_protocol = 'udp',
udp_src_port_mode = 'qwerqwer',
udp_src_port_count = 'weqwer',
udp_src_port = 'qwerqwer',
udp_dst_port_count = 'sfgsdfg',
udp_dst_port = 'sdfgsdfg')
# generate it already with correct arguments
- test_stream = gen_stream(l4_protocol = 'udp',
+ test_stream = self.gen_stream(l4_protocol = 'udp',
udp_src_port_mode = 'decrement',
udp_src_port_count = 10,
udp_src_port = 1234,
def test_pkt_len_by_framesize(self):
# frame_size_step should be 1 (as default)
with self.assertRaises(Exception):
- test_stream = gen_stream(length_mode = 'decrement',
+ test_stream = self.self.gen_stream(length_mode = 'decrement',
frame_size_min = 100,
frame_size_max = 3000,
frame_size_step = 20)
# just check errors, no compare to golden
- gen_stream(length_mode = 'increment',
+ self.gen_stream(length_mode = 'increment',
frame_size_min = 100,
frame_size_max = 3000)
- test_stream = gen_stream(length_mode = 'decrement',
+ test_stream = self.gen_stream(length_mode = 'decrement',
frame_size_min = 100,
frame_size_max = 3000,
name = 'stream-0')
def test_pkt_len_by_l3length(self):
# l3_length_step should be 1
with self.assertRaises(Exception):
- gen_stream(l4_protocol = 'udp',
+ self.gen_stream(l4_protocol = 'udp',
length_mode = 'random',
l3_length_min = 100,
l3_length_max = 400,
l3_length_step = 20)
- test_stream = gen_stream(l4_protocol = 'udp',
+ test_stream = self.gen_stream(l4_protocol = 'udp',
length_mode = 'random',
l3_length_min = 100,
l3_length_max = 400,
output_cap = os.path.join("/tmp/", "{0}_test.cap".format(testname))
golden_cap = os.path.join(self.test_path, "stl/golden/{0}_golden.cap".format(testname))
-
+ if os.path.exists(output_cap):
+ os.unlink(output_cap)
try:
rc = self.run_sim(self.profiles[profile], output_cap, options, silent)
assert_equal(rc, True)
output_cap = "a.pcap"
input_file = os.path.join('stl/', profile)
golden_file = os.path.join('exp',os.path.basename(profile).split('.')[0]+'.pcap');
+ if os.path.exists(output_cap):
+ os.unlink(output_cap)
try:
rc = self.run_sim(input_file, output_cap, options, silent)
assert_equal(rc, True)
def test_tuple_gen (self):
self.golden_run("basic_tuple_gen", "imix_tuple_gen", "-m 50kpps --limit 500 --cores 8", silent = False)
- def test_all_profiles (self):
+ def test_stl_profiles (self):
p = [
["udp_1pkt_1mac_override.py","-m 1 -l 50",True],
["syn_attack.py","-m 1 -l 50",False], # can't compare random now
self.run_py_profile_path (obj[0],obj[1],compare =obj[2], do_no_remove=False)
+ def test_hlt_profiles (self):
+ p = (
+ ['hlt/hlt_udp_inc_len_9k.py', '-m 1 -l 50', False],
+ )
+
+
+ for obj in p:
+ self.run_py_profile_path (obj[0], obj[1], compare =obj[2], do_no_remove=False)
# valgrind tests
def test_valgrind_various_profiles (self):
def __str__ (self):
exc_type, exc_obj, exc_tb = sys.exc_info()
+ if not exc_tb:
+ return self.msg
fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
self.msg = "Operation '{0}' on port(s) '{1}' is not valid while port(s) '{2}'".format(op, port, state)
-# raised when argument is not valid for operation
+# raised when argument value is not valid for operation
class STLArgumentError(STLError):
def __init__ (self, name, got, valid_values = None, extended = None):
self.msg = "Argument: '{0}' invalid value: '{1}'".format(name, got)
if extended:
self.msg += "\n{0}".format(extended)
+# raised when argument type is not valid for operation
+class STLTypeError(STLError):
+ def __init__ (self, arg_name, arg_type, valid_types):
+ self.msg = "Argument: '%s' invalid type: %s, expecting type(s): %s." % (arg_name, arg_type, valid_types)
+
# raised when timeout occurs
class STLTimeoutError(STLError):
def __init__ (self, timeout):
self.msg = "Timeout: operation took more than '{0}' seconds".format(timeout)
-
from collections import namedtuple
from utils.text_opts import *
+from trex_stl_exceptions import *
RpcCmdData = namedtuple('RpcCmdData', ['method', 'params'])
def RC_WARN (warn):
return RC(True, warn, is_warn = True)
+
+
+# validate type of arg
+# example1: validate_type('somearg', somearg, [int, long])
+# example2: validate_type('another_arg', another_arg, str)
+def validate_type(arg_name, arg, valid_types):
+ if type(valid_types) is list:
+ valid_types = tuple(valid_types)
+ if type(valid_types) is type or type(valid_types) is tuple:
+ if isinstance(arg, valid_types):
+ return
+ raise STLTypeError(arg_name, type(arg), valid_types)
+ else:
+ raise STLError('validate_type: valid_types should be type or list or tuple of types')
Code Review / trex.git / commitdiff