diff options
Diffstat (limited to 'scripts/automation/regression/unit_tests')
17 files changed, 1705 insertions, 0 deletions
diff --git a/scripts/automation/regression/unit_tests/__init__.py b/scripts/automation/regression/unit_tests/__init__.py new file mode 100755 index 00000000..8b137891 --- /dev/null +++ b/scripts/automation/regression/unit_tests/__init__.py @@ -0,0 +1 @@ + diff --git a/scripts/automation/regression/unit_tests/functional_tests/config.yaml b/scripts/automation/regression/unit_tests/functional_tests/config.yaml new file mode 100644 index 00000000..4f4c7c40 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/config.yaml @@ -0,0 +1,74 @@ +################################################################ +#### T-Rex nightly test configuration file #### +################################################################ + + +### T-Rex configuration: +# hostname - can be DNS name or IP for the TRex machine for ssh to the box +# password - root password for TRex machine +# is_dual - should the TRex inject with -p ? +# version_path - path to the t-rex version and executable +# cores - how many cores should be used +# latency - rate of latency packets injected by the TRex + +### Router configuration: +# hostname - the router hostname as apears in ______# cli prefix +# ip_address - the router's ip that can be used to communicate with +# image - the desired imaged wished to be loaded as the router's running config +# line_password - router password when access via Telent +# en_password - router password when changing to "enable" mode +# interfaces - an array of client-server pairs, representing the interfaces configurations of the router +# configurations - an array of configurations that could possibly loaded into the router during the test. +# The "clean" configuration is a mandatory configuration the router will load with to run the basic test bench + +### TFTP configuration: +# hostname - the tftp hostname +# ip_address - the tftp's ip address +# images_path - the tftp's relative path in which the router's images are located + +### Test_misc configuration: +# expected_bw - the "golden" bandwidth (in Gbps) results planned on receiving from the test + +trex: + hostname : hostname + password : root password + version_path : not used + cores : 1 + +router: + model : device model + hostname : device hostname + ip_address : device ip + image : device image name + line_password : telnet pass + en_password : enable pass + mgmt_interface : GigabitEthernet0/0/0 + clean_config : path to clean_config file + intf_masking : 255.255.255.0 + ipv6_mask : 64 + interfaces : + - client : + name : GigabitEthernet0/0/1 + src_mac_addr : 0000.0001.0000 + dest_mac_addr : 0000.1000.0000 + server : + name : GigabitEthernet0/0/2 + src_mac_addr : 0000.0002.0000 + dest_mac_addr : 0000.2000.0000 + vrf_name : null + - client : + name : GigabitEthernet0/0/3 + src_mac_addr : 0000.0003.0000 + dest_mac_addr : 0000.3000.0000 + server : + name : GigabitEthernet0/0/4 + src_mac_addr : 0000.0004.0000 + dest_mac_addr : 0000.4000.0000 + vrf_name : dup + + +tftp: + hostname : tftp hostname + ip_address : tftp ip + root_dir : tftp root dir + images_path : path related to root dir diff --git a/scripts/automation/regression/unit_tests/functional_tests/functional_general_test.py b/scripts/automation/regression/unit_tests/functional_tests/functional_general_test.py new file mode 100755 index 00000000..525b58d2 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/functional_general_test.py @@ -0,0 +1,22 @@ +#!/router/bin/python + +from nose.tools import assert_equal +from nose.tools import assert_not_equal +from nose.tools import assert_raises +from nose.tools import raises + + +class CGeneralFunctional_Test(object): + def __init__(self): + pass + + + def setUp(self): + pass + + + def tearDown(self): + pass + +if __name__ == "__main__": + pass diff --git a/scripts/automation/regression/unit_tests/functional_tests/misc_methods_test.py b/scripts/automation/regression/unit_tests/functional_tests/misc_methods_test.py new file mode 100755 index 00000000..096f86d8 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/misc_methods_test.py @@ -0,0 +1,61 @@ +#!/router/bin/python + +import functional_general_test +import misc_methods +from nose.tools import assert_equal +from nose.tools import assert_not_equal +from nose.tools import assert_raises +from nose.tools import raises + + +class MiscMethods_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.ipv4_gen = misc_methods.get_network_addr() + self.ipv6_gen = misc_methods.get_network_addr(ip_type = 'ipv6') + pass + + def test_ipv4_gen(self): + for i in range(1, 255): + assert_equal( next(self.ipv4_gen), [".".join( map(str, [1, 1, i, 0])), '255.255.255.0'] ) + + def test_ipv6_gen(self): + tmp_ipv6_addr = ['2001', 'DB8', 0, '2222', 0, 0, 0, 0] + for i in range(0, 255): + tmp_ipv6_addr[2] = hex(i)[2:] + assert_equal( next(self.ipv6_gen), ":".join( map(str, tmp_ipv6_addr)) ) + + def test_get_ipv4_client_addr(self): + tmp_ipv4_addr = next(self.ipv4_gen)[0] + assert_equal ( misc_methods.get_single_net_client_addr(tmp_ipv4_addr), '1.1.1.1') + assert_raises (ValueError, misc_methods.get_single_net_client_addr, tmp_ipv4_addr, {'3' : 255} ) + + def test_get_ipv6_client_addr(self): + tmp_ipv6_addr = next(self.ipv6_gen) + assert_equal ( misc_methods.get_single_net_client_addr(tmp_ipv6_addr, {'7' : 1}, ip_type = 'ipv6'), '2001:DB8:0:2222:0:0:0:1') + assert_equal ( misc_methods.get_single_net_client_addr(tmp_ipv6_addr, {'7' : 2}, ip_type = 'ipv6'), '2001:DB8:0:2222:0:0:0:2') + assert_raises (ValueError, misc_methods.get_single_net_client_addr, tmp_ipv6_addr, {'7' : 70000} ) + + + @raises(ValueError) + def test_ipv4_client_addr_exception(self): + tmp_ipv4_addr = next(self.ipv4_gen)[0] + misc_methods.get_single_net_client_addr(tmp_ipv4_addr, {'4' : 1}) + + @raises(ValueError) + def test_ipv6_client_addr_exception(self): + tmp_ipv6_addr = next(self.ipv6_gen) + misc_methods.get_single_net_client_addr(tmp_ipv6_addr, {'8' : 1}, ip_type = 'ipv6') + + @raises(StopIteration) + def test_gen_ipv4_to_limit (self): + while(True): + next(self.ipv4_gen) + + @raises(StopIteration) + def test_gen_ipv6_to_limit (self): + while(True): + next(self.ipv6_gen) + + def tearDown(self): + pass diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_cache_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_cache_test.py new file mode 100755 index 00000000..24ccf7a5 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_cache_test.py @@ -0,0 +1,60 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CCommandCache_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.cache = CCommandCache() + self.cache.add('IF', "ip nbar protocol-discovery", 'GigabitEthernet0/0/1') + self.cache.add('IF', "ip nbar protocol-discovery", 'GigabitEthernet0/0/2') + self.cache.add('conf', "arp 1.1.1.1 0000.0001.0000 arpa") + self.cache.add('conf', "arp 1.1.2.1 0000.0002.0000 arpa") + self.cache.add('exec', "show ip nbar protocol-discovery stats packet-count") + + def test_add(self): + assert_equal(self.cache.cache['IF'], + {'GigabitEthernet0/0/1' : ['ip nbar protocol-discovery'], + 'GigabitEthernet0/0/2' : ['ip nbar protocol-discovery'] + }) + assert_equal(self.cache.cache['CONF'], + ["arp 1.1.1.1 0000.0001.0000 arpa", + "arp 1.1.2.1 0000.0002.0000 arpa"] + ) + assert_equal(self.cache.cache['EXEC'], + ["show ip nbar protocol-discovery stats packet-count"]) + + def test_dump_config (self): + import sys + from StringIO import StringIO + saved_stdout = sys.stdout + try: + out = StringIO() + sys.stdout = out + self.cache.dump_config() + output = out.getvalue().strip() + assert_equal(output, + "configure terminal\ninterface GigabitEthernet0/0/1\nip nbar protocol-discovery\ninterface GigabitEthernet0/0/2\nip nbar protocol-discovery\nexit\narp 1.1.1.1 0000.0001.0000 arpa\narp 1.1.2.1 0000.0002.0000 arpa\nexit\nshow ip nbar protocol-discovery stats packet-count" + ) + finally: + sys.stdout = saved_stdout + + def test_get_config_list (self): + assert_equal(self.cache.get_config_list(), + ["configure terminal", "interface GigabitEthernet0/0/1", "ip nbar protocol-discovery", "interface GigabitEthernet0/0/2", "ip nbar protocol-discovery", "exit", "arp 1.1.1.1 0000.0001.0000 arpa", "arp 1.1.2.1 0000.0002.0000 arpa", "exit", "show ip nbar protocol-discovery stats packet-count"] + ) + + def test_clear_cache (self): + self.cache.clear_cache() + assert_equal(self.cache.cache, + {"IF" : {}, + "CONF" : [], + "EXEC" : []} + ) + + def tearDown(self): + self.cache.clear_cache() diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_link_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_link_test.py new file mode 100755 index 00000000..7a31815b --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_cmd_link_test.py @@ -0,0 +1,62 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CCommandLink_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.cache = CCommandCache() + self.cache.add('IF', "ip nbar protocol-discovery", 'GigabitEthernet0/0/1') + self.cache.add('IF', "ip nbar protocol-discovery", 'GigabitEthernet0/0/2') + self.cache.add('conf', "arp 1.1.1.1 0000.0001.0000 arpa") + self.cache.add('conf', "arp 1.1.2.1 0000.0002.0000 arpa") + self.cache.add('exec', "show ip nbar protocol-discovery stats packet-count") + self.com_link = CCommandLink() + + def test_transmit(self): + # test here future implemntatin of platform physical link + pass + + def test_run_cached_command (self): + self.com_link.run_command([self.cache]) + + assert_equal (self.com_link.get_history(), + ["configure terminal", "interface GigabitEthernet0/0/1", "ip nbar protocol-discovery", "interface GigabitEthernet0/0/2", "ip nbar protocol-discovery", "exit", "arp 1.1.1.1 0000.0001.0000 arpa", "arp 1.1.2.1 0000.0002.0000 arpa", "exit", "show ip nbar protocol-discovery stats packet-count"] + ) + + self.com_link.clear_history() + self.com_link.run_single_command(self.cache) + assert_equal (self.com_link.get_history(), + ["configure terminal", "interface GigabitEthernet0/0/1", "ip nbar protocol-discovery", "interface GigabitEthernet0/0/2", "ip nbar protocol-discovery", "exit", "arp 1.1.1.1 0000.0001.0000 arpa", "arp 1.1.2.1 0000.0002.0000 arpa", "exit", "show ip nbar protocol-discovery stats packet-count"] + ) + + def test_run_single_command(self): + self.com_link.run_single_command("show ip nbar protocol-discovery stats packet-count") + assert_equal (self.com_link.get_history(), + ["show ip nbar protocol-discovery stats packet-count"] + ) + + def test_run_mixed_commands (self): + self.com_link.run_single_command("show ip nbar protocol-discovery stats packet-count") + self.com_link.run_command([self.cache]) + self.com_link.run_command(["show ip interface brief"]) + + assert_equal (self.com_link.get_history(), + ["show ip nbar protocol-discovery stats packet-count", + "configure terminal", "interface GigabitEthernet0/0/1", "ip nbar protocol-discovery", "interface GigabitEthernet0/0/2", "ip nbar protocol-discovery", "exit", "arp 1.1.1.1 0000.0001.0000 arpa", "arp 1.1.2.1 0000.0002.0000 arpa", "exit", "show ip nbar protocol-discovery stats packet-count", + "show ip interface brief"] + ) + + def test_clear_history (self): + self.com_link.run_command(["show ip interface brief"]) + self.com_link.clear_history() + assert_equal (self.com_link.get_history(), []) + + def tearDown(self): + self.cache.clear_cache() + + diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_device_cfg_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_device_cfg_test.py new file mode 100755 index 00000000..890d0cb9 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_device_cfg_test.py @@ -0,0 +1,20 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CDeviceCfg_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.dev_cfg = CDeviceCfg('./unit_tests/functional_tests/config.yaml') + + def test_get_interfaces_cfg(self): + assert_equal (self.dev_cfg.get_interfaces_cfg(), + [{'client': {'src_mac_addr': '0000.0001.0000', 'name': 'GigabitEthernet0/0/1', 'dest_mac_addr': '0000.1000.0000'}, 'vrf_name': None, 'server': {'src_mac_addr': '0000.0002.0000', 'name': 'GigabitEthernet0/0/2', 'dest_mac_addr': '0000.2000.0000'}}, {'client': {'src_mac_addr': '0000.0003.0000', 'name': 'GigabitEthernet0/0/3', 'dest_mac_addr': '0000.3000.0000'}, 'vrf_name': 'dup', 'server': {'src_mac_addr': '0000.0004.0000', 'name': 'GigabitEthernet0/0/4', 'dest_mac_addr': '0000.4000.0000'}}] + ) + + def tearDown(self): + pass diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_dual_if_obj_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_dual_if_obj_test.py new file mode 100755 index 00000000..ff54b9ee --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_dual_if_obj_test.py @@ -0,0 +1,31 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CDualIfObj_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.if_1 = CIfObj('gig0/0/1', '1.1.1.1', '2001:DB8:0:2222:0:0:0:1', '0000.0001.0000', '0000.0001.0000', IFType.Client) + self.if_2 = CIfObj('gig0/0/2', '1.1.2.1', '2001:DB8:1:2222:0:0:0:1', '0000.0002.0000', '0000.0002.0000', IFType.Server) + self.if_3 = CIfObj('gig0/0/3', '1.1.3.1', '2001:DB8:2:2222:0:0:0:1', '0000.0003.0000', '0000.0003.0000', IFType.Client) + self.if_4 = CIfObj('gig0/0/4', '1.1.4.1', '2001:DB8:3:2222:0:0:0:1', '0000.0004.0000', '0000.0004.0000', IFType.Server) + self.dual_1 = CDualIfObj(None, self.if_1, self.if_2) + self.dual_2 = CDualIfObj('dup', self.if_3, self.if_4) + + def test_id_allocation(self): + assert (self.dual_1.get_id() < self.dual_2.get_id() < CDualIfObj._obj_id) + + def test_get_vrf_name (self): + assert_equal ( self.dual_1.get_vrf_name() , None ) + assert_equal ( self.dual_2.get_vrf_name() , 'dup' ) + + def test_is_duplicated (self): + assert_equal ( self.dual_1.is_duplicated() , False ) + assert_equal ( self.dual_2.is_duplicated() , True ) + + def tearDown(self): + pass
\ No newline at end of file diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_if_manager_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_if_manager_test.py new file mode 100755 index 00000000..7ba6e66e --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_if_manager_test.py @@ -0,0 +1,40 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CIfManager_Test(functional_general_test.CGeneralFunctional_Test): + + def setUp(self): + self.dev_cfg = CDeviceCfg('./unit_tests/functional_tests/config.yaml') + self.if_mng = CIfManager() + + # main testing method to check the entire class + def test_load_config (self): + self.if_mng.load_config(self.dev_cfg) + + # check the number of items in each qeury + assert_equal( len(self.if_mng.get_if_list()), 4 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Client)), 2 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Client, is_duplicated = True)), 1 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Client, is_duplicated = False)), 1 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Server)), 2 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Server, is_duplicated = True)), 1 ) + assert_equal( len(self.if_mng.get_if_list(if_type = IFType.Server, is_duplicated = False)), 1 ) + assert_equal( len(self.if_mng.get_duplicated_if()), 2 ) + assert_equal( len(self.if_mng.get_dual_if_list()), 2 ) + + # check the classification with intf name + assert_equal( map(CIfObj.get_name, self.if_mng.get_if_list() ), ['GigabitEthernet0/0/1','GigabitEthernet0/0/2','GigabitEthernet0/0/3','GigabitEthernet0/0/4'] ) + assert_equal( map(CIfObj.get_name, self.if_mng.get_if_list(is_duplicated = True) ), ['GigabitEthernet0/0/3','GigabitEthernet0/0/4'] ) + assert_equal( map(CIfObj.get_name, self.if_mng.get_if_list(is_duplicated = False) ), ['GigabitEthernet0/0/1','GigabitEthernet0/0/2'] ) + assert_equal( map(CIfObj.get_name, self.if_mng.get_duplicated_if() ), ['GigabitEthernet0/0/3', 'GigabitEthernet0/0/4'] ) + + # check the classification with vrf name + assert_equal( map(CDualIfObj.get_vrf_name, self.if_mng.get_dual_if_list() ), [None, 'dup'] ) + + def tearDown(self): + pass diff --git a/scripts/automation/regression/unit_tests/functional_tests/platform_if_obj_test.py b/scripts/automation/regression/unit_tests/functional_tests/platform_if_obj_test.py new file mode 100755 index 00000000..534d4170 --- /dev/null +++ b/scripts/automation/regression/unit_tests/functional_tests/platform_if_obj_test.py @@ -0,0 +1,49 @@ +#!/router/bin/python + +from platform_cmd_link import * +import functional_general_test +from nose.tools import assert_equal +from nose.tools import assert_not_equal + + +class CIfObj_Test(functional_general_test.CGeneralFunctional_Test): + test_idx = 1 + + def setUp(self): + self.if_1 = CIfObj('gig0/0/1', '1.1.1.1', '2001:DB8:0:2222:0:0:0:1', '0000.0001.0000', '0000.0001.0000', IFType.Client) + self.if_2 = CIfObj('TenGig0/0/0', '1.1.2.1', '2001:DB8:1:2222:0:0:0:1', '0000.0002.0000', '0000.0002.0000', IFType.Server) + CIfObj_Test.test_idx += 1 + + def test_id_allocation(self): + assert (self.if_1.get_id() < self.if_2.get_id() < CIfObj._obj_id) + + def test_isClient(self): + assert_equal (self.if_1.is_client(), True) + + def test_isServer(self): + assert_equal (self.if_2.is_server(), True) + + def test_get_name (self): + assert_equal (self.if_1.get_name(), 'gig0/0/1') + assert_equal (self.if_2.get_name(), 'TenGig0/0/0') + + def test_get_src_mac_addr (self): + assert_equal (self.if_1.get_src_mac_addr(), '0000.0001.0000') + + def test_get_dest_mac (self): + assert_equal (self.if_2.get_dest_mac(), '0000.0002.0000') + + def test_get_ipv4_addr (self): + assert_equal (self.if_1.get_ipv4_addr(), '1.1.1.1' ) + assert_equal (self.if_2.get_ipv4_addr(), '1.1.2.1' ) + + def test_get_ipv6_addr (self): + assert_equal (self.if_1.get_ipv6_addr(), '2001:DB8:0:2222:0:0:0:1' ) + assert_equal (self.if_2.get_ipv6_addr(), '2001:DB8:1:2222:0:0:0:1' ) + + def test_get_type (self): + assert_equal (self.if_1.get_if_type(), IFType.Client) + assert_equal (self.if_2.get_if_type(), IFType.Server) + + def tearDown(self): + pass diff --git a/scripts/automation/regression/unit_tests/tests_exceptions.py b/scripts/automation/regression/unit_tests/tests_exceptions.py new file mode 100755 index 00000000..604efcc8 --- /dev/null +++ b/scripts/automation/regression/unit_tests/tests_exceptions.py @@ -0,0 +1,37 @@ +#!/router/bin/python + +class TRexInUseError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) + +class TRexRunFailedError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) + +class TRexIncompleteRunError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) + +class TRexLowCpuUtilError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) + +class AbnormalResultError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) + +class ClassificationMissmatchError(Exception): + def __init__(self, value): + self.value = value + def __str__(self): + return repr(self.value) diff --git a/scripts/automation/regression/unit_tests/trex_general_test.py b/scripts/automation/regression/unit_tests/trex_general_test.py new file mode 100755 index 00000000..6a6ad79c --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_general_test.py @@ -0,0 +1,328 @@ +#!/router/bin/python + +__copyright__ = "Copyright 2014" + +""" +Name: + trex_general_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 | | + ------- -------- + +""" +from nose.plugins import Plugin +from nose.plugins.skip import SkipTest +import trex +import misc_methods +import sys +import os +# from CPlatformUnderTest import * +from CPlatform import * +import termstyle +import threading +from tests_exceptions import * +from platform_cmd_link import * +import unittest + + +class CTRexScenario(): + modes = set() # list of modes of this setup: loopback, virtual etc. + server_logs = False + is_test_list = False + is_init = False + trex_crashed = False + configuration = None + trex = None + router = None + router_cfg = None + daemon_log_lines = 0 + setup_name = None + setup_dir = None + router_image = None + trex_version = None + report_dir = 'reports' + # logger = None + +#scenario = CTRexScenario() + +def setUpModule(module): +# print ("") # this is to get a newline after the dots +# print ("setup_module before anything in this file") +# # ff = CTRexScenario() +# scenario.configuration = misc_methods.load_complete_config_file('config/config.yaml') +# scenario.trex = trex.CTRexRunner(scenario.configuration[0], None) +# scenario.router = CPlatform(scenario.configuration[1], False, scenario.configuration[2]) +# scenario.router.platform.preCheck() +# print "Done instantiating trex scenario!" + pass + +def tearDownModule(module): +# print ("") # this is to get a newline after the dots +# scenario.router.platform.postCheck() +# print ("teardown_module after anything in this file") + pass + + + +class CTRexGeneral_Test(unittest.TestCase): + """This class defines the general testcase of the T-Rex traffic generator""" + def __init__ (self, *args, **kwargs): + unittest.TestCase.__init__(self, *args, **kwargs) + # Point test object to scenario global object + self.configuration = CTRexScenario.configuration + self.benchmark = CTRexScenario.benchmark + self.trex = CTRexScenario.trex + self.trex_crashed = CTRexScenario.trex_crashed + self.modes = CTRexScenario.modes + self.skipping = False + self.fail_reasons = [] + if not hasattr(self, 'unsupported_modes'): + self.unsupported_modes = [] + self.is_loopback = True if 'loopback' in self.modes else False + self.is_virt_nics = True if 'virt_nics' in self.modes else False + self.is_VM = True if 'VM' in self.modes else False + + if not CTRexScenario.is_init: + CTRexScenario.trex_version = self.trex.get_trex_version() + if not self.is_loopback: + # initilize the scenario based on received configuration, once per entire testing session + CTRexScenario.router = CPlatform(CTRexScenario.router_cfg['silent_mode']) + device_cfg = CDeviceCfg() + device_cfg.set_platform_config(CTRexScenario.router_cfg['config_dict']) + device_cfg.set_tftp_config(CTRexScenario.router_cfg['tftp_config_dict']) + CTRexScenario.router.load_platform_data_from_file(device_cfg) + CTRexScenario.router.launch_connection(device_cfg) + running_image = CTRexScenario.router.get_running_image_details()['image'] + print 'Current router image: %s' % running_image + if CTRexScenario.router_cfg['forceImageReload']: + needed_image = device_cfg.get_image_name() + if not CTRexScenario.router.is_image_matches(needed_image): + print 'Setting router image: %s' % needed_image + CTRexScenario.router.config_tftp_server(device_cfg) + CTRexScenario.router.load_platform_image(needed_image) + CTRexScenario.router.set_boot_image(needed_image) + CTRexScenario.router.reload_platform(device_cfg) + CTRexScenario.router.launch_connection(device_cfg) + running_image = CTRexScenario.router.get_running_image_details()['image'] # verify image + if not CTRexScenario.router.is_image_matches(needed_image): + self.fail('Unable to set router image: %s, current image is: %s' % (needed_image, running_image)) + else: + print 'Matches needed image: %s' % needed_image + CTRexScenario.router_image = running_image + + + if self.modes: + print termstyle.green('\t!!!\tRunning with modes: %s, not suitable tests will be skipped.\t!!!' % list(self.modes)) + + CTRexScenario.is_init = True + print termstyle.green("Done instantiating T-Rex scenario!\n") + +# raise RuntimeError('CTRexScenario class is not initialized!') + self.router = CTRexScenario.router + + + +# def assert_dict_eq (self, dict, key, val, error=''): +# v1 = int(dict[key])) +# self.assertEqual(v1, int(val), error) +# +# def assert_dict_gt (self, d, key, val, error=''): +# v1 = int(dict[key]) +# self.assert_gt(v1, int(val), error) + + def assertEqual(self, v1, v2, s): + if v1 != v2: + error='ERROR '+str(v1)+' != '+str(v2)+ ' '+s; + self.fail(error) + + def assert_gt(self, v1, v2, s): + if not v1 > v2: + error='ERROR {big} > {small} {str}'.format(big = v1, small = v2, str = s) + self.fail(error) + + def check_results_eq (self,res,name,val): + if res is None: + self.fail('TRex results cannot be None !') + return + + if name not in res: + self.fail('TRex results does not include key %s' % name) + return + + if res[name] != float(val): + self.fail('TRex results[%s]==%f and not as expected %f ' % (name, res[name], val)) + + def check_CPU_benchmark (self, trex_res, err): + #cpu_util = float(trex_res.get_last_value("trex-global.data.m_cpu_util")) + cpu_util = sum([float(x) for x in trex_res.get_value_list("trex-global.data.m_cpu_util")[-4:-1]]) / 3 # mean of 3 values before last + + if cpu_util < 30 and not self.is_virt_nics: + self.fail("CPU is too low (%s%%), can't verify performance in such low CPU%%." % cpu_util ) + + cores = self.get_benchmark_param('cores') + trex_tx_bps = trex_res.get_last_value("trex-global.data.m_total_tx_bytes") + test_norm_cpu = 100.0*(trex_tx_bps/(cores*cpu_util))/1e6 + + print "TRex CPU utilization: %g%%, norm_cpu is : %d Mb/core" % (round(cpu_util), int(test_norm_cpu)) + + #expected_norm_cpu = self.get_benchmark_param('cpu_to_core_ratio') + + #calc_error_precent = abs(100.0*(test_norm_cpu/expected_norm_cpu)-100.0) + +# if calc_error_precent > err: +# msg ='Normalized bandwidth to CPU utilization ratio is %2.0f Mb/core expected %2.0f Mb/core more than %2.0f %% - ERROR' % (test_norm_cpu, expected_norm_cpu, err) +# raise AbnormalResultError(msg) +# else: +# msg ='Normalized bandwidth to CPU utilization ratio is %2.0f Mb/core expected %2.0f Mb/core less than %2.0f %% - OK' % (test_norm_cpu, expected_norm_cpu, err) +# print msg + + + def check_results_gt (self, res, name, val): + if res is None: + self.fail('TRex results canot be None !') + return + + if name not in res: + self.fail('TRex results does not include key %s' % name) + return + + if res[name]< float(val): + self.fail('TRex results[%s]<%f and not as expected greater than %f ' % (name, res[name], val)) + + def check_for_trex_crash(self): + pass + + def get_benchmark_param (self, param, sub_param = None, test_name = None): + if not test_name: + test_name = self.get_name() + if test_name not in self.benchmark: + self.skip('No data in benchmark.yaml for test %s, skipping.' % test_name) + if sub_param: + return self.benchmark[test_name][param].get(sub_param) + else: + return self.benchmark[test_name].get(param) + + def check_general_scenario_results (self, trex_res, check_latency = True): + + try: + # check if test is valid + if not trex_res.is_done_warmup(): + self.fail('T-Rex did not reach warm-up situtaion. Results are not valid.') + + # check T-Rex number of drops + trex_tx_pckt = trex_res.get_last_value("trex-global.data.m_total_tx_pkts") + trex_drops = trex_res.get_total_drops() + trex_drop_rate = trex_res.get_drop_rate() + if ( (trex_drops/trex_tx_pckt) > 0.001) and (trex_drop_rate > 0.0): # deliberately mask kickoff drops when T-Rex first initiated + self.fail('Number of packet drops larger than 0.1% of all traffic') + + # # check T-Rex expected counters + #trex_exp_rate = trex_res.get_expected_tx_rate().get('m_tx_expected_bps') + #assert trex_exp_rate is not None + #trex_exp_gbps = trex_exp_rate/(10**9) + + if check_latency: + # check that max latency does not exceed 1 msec in regular setup or 20ms in VM + allowed_latency = 20000 if self.is_VM else 1000 + if max(trex_res.get_max_latency().values()) > allowed_latency: + print 'LatencyError: Maximal latency exceeds %s (usec)' % allowed_latency + #raise AbnormalResultError('Maximal latency above 1ms') + + # check that avg latency does not exceed 1 msec in regular setup or 3ms in VM + allowed_latency = 3000 if self.is_VM else 1000 + if max(trex_res.get_avg_latency().values()) > allowed_latency: + print 'LatencyError: Average latency exceeds %s (usec)' % allowed_latency + #raise AbnormalResultError('Maximal latency above 1ms') + + if not self.is_loopback: + # check router number of drops --> deliberately masked- need to be figured out!!!!! + pkt_drop_stats = self.router.get_drop_stats() +# assert pkt_drop_stats['total_drops'] < 20 + + # check for trex-router packet consistency + # TODO: check if it's ok + print 'router drop stats: %s' % pkt_drop_stats + print 'TRex drop stats: %s' % trex_drops + #self.assertEqual(pkt_drop_stats, trex_drops, "TRex's and router's drop stats don't match.") + + except KeyError as e: + self.fail(e) + #assert False + + # except AssertionError as e: + # e.args += ('T-Rex has crashed!') + # raise + + # We encountered error, don't fail the test immediately + def fail(self, reason = 'Unknown error'): + print 'Error: %s' % reason + self.fail_reasons.append(reason) + + # skip running of the test, counts as 'passed' but prints 'skipped' + def skip(self, message = ''): + self.skipping = True + raise SkipTest(message) + + # get name of currently running test + def get_name(self): + return self._testMethodName + + def setUp(self): + test_setup_modes_conflict = self.modes & set(self.unsupported_modes) + if test_setup_modes_conflict: + self.skip("The test can't run with following modes of given setup: %s " % test_setup_modes_conflict) + if not self.trex.is_idle(): + print 'Warning: TRex is not idle at setUp, trying to stop it.' + self.trex.force_kill(confirm = False) + if not self.is_loopback: + print '' + self.router.load_clean_config() + self.router.clear_counters() + self.router.clear_packet_drop_stats() + + ######################################################################## + #### DO NOT ADD TESTS TO THIS FILE #### + #### Added tests here will held once for EVERY test sub-class #### + ######################################################################## + + # masked example to such test. uncomment to watch how it affects # +# def test_isInitialized(self): +# assert CTRexScenario.is_init == True + def tearDown(self): + if not self.trex.is_idle(): + print 'Warning: TRex is not idle at tearDown, trying to stop it.' + self.trex.force_kill(confirm = False) + if not self.skipping: + # print server logs of test run + if CTRexScenario.server_logs: + try: + print termstyle.green('\n>>>>>>>>>>>>>>> Daemon log <<<<<<<<<<<<<<<') + daemon_log = self.trex.get_trex_daemon_log() + log_size = len(daemon_log) + print ''.join(daemon_log[CTRexScenario.daemon_log_lines:]) + CTRexScenario.daemon_log_lines = log_size + except Exception as e: + print "Can't get TRex daemon log:", e + try: + print termstyle.green('>>>>>>>>>>>>>>>> Trex log <<<<<<<<<<<<<<<<') + print ''.join(self.trex.get_trex_log()) + except Exception as e: + print "Can't get TRex log:", e + if len(self.fail_reasons): + raise Exception('The test is failed, reasons:\n%s' % '\n'.join(self.fail_reasons)) + + def check_for_trex_crash(self): + pass diff --git a/scripts/automation/regression/unit_tests/trex_imix_test.py b/scripts/automation/regression/unit_tests/trex_imix_test.py new file mode 100755 index 00000000..b56f7f4e --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_imix_test.py @@ -0,0 +1,176 @@ + +#!/router/bin/python +from trex_general_test import CTRexGeneral_Test +from CPlatform import CStaticRouteConfig +from tests_exceptions import * +#import sys +import time; + +class CTRexIMIX_Test(CTRexGeneral_Test): + """This class defines the IMIX testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + # super(CTRexIMIX_Test, self).__init__() + CTRexGeneral_Test.__init__(self, *args, **kwargs) + pass + + def setUp(self): + super(CTRexIMIX_Test, self).setUp() # launch super test class setUp process + # CTRexGeneral_Test.setUp(self) # launch super test class setUp process + # self.router.clear_counters() + pass + + def test_routing_imix_64(self): + # test initializtion + if not self.is_loopback: + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = "config") + +# self.trex.set_yaml_file('cap2/imix_64.yaml') + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + +# trex_res = self.trex.run(multiplier = mult, cores = core, duration = 30, l = 1000, p = True) + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + d = 30, + f = 'cap2/imix_64.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + + self.check_CPU_benchmark(trex_res, 10.0) + + # the name intentionally not matches nose default pattern, including the test should be specified explicitly + def dummy(self): + self.assertEqual(1, 2, 'boo') + self.assertEqual(2, 2, 'boo') + self.assertEqual(2, 3, 'boo') + #print '' + #print dir(self) + #print locals() + #print '' + #print_r(unittest.TestCase) + #print '' + #print_r(self) + print '' + #print unittest.TestCase.shortDescription(self) + #self.skip("I'm just a dummy test") + + + def test_routing_imix (self): + # test initializtion + if not self.is_loopback: + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = "config") + +# self.trex.set_yaml_file('cap2/imix_fast_1g.yaml') + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + d = 60, + f = 'cap2/imix_fast_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + + self.check_CPU_benchmark(trex_res, 10.0) + + + def test_static_routing_imix (self): + if self.is_loopback: # in loopback mode this test acts same as test_routing_imix, disable to avoid duplication + self.skip() + # test initializtion + if not self.is_loopback: + self.router.configure_basic_interfaces() + + # Configure static routing based on benchmark data input + stat_route_dict = self.get_benchmark_param('stat_route_dict') + stat_route_obj = CStaticRouteConfig(stat_route_dict) + self.router.config_static_routing(stat_route_obj, mode = "config") + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + d = 60, + f = 'cap2/imix_fast_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + print ("\nLATEST DUMP:") + print trex_res.get_latest_dump() + + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + + + def test_static_routing_imix_asymmetric (self): + # test initializtion + if not self.is_loopback: + self.router.configure_basic_interfaces() + + # Configure static routing based on benchmark data input + stat_route_dict = self.get_benchmark_param('stat_route_dict') + stat_route_obj = CStaticRouteConfig(stat_route_dict) + self.router.config_static_routing(stat_route_obj, mode = "config") + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + nc = True, + d = 100, + f = 'cap2/imix_fast_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResults instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + + self.check_CPU_benchmark(trex_res, 10) + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + # remove nbar config here + pass + +if __name__ == "__main__": + pass diff --git a/scripts/automation/regression/unit_tests/trex_ipv6_test.py b/scripts/automation/regression/unit_tests/trex_ipv6_test.py new file mode 100755 index 00000000..bffb4754 --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_ipv6_test.py @@ -0,0 +1,102 @@ +#!/router/bin/python +from trex_general_test import CTRexGeneral_Test +from tests_exceptions import * +import time +from nose.tools import assert_equal + +class CTRexIPv6_Test(CTRexGeneral_Test): + """This class defines the IPv6 testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + super(CTRexIPv6_Test, self).__init__(*args, **kwargs) + pass + + def setUp(self): + super(CTRexIPv6_Test, self).setUp() # launch super test class setUp process +# print " before sleep setup !!" +# time.sleep(100000); +# pass + + def test_ipv6_simple(self): + if self.is_virt_nics: + self.skip('--ipv6 flag does not work correctly in with virtual NICs') # TODO: fix + # test initializtion + if not self.is_loopback: + self.router.configure_basic_interfaces() + + self.router.config_pbr(mode = "config") + self.router.config_ipv6_pbr(mode = "config") + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + ipv6 = True, + d = 60, + f = 'avl/sfr_delay_10_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + + self.check_CPU_benchmark (trex_res, 10.0) + + assert True + + + def test_ipv6_negative (self): + if self.is_loopback: + self.skip('The test checks ipv6 drops by device and we are in loopback setup') + # test initializtion + self.router.configure_basic_interfaces() + + # NOT CONFIGURING IPv6 INTENTIONALLY TO GET DROPS! + self.router.config_pbr(mode = "config") + + # same params as test_ipv6_simple + mult = self.get_benchmark_param('multiplier', test_name = 'test_ipv6_simple') + core = self.get_benchmark_param('cores', test_name = 'test_ipv6_simple') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + ipv6 = True, + d = 60, + f = 'avl/sfr_delay_10_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + trex_tx_pckt = float(trex_res.get_last_value("trex-global.data.m_total_tx_pkts")) + trex_drops = int(trex_res.get_total_drops()) + + trex_drop_rate = trex_res.get_drop_rate() + + # make sure that at least 50% of the total transmitted packets failed + self.assert_gt((trex_drops/trex_tx_pckt), 0.5, 'packet drop ratio is not high enough') + + + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + # remove config here + pass + +if __name__ == "__main__": + pass diff --git a/scripts/automation/regression/unit_tests/trex_nat_test.py b/scripts/automation/regression/unit_tests/trex_nat_test.py new file mode 100755 index 00000000..452f7ecf --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_nat_test.py @@ -0,0 +1,164 @@ +#!/router/bin/python +from trex_general_test import CTRexGeneral_Test +from tests_exceptions import * +import time +from CPlatform import CStaticRouteConfig, CNatConfig +from nose.tools import assert_equal + + +class CTRexNoNat_Test(CTRexGeneral_Test):#(unittest.TestCase): + """This class defines the NAT testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + super(CTRexNoNat_Test, self).__init__(*args, **kwargs) + self.unsupported_modes = ['loopback'] # NAT requires device + pass + + def setUp(self): + super(CTRexNoNat_Test, self).setUp() # launch super test class setUp process + pass + + def check_nat_stats (self, nat_stats): + pass + + + def test_nat_learning(self): + # test initializtion + self.router.configure_basic_interfaces() + + stat_route_dict = self.get_benchmark_param('stat_route_dict') + stat_route_obj = CStaticRouteConfig(stat_route_dict) + self.router.config_static_routing(stat_route_obj, mode = "config") + + self.router.config_nat_verify() # shutdown duplicate interfaces + +# self.trex.set_yaml_file('cap2/http_simple.yaml') + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + +# trex_res = self.trex.run(multiplier = mult, cores = core, duration = 100, l = 1000, learn_verify = True) + ret = self.trex.start_trex( + c = core, + m = mult, + learn_verify = True, + d = 100, + f = 'cap2/http_simple.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + print ("\nLATEST DUMP:") + print trex_res.get_latest_dump() + + + expected_nat_opened = self.get_benchmark_param('nat_opened') + learning_stats = trex_res.get_last_value("trex-global.data", ".*nat.*") # extract all nat data + + if self.get_benchmark_param('allow_timeout_dev'): + nat_timeout_ratio = learning_stats['m_total_nat_time_out']/learning_stats['m_total_nat_open'] + if nat_timeout_ratio > 0.005: + self.fail('TRex nat_timeout ratio %f > 0.005 (0.5%) and not as expected to be less than 0.5%' %(nat_timeout_ratio)) + else: + self.check_results_eq (learning_stats, 'm_total_nat_time_out', 0.0) + self.check_results_eq (learning_stats, 'm_total_nat_no_fid', 0.0) + self.check_results_gt (learning_stats, 'm_total_nat_learn_error', 0.0) +# + self.check_results_gt (learning_stats, 'm_total_nat_open', expected_nat_opened) + + self.check_general_scenario_results(trex_res) + + # self.check_CPU_benchmark(trex_res, 10) + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + pass + + +class CTRexNat_Test(CTRexGeneral_Test):#(unittest.TestCase): + """This class defines the NAT testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + super(CTRexNat_Test, self).__init__(*args, **kwargs) + self.unsupported_modes = ['loopback'] # NAT requires device + pass + + def setUp(self): + super(CTRexNat_Test, self).setUp() # launch super test class setUp process + # config nat here + + + def check_nat_stats (self, nat_stats): + pass + + + def test_nat_simple(self): + # test initializtion + self.router.configure_basic_interfaces() + + + stat_route_dict = self.get_benchmark_param('stat_route_dict') + stat_route_obj = CStaticRouteConfig(stat_route_dict) + self.router.config_static_routing(stat_route_obj, mode = "config") + + nat_dict = self.get_benchmark_param('nat_dict') + nat_obj = CNatConfig(nat_dict) + self.router.config_nat(nat_obj) + +# self.trex.set_yaml_file('cap2/http_simple.yaml') + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + +# trex_res = self.trex.run(nc=False,multiplier = mult, cores = core, duration = 100, l = 1000, learn = True) + ret = self.trex.start_trex( + c = core, + m = mult, + learn = True, + d = 100, + f = 'cap2/http_simple.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + print ("\nLATEST DUMP:") + print trex_res.get_latest_dump() + + trex_nat_stats = trex_res.get_last_value("trex-global.data", ".*nat.*") # extract all nat data + if self.get_benchmark_param('allow_timeout_dev'): + nat_timeout_ratio = trex_nat_stats['m_total_nat_time_out']/trex_nat_stats['m_total_nat_open'] + if nat_timeout_ratio > 0.005: + self.fail('TRex nat_timeout ratio %f > 0.005 (0.5%) and not as expected to be less than 0.5%' %(nat_timeout_ratio)) + else: + self.check_results_eq (trex_nat_stats,'m_total_nat_time_out', 0.0) + self.check_results_eq (trex_nat_stats,'m_total_nat_no_fid', 0.0) + self.check_results_gt (trex_nat_stats,'m_total_nat_open', 6000) + + + self.check_general_scenario_results(trex_res) +## test_norm_cpu = 2*(trex_res.result['total-tx']/(core*trex_res.result['cpu_utilization'])) +# trex_tx_pckt = trex_res.get_last_value("trex-global.data.m_total_tx_bps") +# cpu_util = int(trex_res.get_last_value("trex-global.data.m_cpu_util")) +# test_norm_cpu = 2*(trex_tx_pckt/(core*cpu_util)) +# print "test_norm_cpu is: ", test_norm_cpu + + # self.check_CPU_benchmark(trex_res, 10) + + #if ( abs((test_norm_cpu/self.get_benchmark_param('cpu_to_core_ratio')) - 1) > 0.03): + # raiseraise AbnormalResultError('Normalized bandwidth to CPU utilization ratio exceeds 3%') + + nat_stats = self.router.get_nat_stats() + print nat_stats + + self.assert_gt(nat_stats['total_active_trans'], 5000, 'total active translations is not high enough') + self.assert_gt(nat_stats['dynamic_active_trans'], 5000, 'total dynamic active translations is not high enough') + self.assertEqual(nat_stats['static_active_trans'], 0, "NAT statistics nat_stats['static_active_trans'] should be zero") + self.assert_gt(nat_stats['num_of_hits'], 50000, 'total nat hits is not high enough') + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + self.router.clear_nat_translations() + + +if __name__ == "__main__": + pass diff --git a/scripts/automation/regression/unit_tests/trex_nbar_test.py b/scripts/automation/regression/unit_tests/trex_nbar_test.py new file mode 100755 index 00000000..e4f7eb37 --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_nbar_test.py @@ -0,0 +1,193 @@ +#!/router/bin/python +from trex_general_test import CTRexGeneral_Test +from tests_exceptions import * +from interfaces_e import IFType +from nose.tools import nottest +from misc_methods import print_r + +class CTRexNbar_Test(CTRexGeneral_Test): + """This class defines the NBAR testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + super(CTRexNbar_Test, self).__init__(*args, **kwargs) + self.unsupported_modes = ['loopback'] # obviously no NBar in loopback + pass + + def setUp(self): + super(CTRexNbar_Test, self).setUp() # launch super test class setUp process +# self.router.kill_nbar_flows() + self.router.clear_cft_counters() + self.router.clear_nbar_stats() + + def match_classification (self): + nbar_benchmark = self.get_benchmark_param("nbar_classification") + test_classification = self.router.get_nbar_stats() + print "TEST CLASSIFICATION:" + print test_classification + missmatchFlag = False + missmatchMsg = "NBAR classification contians a missmatch on the following protocols:" + fmt = '\n\t{0:15} | Expected: {1:>3.2f}%, Got: {2:>3.2f}%' + noise_level = 0.045 # percents + + for cl_intf in self.router.get_if_manager().get_if_list(if_type = IFType.Client): + client_intf = cl_intf.get_name() + + # removing noise classifications + for key, value in test_classification[client_intf]['percentage'].items(): + if value <= noise_level: + print 'Removing noise classification: %s' % key + del test_classification[client_intf]['percentage'][key] + + if len(test_classification[client_intf]['percentage']) != (len(nbar_benchmark) + 1): # adding 'total' key to nbar_benchmark + raise ClassificationMissmatchError ('The total size of classification result does not match the provided benchmark.') + + for protocol, bench in nbar_benchmark.iteritems(): + if protocol != 'total': + try: + bench = float(bench) + protocol = protocol.replace('_','-') + protocol_test_res = test_classification[client_intf]['percentage'][protocol] + deviation = 100 * abs(bench/protocol_test_res - 1) # percents + difference = abs(bench - protocol_test_res) + if (deviation > 10 and difference > noise_level): # allowing 10% deviation and 'noise_level'% difference + missmatchFlag = True + missmatchMsg += fmt.format(protocol, bench, protocol_test_res) + except KeyError as e: + missmatchFlag = True + print e + print "Changes missmatchFlag to True. ", "\n\tProtocol {0} isn't part of classification results on interface {intf}".format( protocol, intf = client_intf ) + missmatchMsg += "\n\tProtocol {0} isn't part of classification results on interface {intf}".format( protocol, intf = client_intf ) + except ZeroDivisionError as e: + print "ZeroDivisionError: %s" % protocol + pass + if missmatchFlag: + self.fail(missmatchMsg) + + + def test_nbar_simple(self): + # test initializtion + deviation_compare_value = 0.03 # default value of deviation - 3% + self.router.configure_basic_interfaces() + + self.router.config_pbr(mode = "config") + self.router.config_nbar_pd() + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + d = 100, + f = 'avl/sfr_delay_10_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + print ("\nLATEST DUMP:") + print trex_res.get_latest_dump() + + + self.check_general_scenario_results(trex_res, check_latency = False) + # test_norm_cpu = 2*(trex_res.result['total-tx']/(core*trex_res.result['cpu_utilization'])) + trex_tx_pckt = trex_res.get_last_value("trex-global.data.m_total_tx_pkts") + cpu_util = trex_res.get_last_value("trex-global.data.m_cpu_util") + cpu_util_hist = trex_res.get_value_list("trex-global.data.m_cpu_util") + print "cpu util is:", cpu_util + print cpu_util_hist + test_norm_cpu = 2 * trex_tx_pckt / (core * cpu_util) + print "test_norm_cpu is:", test_norm_cpu + + + if self.get_benchmark_param('cpu2core_custom_dev'): + # check this test by custom deviation + deviation_compare_value = self.get_benchmark_param('cpu2core_dev') + print "Comparing test with custom deviation value- {dev_val}%".format( dev_val = int(deviation_compare_value*100) ) + + # need to be fixed ! + #if ( abs((test_norm_cpu/self.get_benchmark_param('cpu_to_core_ratio')) - 1) > deviation_compare_value): + # raise AbnormalResultError('Normalized bandwidth to CPU utilization ratio exceeds benchmark boundaries') + + self.match_classification() + + assert True + + @nottest + def test_rx_check (self): + # test initializtion + self.router.configure_basic_interfaces() + + self.router.config_pbr(mode = "config") + self.router.config_nbar_pd() + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + rx_check = sample_rate, + d = 100, + f = 'cap2/sfr.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + print ("\nLATEST DUMP:") + print trex_res.get_latest_dump() + + self.check_general_scenario_results(trex_res) + + self.check_CPU_benchmark(trex_res, 10) + +# if trex_res.result['rx_check_tx']==trex_res.result['rx_check_rx']: # rx_check verification shoud pass +# assert trex_res.result['rx_check_verification'] == "OK" +# else: +# assert trex_res.result['rx_check_verification'] == "FAIL" + + # the name intentionally not matches nose default pattern, including the test should be specified explicitly + def NBarLong(self): + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = "config") + self.router.config_nbar_pd() + + mult = self.get_benchmark_param('multiplier') + core = self.get_benchmark_param('cores') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + d = 18000, # 5 hours + f = 'avl/sfr_delay_10_1g.yaml', + l = 1000) + + trex_res = self.trex.sample_to_run_finish() + + # trex_res is a CTRexResult instance- and contains the summary of the test results + # you may see all the results keys by simply calling here for 'print trex_res.result' + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res, check_latency = False) + + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + pass + +if __name__ == "__main__": + pass diff --git a/scripts/automation/regression/unit_tests/trex_rx_test.py b/scripts/automation/regression/unit_tests/trex_rx_test.py new file mode 100755 index 00000000..a37615c4 --- /dev/null +++ b/scripts/automation/regression/unit_tests/trex_rx_test.py @@ -0,0 +1,285 @@ +#!/router/bin/python +from trex_general_test import CTRexGeneral_Test +from CPlatform import CStaticRouteConfig, CNatConfig +from tests_exceptions import * +#import sys +import time +import copy +from nose.tools import nottest +import traceback + +class CTRexRx_Test(CTRexGeneral_Test): + """This class defines the rx testcase of the T-Rex traffic generator""" + def __init__(self, *args, **kwargs): + CTRexGeneral_Test.__init__(self, *args, **kwargs) + self.unsupported_modes = ['virt_nics'] # TODO: fix + pass + + def setUp(self): + CTRexGeneral_Test.setUp(self) + pass + + + def check_rx_errors(self, trex_res): + try: + # counters to check + + latency_counters_display = {'m_unsup_prot': 0, 'm_no_magic': 0, 'm_no_id': 0, 'm_seq_error': 0, 'm_length_error': 0, 'm_no_ipv4_option': 0, 'm_tx_pkt_err': 0} + rx_counters = {'m_err_drop': 0, 'm_err_aged': 0, 'm_err_no_magic': 0, 'm_err_wrong_pkt_id': 0, 'm_err_fif_seen_twice': 0, 'm_err_open_with_no_fif_pkt': 0, 'm_err_oo_dup': 0, 'm_err_oo_early': 0, 'm_err_oo_late': 0, 'm_err_flow_length_changed': 0} + + # get relevant TRex results + + try: + ports_names = trex_res.get_last_value('trex-latecny-v2.data', 'port\-\d+') + if not ports_names: + raise AbnormalResultError('Could not find ports info in TRex results, path: trex-latecny-v2.data.port-*') + for port_name in ports_names: + path = 'trex-latecny-v2.data.%s.stats' % port_name + port_result = trex_res.get_last_value(path) + if not port_result: + raise AbnormalResultError('Could not find port stats in TRex results, path: %s' % path) + for key in latency_counters_display: + latency_counters_display[key] += port_result[key] + + # using -k flag in TRex produces 1 error per port in latency counter m_seq_error, allow it until issue resolved. For comparing use dict with reduces m_seq_error number. + latency_counters_compare = copy.deepcopy(latency_counters_display) + latency_counters_compare['m_seq_error'] = max(0, latency_counters_compare['m_seq_error'] - len(ports_names)) + + path = 'rx-check.data.stats' + rx_check_results = trex_res.get_last_value(path) + if not rx_check_results: + raise AbnormalResultError('No TRex results by path: %s' % path) + for key in rx_counters: + rx_counters[key] = rx_check_results[key] + + path = 'rx-check.data.stats.m_total_rx' + total_rx = trex_res.get_last_value(path) + if not total_rx: + raise AbnormalResultError('No TRex results by path: %s' % path) + + + print 'Total packets checked: %s' % total_rx + print 'Latency counters: %s' % latency_counters_display + print 'rx_check counters: %s' % rx_counters + + except KeyError as e: + self.fail('Expected key in TRex result was not found.\n%s' % traceback.print_exc()) + + # the check. in loopback expect 0 problems, at others allow errors <error_tolerance>% of total_rx + + total_errors = sum(rx_counters.values()) + sum(latency_counters_compare.values()) + error_tolerance = self.get_benchmark_param('error_tolerance') + if not error_tolerance: + error_tolerance = 0 + error_percentage = float(total_errors) * 100 / total_rx + + if total_errors > 0: + if self.is_loopback or error_percentage > error_tolerance: + self.fail('Too much errors in rx_check. (~%s%% of traffic)' % error_percentage) + else: + print 'There are errors in rx_check (%f%%), not exceeding allowed limit (%s%%)' % (error_percentage, error_tolerance) + else: + print 'No errors in rx_check.' + except Exception as e: + print traceback.print_exc() + self.fail('Errors in rx_check: %s' % e) + + def test_rx_check_sfr(self): + if not self.is_loopback: + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = 'config') + + core = self.get_benchmark_param('cores') + mult = self.get_benchmark_param('multiplier') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + rx_check = sample_rate, + d = 100, + f = 'avl/sfr_delay_10_1g_no_bundeling.yaml', + l = 1000, + k = 10, + learn_verify = True, + l_pkt_mode = 2) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + #print ("\nLATEST DUMP:") + #print trex_res.get_latest_dump() + + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + + + def test_rx_check_http(self): + if not self.is_loopback: + # TODO: skip as test_rx_check_http_negative will cover it + #self.skip('This test is covered by test_rx_check_http_negative') + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = "config") + + core = self.get_benchmark_param('cores') + mult = self.get_benchmark_param('multiplier') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + rx_check = sample_rate, + d = 100, + f = 'cap2/http_simple.yaml', + l = 1000, + k = 10, + learn_verify = True, + l_pkt_mode = 2) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + + + def test_rx_check_sfr_ipv6(self): + if not self.is_loopback: + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = 'config') + self.router.config_ipv6_pbr(mode = "config") + + core = self.get_benchmark_param('cores') + mult = self.get_benchmark_param('multiplier') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + rx_check = sample_rate, + d = 100, + f = 'avl/sfr_delay_10_1g_no_bundeling.yaml', + l = 1000, + k = 10, + ipv6 = True) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + #print ("\nLATEST DUMP:") + #print trex_res.get_latest_dump() + + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + + + def test_rx_check_http_ipv6(self): + if not self.is_loopback: + self.router.configure_basic_interfaces() + self.router.config_pbr(mode = "config") + self.router.config_ipv6_pbr(mode = "config") + + core = self.get_benchmark_param('cores') + mult = self.get_benchmark_param('multiplier') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + p = True, + nc = True, + rx_check = sample_rate, + d = 100, + f = 'cap2/http_simple.yaml', + l = 1000, + k = 10, + ipv6 = True) + + trex_res = self.trex.sample_to_run_finish() + + print ("\nLATEST RESULT OBJECT:") + print trex_res + + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + + @nottest + def test_rx_check_http_negative(self): + if self.is_loopback: + self.skip('This test uses NAT, not relevant for loopback') + + self.router.configure_basic_interfaces() + stat_route_dict = self.get_benchmark_param('stat_route_dict') + stat_route_obj = CStaticRouteConfig(stat_route_dict) + self.router.config_static_routing(stat_route_obj, mode = "config") + + core = self.get_benchmark_param('cores') + mult = self.get_benchmark_param('multiplier') + sample_rate = self.get_benchmark_param('rx_sample_rate') + + ret = self.trex.start_trex( + c = core, + m = mult, + #p = True, + #nc = True, + rx_check = sample_rate, + d = 80, + f = 'cap2/http_simple.yaml', + l = 1000, + k = 10, + learn_verify = True, + l_pkt_mode = 2) + + print 'Run for 2 minutes, expect no errors' + trex_res = self.trex.sample_x_seconds(60) + print ("\nLATEST RESULT OBJECT:") + print trex_res + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + + try: + # TODO: add nat/zbf config for router + nat_dict = self.get_benchmark_param('nat_dict') + nat_obj = CNatConfig(nat_dict) + self.router.config_nat(nat_obj) + self.router.config_nat_verify() + self.router.config_zbf() + + print 'Run until finish, expect errors' + trex_res = self.trex.sample_to_run_finish() + + self.router.config_no_zbf() + self.router.clear_nat_translations() + print ("\nLATEST RESULT OBJECT:") + print trex_res + nat_stats = self.router.get_nat_stats() + print nat_stats + self.check_general_scenario_results(trex_res) + self.check_CPU_benchmark(trex_res, 10) + self.check_rx_errors(trex_res) + self.fail('Expected errors here, got none.') + except Exception as e: + print 'Got errors as expected: %s' % e + pass + + def tearDown(self): + CTRexGeneral_Test.tearDown(self) + pass + +if __name__ == "__main__": + pass |