def configure(self, options, conf):
self.functional = options.functional
- if self.functional:
+ self.collect_only = options.collect_only
+ if self.functional or self.collect_only:
return
if CTRexScenario.setup_dir and options.config_path:
raise Exception('Please either define --cfg or use env. variable SETUP_DIR, not both.')
self.loggerPath = options.log_path
def begin (self):
- if self.functional:
+ if self.functional or self.collect_only:
return
# initialize CTRexScenario global testing class, to be used by all tests
CTRexScenario.configuration = self.configuration
CustomLogger.setup_custom_logger('TRexLogger')
def finalize(self, result):
- if self.functional:
+ if self.functional or self.collect_only:
return
CTRexScenario.is_init = False
stop_trex_remote_server(self.configuration.trex)
CTRexScenario.setup_name = os.path.basename(CTRexScenario.setup_dir)
xml_name = 'report_%s.xml' % CTRexScenario.setup_name
- nose_argv= sys.argv + ['-s', '-v', '--exe', '--rednose', '--detailed-errors', '--with-xunit', '--xunit-file=%s/%s' % (CTRexScenario.report_dir, xml_name)]
-
- for arg in sys.argv:
+ nose_argv = ['-s', '-v', '--exe', '--rednose', '--detailed-errors']
+ if '--collect-only' in sys.argv: # this is a user trying simply to view the available tests. no need xunit.
+ CTRexScenario.is_test_list = True
+ else:
+ nose_argv += ['--with-xunit', '--xunit-file=%s/%s' % (CTRexScenario.report_dir, xml_name)]
+ set_report_dir(CTRexScenario.report_dir)
+
+ for i, arg in enumerate(sys.argv):
if 'unit_tests/' in arg:
specific_tests = True
+ sys.argv[i] = arg[arg.find('unit_tests/'):]
if 'log-path' in arg:
disableLogCapture = True
- if arg == '--collect-only': # this is a user trying simply to view the available tests. removing xunit param from nose args
- nose_argv[-3:-1] = []
- CTRexScenario.is_test_list = True
- else:
- set_report_dir(CTRexScenario.report_dir)
+
+ nose_argv += sys.argv
# Run all of the unit tests or just the selected ones
if not specific_tests:
- nose_argv += ['unit_tests']
+ if '--functional' in sys.argv:
+ nose_argv += ['unit_tests/functional_tests']
+ else:
+ nose_argv += ['unit_tests']
if disableLogCapture:
nose_argv += ['--nologcapture']
import os
import unittest
+from trex_stl_lib.trex_stl_hltapi import STLHltStream
+from trex_stl_lib.trex_stl_types import validate_type
from nose.plugins.attrib import attr
def compare_yamls(yaml1, yaml2):
- from trex_stl_lib.trex_stl_types import validate_type
validate_type('yaml1', yaml1, str)
validate_type('yaml2', yaml2, str)
i = 0
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 = self.gen_stream(name = 'stream-0')
+ test_stream = STLHltStream(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 = 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',
- ip_dst_count = 2,
- ip_dst_mode = 'random',
- name = 'stream-0')
+ test_stream = STLHltStream(ip_src_addr = '192.168.1.1',
+ ip_src_mode = 'increment',
+ ip_src_count = 5,
+ ip_dst_addr = '5.5.5.5',
+ ip_dst_count = 2,
+ ip_dst_mode = 'random',
+ name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
# Eth / IP / TCP, tcp ports are changed by VM
def test_tcp_ranges(self):
- 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,
- tcp_dst_port = 1234,
- name = 'stream-0')
+ test_stream = STLHltStream(tcp_src_port_mode = 'decrement',
+ tcp_src_port_count = 10,
+ tcp_dst_port_mode = 'random',
+ tcp_dst_port_count = 10,
+ tcp_dst_port = 1234,
+ name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
# Eth / IP / UDP, udp ports are changed by VM
def test_udp_ranges(self):
# UDP is not set, expecting ignore of wrong UDP arguments
- 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_count = 'sfgsdfg',
- udp_dst_port = 'sdfgsdfg')
+ STLHltStream(udp_src_port_mode = 'qwerqwer',
+ udp_src_port_count = 'weqwer',
+ udp_src_port = 'qwerqwer',
+ udp_dst_port_mode = 'qwerqwe',
+ udp_dst_port_count = 'sfgsdfg',
+ udp_dst_port = 'sdfgsdfg')
# UDP is set, expecting fail due to wrong UDP arguments
with self.assertRaises(Exception):
- self.gen_stream(l4_protocol = 'udp',
- udp_src_port_mode = 'qwerqwer',
- udp_src_port_count = 'weqwer',
- udp_src_port = 'qwerqwer',
- udp_dst_port_mode = 'qwerqwe',
- udp_dst_port_count = 'sfgsdfg',
- udp_dst_port = 'sdfgsdfg')
+ STLHltStream(l4_protocol = 'udp',
+ udp_src_port_mode = 'qwerqwer',
+ udp_src_port_count = 'weqwer',
+ udp_src_port = 'qwerqwer',
+ udp_dst_port_mode = 'qwerqwe',
+ udp_dst_port_count = 'sfgsdfg',
+ udp_dst_port = 'sdfgsdfg')
# generate it already with correct arguments
- test_stream = self.gen_stream(l4_protocol = 'udp',
- udp_src_port_mode = 'decrement',
- udp_src_port_count = 10,
- udp_src_port = 1234,
- udp_dst_port_mode = 'increment',
- udp_dst_port_count = 10,
- udp_dst_port = 1234,
- name = 'stream-0')
+ test_stream = STLHltStream(l4_protocol = 'udp',
+ udp_src_port_mode = 'decrement',
+ udp_src_port_count = 10,
+ udp_src_port = 1234,
+ udp_dst_port_mode = 'increment',
+ udp_dst_port_count = 10,
+ udp_dst_port = 1234,
+ name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
def test_pkt_len_by_framesize(self):
# frame_size_step should be 1 (as default)
with self.assertRaises(Exception):
- test_stream = self.self.gen_stream(length_mode = 'decrement',
- frame_size_min = 100,
- frame_size_max = 3000,
- frame_size_step = 20)
+ test_stream = STLHltStream(length_mode = 'decrement',
+ frame_size_min = 100,
+ frame_size_max = 3000,
+ frame_size_step = 20)
# just check errors, no compare to golden
- self.gen_stream(length_mode = 'increment',
- frame_size_min = 100,
- frame_size_max = 3000)
- test_stream = self.gen_stream(length_mode = 'decrement',
- frame_size_min = 100,
- frame_size_max = 3000,
- name = 'stream-0')
+ STLHltStream(length_mode = 'increment',
+ frame_size_min = 100,
+ frame_size_max = 3000)
+ test_stream = STLHltStream(length_mode = 'decrement',
+ frame_size_min = 100,
+ frame_size_max = 3000,
+ name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
def test_pkt_len_by_l3length(self):
# l3_length_step should be 1
with self.assertRaises(Exception):
- self.gen_stream(l4_protocol = 'udp',
- length_mode = 'random',
- l3_length_min = 100,
- l3_length_max = 400,
- l3_length_step = 20)
- test_stream = self.gen_stream(l4_protocol = 'udp',
- length_mode = 'random',
- l3_length_min = 100,
- l3_length_max = 400,
- name = 'stream-0')
+ STLHltStream(l4_protocol = 'udp',
+ length_mode = 'random',
+ l3_length_min = 100,
+ l3_length_max = 400,
+ l3_length_step = 20)
+ test_stream = STLHltStream(l4_protocol = 'udp',
+ length_mode = 'random',
+ l3_length_min = 100,
+ l3_length_max = 400,
+ name = 'stream-0')
self.test_yaml = test_stream.dump_to_yaml(self.yaml_save_location())
self.golden_yaml = '''
- name: stream-0
return HLT_ERR('Could not generate imix streams: %s' % e)
return HLT_OK(stream_id = streams_per_port)
try:
- stream_obj = CTRexHltApiBuilder.generate_stream(**kwargs)
+ stream_obj = STLHltStream(**kwargs)
except Exception as e:
return HLT_ERR('Could not create stream: %s' % e)
return '\n'. join([str(response) for response in responses])
return responses
-class CTRexHltApiBuilder:
- @staticmethod
- def generate_stream(**user_kwargs):
+class STLHltStream(STLStream):
+ def __init__(self, **user_kwargs):
kwargs = merge_kwargs(traffic_config_kwargs, user_kwargs)
try:
- packet = CTRexHltApiBuilder.generate_packet(**kwargs)
- #except Exception as e:
- except ValueError as e:
+ packet = STLHltStream.generate_packet(**kwargs)
+ except Exception as e:
raise Exception('Could not generate packet: %s' % e)
try:
raise Exception('Could not create transmit_mode class %s: %s' % (transmit_mode, e))
try:
- stream_obj = STLStream(packet = packet,
- #enabled = True,
- #self_start = True,
- mode = transmit_mode_class,
- #rx_stats = rx_stats,
- #next_stream_id = -1,
- stream_id = kwargs.get('stream_id'),
- name = kwargs.get('name'),
- )
+ STLStream.__init__(self,
+ packet = packet,
+ #enabled = True,
+ #self_start = True,
+ mode = transmit_mode_class,
+ #rx_stats = rx_stats,
+ #next_stream_id = -1,
+ stream_id = kwargs.get('stream_id'),
+ name = kwargs.get('name'),
+ )
except Exception as e:
raise Exception('Could not create stream: %s' % e)
debug_filename = kwargs.get('save_to_yaml')
if type(debug_filename) is str:
stream_obj.dump_to_yaml(debug_filename)
- return stream_obj
@staticmethod
def generate_packet(**user_kwargs):
kwargs = merge_kwargs(traffic_config_kwargs, user_kwargs)
- pkt = STLPktBuilder()
vm_cmds = []
fix_ipv4_checksum = False
raise Exception('Packet length is bigger than defined by frame_size* or l3_length*')
base_pkt /= '!' * payload_len
+ pkt = STLPktBuilder()
pkt.set_packet(base_pkt)
if fix_ipv4_checksum and l3_layer.name == 'IP' and kwargs['ip_checksum'] is None:
vm_cmds.append(CTRexVmDescFixIpv4(offset = 'IP'))
debug_filename = kwargs.get('save_to_pcap')
if type(debug_filename) is str:
pkt.dump_pkt_to_pcap(debug_filename)
- #pkt.compile()
- #pkt.dump_scripts()
return pkt
Code Review / trex.git / commitdiff