print format_text("\n\nEvent log was cleared\n\n")
+ def complete_profile(self, text, line, begidx, endidx):
+ return self.complete_start(text,line, begidx, endidx)
+
+ def do_profile (self, line):
+ '''shows information about a profile'''
+ self.stateless_client.show_profile_line(line)
+
# tui
@verify_connected
def do_tui (self, line):
# pack the profile
try:
for port, t in zip(opts.ports, tunables):
- profile = STLProfile.load(opts.file[0], direction = (port % 2), port = port, **t)
+
+ # give priority to the user configuration over default direction
+ if not 'direction' in t:
+ t['direction'] = (port % 2)
+
+ profile = STLProfile.load(opts.file[0], **t)
+
self.add_streams(profile.get_streams(), ports = port)
except STLError as e:
return
+
+ @__console
+ def show_profile_line (self, line):
+ '''Shows profile information'''
+
+ parser = parsing_opts.gen_parser(self,
+ "port",
+ self.show_profile_line.__doc__,
+ parsing_opts.FILE_PATH)
+
+ opts = parser.parse_args(line.split())
+ if opts is None:
+ return
+
+ info = STLProfile.get_info(opts.file[0])
+
+ self.logger.log(format_text('\nProfile Information:\n', 'bold'))
+
+ # general info
+ self.logger.log(format_text('\nGeneral Information:', 'underline'))
+ self.logger.log('Filename: {:^12}'.format(opts.file[0]))
+ self.logger.log('Stream count: {:^12}'.format(info['stream_count']))
+
+ # specific info
+ profile_type = info['type']
+ self.logger.log(format_text('\nSpecific Information:', 'underline'))
+
+ if profile_type == 'python':
+ self.logger.log('Type: {:^12}'.format('Python Module'))
+ self.logger.log('Tunables: {:^12}'.format(['{0} = {1}'.format(k ,v) for k, v in info['tunables'].iteritems()]))
+
+ elif profile_type == 'yaml':
+ self.logger.log('Type: {:^12}'.format('YAML'))
+
+ elif profile_type == 'pcap':
+ self.logger.log('Type: {:^12}'.format('PCAP file'))
+
+ self.logger.log("")
+
mult = "1",
duration = -1,
mode = 'none',
- silent = False):
+ silent = False,
+ tunables = None):
if not mode in ['none', 'gdb', 'valgrind', 'json', 'yaml','pkt','native']:
raise STLArgumentError('mode', mode)
stream_list = [x for x in input_list if isinstance(x, STLStream)]
# handle YAMLs
+ if tunables == None:
+ tunables = {}
+ else:
+ tunables = tunables[0]
+
for input_file in input_files:
try:
- profile = STLProfile.load(input_file, direction = (self.port_id % 2), port = self.port_id)
+ if not 'direction' in tunables:
+ tunables['direction'] = self.port_id % 2
+
+ profile = STLProfile.load(input_file, **tunables)
+
except STLError as e:
s = format_text("\nError while loading profile '{0}'\n".format(input_file), 'bold')
s += "\n" + e.brief()
default = -1,
type = float)
+
+ parser.add_argument('-t',
+ help = 'sets tunable for a profile',
+ dest = 'tunables',
+ default = None,
+ type = parsing_opts.decode_tunables)
+
parser.add_argument('-p', '--path',
help = "BP sim path",
dest = 'bp_sim_path',
mult = options.mult,
duration = options.duration,
mode = mode,
- silent = options.silent)
+ silent = options.silent,
+ tunables = options.tunables)
except KeyboardInterrupt as e:
print "\n\n*** Caught Ctrl + C... Exiting...\n\n"
streams : list of :class:`trex_stl_lib.trex_stl_streams.STLStream`
a list of stream objects
-
"""
raise STLArgumentError('streams', streams, valid_values = STLStream)
self.streams = streams
+ self.meta = None
def get_streams (self):
yaml_loader = YAMLLoader(yaml_file)
streams = yaml_loader.parse()
- return STLProfile(streams)
+ profile = STLProfile(streams)
+ profile.meta = {'type': 'yaml'}
+
+ return profile
+
+ @staticmethod
+ def get_module_tunables(module):
+ # remove self and variables
+ func = module.register().get_streams
+ argc = func.__code__.co_argcount
+ tunables = func.__code__.co_varnames[1:argc]
+
+ # fetch defaults
+ defaults = func.func_defaults
+ if len(defaults) != (argc - 1):
+ raise STLError("Module should provide default values for all arguments on get_streams()")
+
+ output = {}
+ for t, d in zip(tunables, defaults):
+ output[t] = d
+
+ return output
@staticmethod
module = __import__(file, globals(), locals(), [], -1)
reload(module) # reload the update
+ t = STLProfile.get_module_tunables(module)
+ for arg in kwargs:
+ if not arg in t:
+ raise STLError("profile {0} does not support tunable '{1}' - supported tunables are: '{2}'".format(python_file, arg, t))
+
streams = module.register().get_streams(direction = direction, **kwargs)
+ profile = STLProfile(streams)
- return STLProfile(streams)
+ profile.meta = {'type': 'python',
+ 'tunables': t}
+
+ return profile
except Exception as e:
a, b, tb = sys.exc_info()
last_ts_usec = ts_usec
+
+ profile = STLProfile(streams)
+ profile.meta = {'type': 'pcap'}
- return STLProfile(streams)
+ return profile
else:
raise STLError("unknown profile file type: '{0}'".format(suffix))
+ profile.meta['stream_count'] = len(profile.get_streams()) if isinstance(profile.get_streams(), list) else 1
return profile
+ @staticmethod
+ def get_info (filename):
+ profile = STLProfile.load(filename)
+ return profile.meta
+
def dump_as_pkt (self):
""" dump the profile as scapy packet. in case it is raw convert to scapy and dump it"""
cnt=0;
def __init__ (self, pcap_file):
self.pcap_file = pcap_file
- def get_streams (self, direction = 0, **kwargs):
+ def get_streams (self, direction = 0, ipg_usec = 10.0, loop_count = 1, **kwargs):
- profile = STLProfile.load_pcap(self.pcap_file, ipg_usec = kwargs.get('ipg_usec', 10.0), loop_count = kwargs.get('loop_count', 1))
+ profile = STLProfile.load_pcap(self.pcap_file, ipg_usec = ipg_usec, loop_count = loop_count)
return profile.get_streams()
return vm
- def get_streams (self, direction = 0, **kwargs):
-
- ip_src_range = kwargs.get('ip_src_range', None)
- ip_dst_range = kwargs.get('up_dst_range', {'start' : '10.0.0.1', 'end': '10.0.0.254'})
+ def get_streams (self,
+ direction = 0,
+ ipg_usec = 10.0,
+ loop_count = 5,
+ ip_src_range = None,
+ ip_dst_range = {'start' : '10.0.0.1', 'end': '10.0.0.254'},
+ **kwargs):
vm = self.create_vm(ip_src_range, ip_dst_range)
profile = STLProfile.load_pcap(self.pcap_file,
- ipg_usec = kwargs.get('ipg_usec', 10.0),
- loop_count = kwargs.get('loop_count', 5),
+ ipg_usec = ipg_usec,
+ loop_count = loop_count,
vm = vm)
return profile.get_streams()
Code Review / trex.git / commitdiff