stl_general_test.py revision 7ca1b61c
1import os, sys
2import unittest
3from trex import CTRexScenario
4from stateful_tests.trex_general_test import CTRexGeneral_Test
5from trex_stl_lib.api import *
6import time
7from nose.tools import nottest
8
9class CStlGeneral_Test(CTRexGeneral_Test):
10    """This class defines the general stateless testcase of the TRex traffic generator"""
11
12    def setUp(self):
13        self.stl_trex = CTRexScenario.stl_trex if CTRexScenario.stl_trex else 'mock'
14        CTRexGeneral_Test.setUp(self)
15        # check basic requirements, should be verified at test_connectivity, here only skip test
16        if CTRexScenario.stl_init_error:
17            self.skip(CTRexScenario.stl_init_error)
18
19    def connect(self, tries = 10):
20        # need delay and check only because TRex process might be still starting
21        sys.stdout.write('Connecting')
22        for i in range(tries):
23            try:
24                sys.stdout.write('.')
25                sys.stdout.flush()
26                self.stl_trex.connect()
27                print('')
28                return True
29            except:
30                time.sleep(0.5)
31        print('')
32        return False
33
34    def map_ports(self, tries = 10):
35        sys.stdout.write('Mapping ports')
36        for i in range(tries):
37            sys.stdout.write('.')
38            sys.stdout.flush()
39            try:
40                CTRexScenario.stl_ports_map = stl_map_ports(self.stl_trex)
41                if self.verify_bidirectional(CTRexScenario.stl_ports_map):
42                    print('')
43                    return True
44            except Exception as e:
45                print('\nException during mapping: %s' % e)
46                return False
47            time.sleep(0.5)
48        print('')
49        return False
50
51    # verify all the ports are bidirectional
52    @staticmethod
53    def verify_bidirectional(mapping_dict):
54        if len(mapping_dict['unknown']):
55            return False
56        if len(mapping_dict['bi']) * 2 == len(mapping_dict['map']):
57            return True
58        return False
59
60    @staticmethod
61    def get_port_count():
62        return CTRexScenario.stl_trex.get_port_count()
63
64    @staticmethod
65    def is_connected():
66        return CTRexScenario.stl_trex.is_connected()
67
68class STLBasic_Test(CStlGeneral_Test):
69    # will run it first explicitly, check connectivity and configure routing
70    @nottest
71    def test_connectivity(self):
72        if not self.is_loopback:
73            try:
74                sys.stdout.flush()
75                sys.stdout.write('Configuring DUT... ')
76                start_time = time.time()
77                if CTRexScenario.router_cfg['forceCleanConfig']:
78                    CTRexScenario.router.load_clean_config()
79                CTRexScenario.router.configure_basic_interfaces()
80                CTRexScenario.router.config_pbr(mode = "config")
81                CTRexScenario.router.config_ipv6_pbr(mode = "config")
82                sys.stdout.write('done. (%ss)\n' % int(time.time() - start_time))
83            except Exception as e:
84                print('')
85                CTRexScenario.stl_init_error = 'Could not configure device, err: %s' % e
86                self.fail(CTRexScenario.stl_init_error)
87                return
88
89        try:
90            sys.stdout.write('Starting TRex... ')
91            start_time = time.time()
92            cores = self.configuration.trex.get('trex_cores', 1)
93            if self.is_virt_nics and cores > 1:
94                raise Exception('Number of cores should be 1 with virtual NICs')
95            if not CTRexScenario.no_daemon:
96                self.trex.start_stateless(c = cores)
97            self.stl_trex = STLClient(username = 'TRexRegression',
98                                      server = self.configuration.trex['trex_name'],
99                                      verbose_level = CTRexScenario.json_verbose)
100            CTRexScenario.stl_trex = self.stl_trex
101            sys.stdout.write('done. (%ss)\n' % int(time.time() - start_time))
102        except Exception as e:
103            print('')
104            CTRexScenario.stl_init_error = 'Could not start stateless TRex, err: %s' % e
105            self.fail(CTRexScenario.stl_init_error)
106            return
107
108        if not self.connect():
109            CTRexScenario.stl_init_error = 'Client could not connect'
110            self.fail(CTRexScenario.stl_init_error)
111            return
112        print('Connected')
113        if not self.map_ports():
114            CTRexScenario.stl_init_error = 'Client could not map ports'
115            self.fail(CTRexScenario.stl_init_error)
116            return
117        print('Got ports mapping: %s' % CTRexScenario.stl_ports_map)
118