#!/usr/bin/env python __copyright__ = "Copyright 2014" """ Name: trex_unit_test.py Description: This script creates the functionality to test the performance of the T-Rex traffic generator The tested scenario is a T-Rex TG directly connected to a Cisco router. :: Topology: ------- -------- | | Tx---1gig/10gig----Rx | | | T-Rex | | router | | | Rx---1gig/10gig----Tx | | ------- -------- """ import os import sys import outer_packages import nose from nose.plugins import Plugin import logging import CustomLogger import misc_methods from rednose import RedNose import termstyle from unit_tests.trex_general_test import CTRexScenario from client.trex_client import * from common.trex_exceptions import * import trex import socket from pprint import pprint import subprocess import re 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 /scripts if not latest_build_path: latest_build_path = check_trex_path(os.path.join(os.pardir, os.pardir)) if not latest_build_path: raise Exception('Could not determine trex_under_test folder, try setting env.var. TREX_UNDER_TEST') return latest_build_path def _start_stop_trex_remote_server(trex_data, command): # start t-rex server as daemon process # subprocess.call(["/usr/bin/python", "trex_daemon_server", "restart"], cwd = trex_latest_build) misc_methods.run_remote_command(trex_data['trex_name'], trex_data['trex_password'], command) def start_trex_remote_server(trex_data, kill_running = False): if kill_running: (return_code, stdout, stderr) = misc_methods.run_remote_command(trex_data['trex_name'], trex_data['trex_password'], 'ps -u root --format comm,pid,cmd | grep t-rex-64') if stdout: for process in stdout.split('\n'): try: proc_name, pid, full_cmd = re.split('\s+', process, maxsplit=2) if proc_name.find('t-rex-64') >= 0: print 'Killing remote process: %s' % full_cmd misc_methods.run_remote_command(trex_data['trex_name'], trex_data['trex_password'], 'kill %s' % pid) except: continue _start_stop_trex_remote_server(trex_data, DAEMON_START_COMMAND) def stop_trex_remote_server(trex_data): _start_stop_trex_remote_server(trex_data, DAEMON_STOP_COMMAND) class CTRexTestConfiguringPlugin(Plugin): 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('--verbose-mode', '--verbose_mode', action="store_true", default = False, dest="verbose_mode", help="Print RPC command and router commands.") 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('--functional', action="store_true", default = False, dest="functional", help="Don't connect to remote server for runnning daemon (For functional tests).") def configure(self, options, conf): self.functional = options.functional if self.functional: return if CTRexScenario.setup_dir and options.config_path: raise Exception('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 options.config_path: self.configuration = misc_methods.load_complete_config_file(os.path.join(options.config_path, 'config.yaml')) self.benchmark = misc_methods.load_benchmark_config_file(os.path.join(options.config_path, 'benchmark.yaml')) self.enabled = True else: raise Exception('Please specify path to config.yaml using --cfg parameter or env. variable SETUP_DIR') self.modes = self.configuration.trex.get('modes', []) self.kill_running = options.kill_running self.load_image = options.load_image self.verbose_mode = options.verbose_mode self.clean_config = False if options.skip_clean_config else True self.server_logs = options.server_logs if options.log_path: self.loggerPath = options.log_path def begin (self): if self.functional: return # initialize CTRexScenario global testing class, to be used by all tests CTRexScenario.configuration = self.configuration CTRexScenario.benchmark = self.benchmark CTRexScenario.modes = set(self.modes) CTRexScenario.server_logs = self.server_logs # launch TRex daemon on relevant setup start_trex_remote_server(self.configuration.trex, self.kill_running) CTRexScenario.trex = CTRexClient(trex_host = self.configuration.trex['trex_name'], verbose = self.verbose_mode) if 'loopback' not in self.modes: CTRexScenario.router_cfg = dict( config_dict = self.configuration.router, forceImageReload = self.load_image, silent_mode = not self.verbose_mode, forceCleanConfig = self.clean_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): if self.functional: return CTRexScenario.is_init = False stop_trex_remote_server(self.configuration.trex) 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) 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 def set_report_dir (report_dir): if not os.path.exists(report_dir): os.mkdir(report_dir) if __name__ == "__main__": # setting defaults. By default we run all the test suite specific_tests = False disableLogCapture = False long_test = False xml_name = 'unit_test.xml' CTRexScenario.report_dir = 'reports' CTRexScenario.scripts_path = get_trex_path() DAEMON_STOP_COMMAND = 'cd %s; ./trex_daemon_server stop; sleep 1; ./trex_daemon_server stop;' % CTRexScenario.scripts_path DAEMON_START_COMMAND = DAEMON_STOP_COMMAND + 'sleep 1; rm /var/log/trex/trex_daemon_server.log; ./trex_daemon_server start; sleep 2; ./trex_daemon_server show' setup_dir = os.getenv('SETUP_DIR', '').rstrip('/') CTRexScenario.setup_dir = check_setup_path(setup_dir) if not CTRexScenario.setup_dir: CTRexScenario.setup_dir = check_setup_path(os.path.join('setups', setup_dir)) if CTRexScenario.setup_dir: 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: if 'unit_tests/' in arg: specific_tests = True 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) # Run all of the unit tests or just the selected ones if not specific_tests: nose_argv += ['unit_tests'] if disableLogCapture: nose_argv += ['--nologcapture'] try: config_plugin = CTRexTestConfiguringPlugin() red_nose = RedNose() try: result = nose.run(argv = nose_argv, addplugins = [red_nose, config_plugin]) except socket.error: # handle consecutive tests exception, try once again print "TRex connectivity error identified. Possibly due to consecutive nightly runs.\nRetrying..." result = nose.run(argv = nose_argv, addplugins = [red_nose, config_plugin]) finally: save_setup_info() if (result == True and not CTRexScenario.is_test_list): print termstyle.green(""" ..::''''::.. .;'' ``;. :: :: :: :: :: :: :: :: :: :: :: :: :: .:' :: :: `:. :: :: : : :: :: `:. .:' :: `;..``::::''..;' ``::,,,,::'' ___ ___ __________ / _ \/ _ | / __/ __/ / / ___/ __ |_\ \_\ \/_/ /_/ /_/ |_/___/___(_) """) sys.exit(0) else: sys.exit(-1) finally: pass