#!/usr/bin/env python
__copyright__ = "Copyright 2014"
"""
Name:
trex_unit_test.py
Description:
This script creates the functionality to test the performance of the TRex traffic generator
The tested scenario is a TRex TG directly connected to a Cisco router.
::
Topology:
------- --------
| | Tx---1gig/10gig----Rx | |
| TRex | | router |
| | Rx---1gig/10gig----Tx | |
------- --------
"""
import os
import sys
import outer_packages
import datetime
import nose
from nose.plugins import Plugin
from nose.plugins.xunit import escape_cdata
from nose.selector import Selector
from nose.exc import SkipTest
from nose.pyversion import force_unicode, format_exception
import CustomLogger
import misc_methods
from rednose import RedNose
import termstyle
from trex import CTRexScenario
from trex_stf_lib.trex_client import *
from trex_stf_lib.trex_exceptions import *
from trex_stl_lib.api import *
from trex_stl_lib.utils.GAObjClass import GAmanager_Regression
import trex_elk
import trex
import socket
from pprint import pprint
import time
from distutils.dir_util import mkpath
import re
from io import StringIO
TEST_ID = re.compile(r'^(.*?)(\(.*\))$')
def id_split(idval):
m = TEST_ID.match(idval)
if m:
name, fargs = m.groups()
head, tail = name.rsplit(".", 1)
return [head, tail+fargs]
else:
return idval.rsplit(".", 1)
# nose overrides
# option to select wanted test by name without file, class etc.
def new_Selector_wantMethod(self, method, orig_Selector_wantMethod = Selector.wantMethod):
result = orig_Selector_wantMethod(self, method)
return result and (not CTRexScenario.test or list(filter(lambda t: t in getattr(method, '__name__', ''), CTRexScenario.test.split(','))))
Selector.wantMethod = new_Selector_wantMethod
def new_Selector_wantFunction(self, function, orig_Selector_wantFunction = Selector.wantFunction):
result = orig_Selector_wantFunction(self, function)
return result and (not CTRexScenario.test or list(filter(lambda t: t in getattr(function, '__name__', ''), CTRexScenario.test.split(','))))
Selector.wantFunction = new_Selector_wantFunction
# override nose's strange representation of setUpClass errors
def __suite_repr__(self):
if hasattr(self.context, '__module__'): # inside class, setUpClass etc.
class_repr = nose.suite._strclass(self.context)
else: # outside of class, setUpModule etc.
class_repr = nose.suite._strclass(self.__class__)
return '%s.%s' % (class_repr, getattr(self.context, '__name__', self.context))
nose.suite.ContextSuite.__repr__ = __suite_repr__
nose.suite.ContextSuite.__str__ = __suite_repr__
# /nose overrides
def fatal(txt):
print(txt)
sys.exit(1)
def check_trex_path(trex_path):
if os.path.isfile('%s/trex_daemon_server' % trex_path):
return os.path.abspath(trex_path)
def check_setup_path(setup_path):
if os.path.isfile('%s/config.yaml' % setup_path):
return os.path.abspath(setup_path)
def get_trex_path():
latest_build_path = check_trex_path(os.getenv('TREX_UNDER_TEST')) # TREX_UNDER_TEST is env var pointing to <trex-core>/scripts
if not latest_build_path:
latest_build_path = check_trex_path(os.path.join(os.pardir, os.pardir))
if not latest_build_path:
fatal('Could not determine trex_under_test folder, try setting env.var. TREX_UNDER_TEST')
return latest_build_path
def address_to_ip(address):
for i in range(5):
try:
return socket.gethostbyname(address)
except:
continue
return socket.gethostbyname(address)
class TRexTee(object):
def __init__(self, encoding, *args):
self._encoding = encoding
self._streams = args
def write(self, data):
data = force_unicode(data, self._encoding)
for s in self._streams:
s.write(data)
def writelines(self, lines):
for line in lines:
self.write(line)
def flush(self):
for s in self._streams:
s.flush()
def isatty(self):
return False
class CTRexTestConfiguringPlugin(Plugin):
encoding = 'UTF-8'
def __init__(self):
super(CTRexTestConfiguringPlugin, self).__init__()
self._capture_stack = []
self._currentStdout = None
self._currentStderr = None
def _timeTaken(self):
if hasattr(self, '_timer'):
taken = time.time() - self._timer
else:
# test died before it ran (probably error in setup())
# or success/failure added before test started probably
# due to custom TestResult munging
taken = 0.0
return taken
def _startCapture(self):
self._capture_stack.append((sys.stdout, sys.stderr))
self._currentStdout = StringIO()
self._currentStderr = StringIO()
sys.stdout = TRexTee(self.encoding, self._currentStdout, sys.stdout)
sys.stderr = TRexTee(self.encoding, self._currentStderr, sys.stderr)
def startContext(self, context):
self._startCapture()
def stopContext(self, context):
self._endCapture()
def beforeTest(self, test):
self._timer = time.time()
self._startCapture()
def _endCapture(self):
if self._capture_stack:
sys.stdout, sys.stderr = self._capture_stack.pop()
def afterTest(self, test):
self._endCapture()
self._currentStdout = None
self._currentStderr = None
def _getCapturedStdout(self):
if self._currentStdout:
value = self._currentStdout.getvalue()
if value:
return '<system-out><![CDATA[%s]]></system-out>' % escape_cdata(
value)
return ''
def _getCapturedStderr(self):
if self._currentStderr:
value = self._currentStderr.getvalue()
if value:
return '<system-err><![CDATA[%s]]></system-err>' % escape_cdata(
value)
return ''
def addError(self, test, err, capt=None):
elk = CTRexScenario.elk
if elk:
taken = self._timeTaken()
id = test.id()
err_msg=self._getCapturedStdout()+self._getCapturedStderr();
name=id_split(id)[-1]
elk_obj = trex.copy_elk_info ()
elk_obj['test']={
"name" : name,
"name_key" : name,
"name_full" : id,
"type" : self.get_operation_mode (),
"duration_sec" : taken,
"result" : "ERROR",
"stdout" : err_msg,
};
#pprint(elk_obj['test']);
elk.reg.push_data(elk_obj)
def addFailure(self, test, err, capt=None, tb_info=None):
elk = CTRexScenario.elk
if elk:
taken = self._timeTaken()
tb = format_exception(err, self.encoding)
id = test.id()
err_msg=self._getCapturedStdout()+self._getCapturedStderr();
name=id_split(id)[-1]
elk_obj = trex.copy_elk_info ()
elk_obj['test']={
"name" : name,
"name_key" : name,
"name_full" : id,
"type" : self.get_operation_mode (),
"duration_sec" : taken,
"result" : "FAILURE",
"stdout" : err_msg,
};
#pprint(elk_obj['test']);
elk.reg.push_data(elk_obj)
def addSuccess(self, test, capt=None):
elk = CTRexScenario.elk
if elk:
taken = self._timeTaken()
id = test.id()
name=id_split(id)[-1]
elk_obj = trex.copy_elk_info ()
elk_obj['test']={
"name" : name,
"name_key" : name,
"name_full" : id,
"type" : self.get_operation_mode (),
"duration_sec" : taken,
"result" : "PASS",
"stdout" : "",
};
#pprint(elk_obj['test']);
elk.reg.push_data(elk_obj)
def get_operation_mode (self):
if self.stateful:
return('stateful');
return('stateless');
##### option/configure
def options(self, parser, env = os.environ):
super(CTRexTestConfiguringPlugin, self).options(parser, env)
parser.add_option('--cfg', '--trex-scenario-config', action='store',
dest='config_path',
help='Specify path to folder with config.yaml and benchmark.yaml')
parser.add_option('--skip-clean', '--skip_clean', action='store_true',
dest='skip_clean_config',
help='Skip the clean configuration replace on the platform.')
parser.add_option('--load-image', '--load_image', action='store_true', default = False,
dest='load_image',
help='Install image specified in config file on router.')
parser.add_option('--log-path', '--log_path', action='store',
dest='log_path',
help='Specify path for the tests` log to be saved at. Once applied, logs capturing by nose will be disabled.') # Default is CURRENT/WORKING/PATH/trex_log/trex_log.log')
parser.add_option('--json-verbose', '--json_verbose', action="store_true", default = False,
dest="json_verbose",
help="Print JSON-RPC commands.")
parser.add_option('--telnet-verbose', '--telnet_verbose', action="store_true", default = False,
dest="telnet_verbose",
help="Print telnet commands and responces.")
parser.add_option('--server-logs', '--server_logs', action="store_true", default = False,
dest="server_logs",
help="Print server side (TRex and trex_daemon) logs per test.")
parser.add_option('--kill-running', '--kill_running', action="store_true", default = False,
dest="kill_running",
help="Kills running TRex process on remote server (useful for regression).")
parser.add_option('--func', '--functional', action="store_true", default = False,
dest="functional",
help="Run functional tests.")
parser.add_option('--stl', '--stateless', action="store_true", default = False,
dest="stateless",
help="Run stateless tests.")
parser.add_option('--stf', '--stateful', action="store_true", default = False,
dest="stateful",
help="Run stateful tests.")
parser.add_option('--pkg', type = str,
help="Run with given TRex package. Make sure the path available at server machine. Implies --restart-daemon.")
parser.add_option('--restart-daemon', action="store_true", default = False,
help="Flag that specifies to restart daemon. Implied by --pkg.")
parser.add_option('--collect', action="store_true", default = False,
help="Alias to --collect-only.")
parser.add_option('--warmup', action="store_true", default = False,
help="Warm up the system for stateful: run 30 seconds 9k imix test without check of results.")
parser.add_option('--test-client-package', '--test_client_package', action="store_true", default = False,
help="Includes tests of client package.")
parser.add_option('--long', action="store_true", default = False,
help="Flag of long tests (stability).")
parser.add_option('--ga', action="store_true", default = False,
help="Flag to send benchmarks to GA.")
parser.add_option('--no-daemon', action="store_true", default = False,
dest="no_daemon",
help="Flag that specifies to use running stl server, no need daemons.")
parser.add_option('--debug-image', action="store_true", default = False,
help="Flag that specifies to use t-rex-64-debug as TRex executable.")
parser.add_option('--trex-args', default = '',
help="Additional TRex arguments (--no-watchdog etc.).")
parser.add_option('-t', '--test', type = str,
help = 'Test name to run (without file, class etc.). Can choose several names splitted by comma.')
parser.add_option('--no-dut-config', action = 'store_true',
help = 'Skip the config of DUT to save time. Implies --skip-clean.')
def configure(self, options, conf):
self.collect_only = options.collect_only
self.functional = options.functional
self.stateless = options.stateless
self.stateful = options.stateful
self.pkg = options.pkg
self.restart_daemon = options.restart_daemon
self.json_verbose = options.json_verbose
self.telnet_verbose = options.telnet_verbose
self.no_daemon = options.no_daemon
CTRexScenario.test = options.test
if self.no_daemon and (self.pkg or self.restart_daemon):
fatal('You have specified both --no-daemon and either --pkg or --restart-daemon at same time.')
if self.no_daemon and self.stateful :
fatal("Can't run stateful without daemon.")
if self.collect_only or self.functional:
return
if CTRexScenario.setup_dir and options.config_path:
fatal('Please either define --cfg or use env. variable SETUP_DIR, not both.')
if not options.config_path and CTRexScenario.setup_dir:
options.config_path = CTRexScenario.setup_dir
if not options.config_path:
fatal('Please specify path to config.yaml using --cfg parameter or env. variable SETUP_DIR')
options.config_path = options.config_path.rstrip('/')
CTRexScenario.setup_name = os.path.basename(options.config_path)
self.configuration = misc_methods.load_complete_config_file(os.path.join(options.config_path, 'config.yaml'))
self.configuration.trex['trex_name'] = address_to_ip(self.configuration.trex['trex_name']) # translate hostname to ip
self.benchmark = misc_methods.load_benchmark_config_file(os.path.join(options.config_path, 'benchmark.yaml'))
self.enabled = True
self.modes = self.configuration.trex.get('modes', [])
self.kill_running = options.kill_running
self.load_image = options.load_image
self.clean_config = False if options.skip_clean_config else True
self.no_dut_config = options.no_dut_config
if self.no_dut_config:
self.clean_config = False
self.server_logs = options.server_logs
if options.log_path:
self.loggerPath = options.log_path
# initialize CTRexScenario global testing class, to be used by all tests
CTRexScenario.configuration = self.configuration
CTRexScenario.no_daemon = options.no_daemon
CTRexScenario.benchmark = self.benchmark
CTRexScenario.modes = set(self.modes)
CTRexScenario.server_logs = self.server_logs
CTRexScenario.debug_image = options.debug_image
CTRexScenario.json_verbose = self.json_verbose
if not self.no_daemon:
CTRexScenario.trex = CTRexClient(trex_host = self.configuration.trex['trex_name'],
verbose = self.json_verbose,
debug_image = options.debug_image,
trex_args = options.trex_args)
if self.pkg or self.restart_daemon:
if not CTRexScenario.trex.check_master_connectivity():
fatal('Could not connect to master daemon')
if options.ga and CTRexScenario.setup_name:
CTRexScenario.GAManager = GAmanager_Regression(
GoogleID = CTRexScenario.global_cfg['google']['id'],
AnalyticsUserID = CTRexScenario.setup_name,
QueueSize = CTRexScenario.global_cfg['google']['queue_size'],
Timeout = CTRexScenario.global_cfg['google']['timeout'], # seconds
UserPermission = 1,
BlockingMode = CTRexScenario.global_cfg['google']['blocking'],
appName = 'TRex',
appVer = CTRexScenario.trex_version)
CTRexScenario.elk = trex_elk.TRexEs(
CTRexScenario.global_cfg['elk']['server'],
CTRexScenario.global_cfg['elk']['port']);
self.set_cont_elk_info ()
def set_cont_elk_info (self):
elk_info={}
timestamp = datetime.datetime.now() - datetime.timedelta(hours=2); # need to update this
info = {};
img={}
img['sha'] = "v2.14" #TBD
img['build_time'] = timestamp.strftime("%Y-%m-%d %H:%M:%S")
img['version'] = "v2.14" #TBD need to fix
img['formal'] = False
setup = {}
setup['distro'] = 'None' #TBD 'Ubunto14.03'
setup['kernel'] = 'None' #TBD '2.6.12'
setup['baremetal'] = True #TBD
setup['hypervisor'] = 'None' #TBD
setup['name'] = CTRexScenario.setup_name
setup['cpu-sockets'] = 0 #TBD 2
setup['cores'] = 0 #TBD 16
setup['cpu-speed'] = -1 #TBD 3.5
setup['dut'] = 'None' #TBD 'loopback'
setup['drv-name'] = 'None' #TBD 'mlx5'
setup['nic-ports'] = 0 #TBD 2
setup['total-nic-ports'] = 0 #TBD 2
setup['nic-speed'] = "None" #"40GbE" TBD
info['image'] = img
info['setup'] = setup
elk_info['info'] =info;
elk_info['timestamp'] = timestamp.strftime("%Y-%m-%d %H:%M:%S") # need to update it
elk_info['build_id'] = os.environ.get('BUILD_ID')
elk_info['scenario'] = os.environ.get('SCENARIO')
CTRexScenario.elk_info = elk_info
def begin (self):
client = CTRexScenario.trex
if self.pkg and not CTRexScenario.pkg_updated:
if client.master_daemon.is_trex_daemon_running() and client.get_trex_cmds() and not self.kill_running:
fatal("Can't update TRex, it's running. Consider adding --kill-running flag.")
print('Updating TRex to %s' % self.pkg)
if not client.master_daemon.update_trex(self.pkg):
fatal('Failed to update TRex.')
else:
print('Updated.')
CTRexScenario.pkg_updated = True
if self.functional or self.collect_only:
return
if self.pkg or self.restart_daemon:
print('Restarting TRex daemon server')
res = client.restart_trex_daemon()
if not res:
fatal('Could not restart TRex daemon server')
print('Restarted.')
if self.kill_running:
client.kill_all_trexes()
else:
if client.get_trex_cmds():
fatal('TRex is already running')
if not self.no_daemon:
try:
client.check_server_connectivity()
except Exception as e:
fatal(e)
if 'loopback' not in self.modes:
CTRexScenario.router_cfg = dict(config_dict = self.configuration.router,
forceImageReload = self.load_image,
silent_mode = not self.telnet_verbose,
forceCleanConfig = self.clean_config,
no_dut_config = self.no_dut_config,
tftp_config_dict = self.configuration.tftp)
try:
CustomLogger.setup_custom_logger('TRexLogger', self.loggerPath)
except AttributeError:
CustomLogger.setup_custom_logger('TRexLogger')
def finalize(self, result):
while self._capture_stack:
self._endCapture()
if self.functional or self.collect_only:
return
#CTRexScenario.is_init = False
if self.stateful:
CTRexScenario.trex = None
if self.stateless:
if self.no_daemon:
if CTRexScenario.stl_trex and CTRexScenario.stl_trex.is_connected():
CTRexScenario.stl_trex.disconnect()
else:
CTRexScenario.trex.force_kill(False)
CTRexScenario.stl_trex = None
def save_setup_info():
try:
if CTRexScenario.setup_name and CTRexScenario.trex_version:
setup_info = ''
for key, value in CTRexScenario.trex_version.items():
setup_info += '{0:8}: {1}\n'.format(key, value)
cfg = CTRexScenario.configuration
setup_info += 'Server: %s, Modes: %s' % (cfg.trex.get('trex_name'), cfg.trex.get('modes'))
if cfg.router:
setup_info += '\nRouter: Model: %s, Image: %s' % (cfg.router.get('model'), CTRexScenario.router_image)
if CTRexScenario.debug_image:
setup_info += '\nDebug image: %s' % CTRexScenario.debug_image
with open('%s/report_%s.info' % (CTRexScenario.report_dir, CTRexScenario.setup_name), 'w') as f:
f.write(setup_info)
except Exception as err:
print('Error saving setup info: %s ' % err)
if __name__ == "__main__":
# setting defaults. By default we run all the test suite
specific_tests = False
CTRexScenario.report_dir = 'reports'
need_to_copy = False
setup_dir = os.getenv('SETUP_DIR', '').rstrip('/')
CTRexScenario.setup_dir = check_setup_path(setup_dir)
CTRexScenario.scripts_path = get_trex_path()
if not CTRexScenario.setup_dir:
CTRexScenario.setup_dir = check_setup_path(os.path.join('setups', setup_dir))
nose_argv = ['', '-s', '-v', '--exe', '--rednose', '--detailed-errors']
test_client_package = False
if '--test-client-package' in sys.argv:
test_client_package = True
if '--collect' in sys.argv:
sys.argv.append('--collect-only')
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
xml_arg = ''
else:
xml_name = 'unit_test.xml'
if CTRexScenario.setup_dir:
CTRexScenario.setup_name = os.path.basename(CTRexScenario.setup_dir)
xml_name = 'report_%s.xml' % CTRexScenario.setup_name
xml_arg= '--xunit-file=%s/%s' % (CTRexScenario.report_dir, xml_name)
mkpath(CTRexScenario.report_dir)
sys_args = sys.argv[:]
for i, arg in enumerate(sys.argv):
if 'log-path' in arg:
nose_argv += ['--nologcapture']
else:
for tests_type in CTRexScenario.test_types.keys():
if tests_type in arg:
specific_tests = True
CTRexScenario.test_types[tests_type].append(arg[arg.find(tests_type):])
sys_args.remove(arg)
if not specific_tests:
for key in ('--func', '--functional'):
if key in sys_args:
CTRexScenario.test_types['functional_tests'].append('functional_tests')
sys_args.remove(key)
for key in ('--stf', '--stateful'):
if key in sys_args:
CTRexScenario.test_types['stateful_tests'].append('stateful_tests')
sys_args.remove(key)
for key in ('--stl', '--stateless'):
if key in sys_args:
CTRexScenario.test_types['stateless_tests'].append('stateless_tests')
sys_args.remove(key)
# Run all of the tests or just the selected ones
if not sum([len(x) for x in CTRexScenario.test_types.values()]):
for key in CTRexScenario.test_types.keys():
CTRexScenario.test_types[key].append(key)
nose_argv += sys_args
addplugins = [RedNose(), CTRexTestConfiguringPlugin()]
result = True
try:
if len(CTRexScenario.test_types['functional_tests']):
additional_args = ['--func'] + CTRexScenario.test_types['functional_tests']
if xml_arg:
additional_args += ['--with-xunit', xml_arg.replace('.xml', '_functional.xml')]
result = nose.run(argv = nose_argv + additional_args, addplugins = addplugins)
if len(CTRexScenario.test_types['stateless_tests']):
additional_args = ['--stl', 'stateless_tests/stl_general_test.py:STLBasic_Test.test_connectivity'] + CTRexScenario.test_types['stateless_tests']
if not test_client_package:
additional_args.extend(['-a', '!client_package'])
if xml_arg:
additional_args += ['--with-xunit', xml_arg.replace('.xml', '_stateless.xml')]
result = nose.run(argv = nose_argv + additional_args, addplugins = addplugins) and result
if len(CTRexScenario.test_types['stateful_tests']):
additional_args = ['--stf']
if '--warmup' in sys.argv:
additional_args.append('stateful_tests/trex_imix_test.py:CTRexIMIX_Test.test_warm_up')
additional_args += CTRexScenario.test_types['stateful_tests']
if not test_client_package:
additional_args.extend(['-a', '!client_package'])
if xml_arg:
additional_args += ['--with-xunit', xml_arg.replace('.xml', '_stateful.xml')]
result = nose.run(argv = nose_argv + additional_args, addplugins = addplugins) and result
#except Exception as e:
# result = False
# print(e)
finally:
save_setup_info()
if not CTRexScenario.is_test_list:
if result == True:
print(termstyle.green("""
..::''''::..
.;'' ``;.
:: :: :: ::
:: :: :: ::
:: :: :: ::
:: .:' :: :: `:. ::
:: : : ::
:: `:. .:' ::
`;..``::::''..;'
``::,,,,::''
___ ___ __________
/ _ \/ _ | / __/ __/ /
/ ___/ __ |_\ \_\ \/_/
/_/ /_/ |_/___/___(_)
"""))
sys.exit(0)
else:
print(termstyle.red("""
/\_/\
( o.o )
> ^ <
This cat is sad, test failed.
"""))
sys.exit(-1)