1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
import sys
import os
import time
import cStringIO
api_path = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(api_path, '../automation/trex_control_plane/client/'))
from trex_stateless_client import CTRexStatelessClient, LoggerApi
from common import trex_stats
# a basic test object
class BasicTestAPI(object):
# exception class
class Failure(Exception):
def __init__ (self, rc):
self.msg = str(rc)
def __str__ (self):
s = "\n\n******\n"
s += "Test has failed.\n\n"
s += "Error reported:\n\n {0}\n".format(self.msg)
return s
# logger for test
class Logger(LoggerApi):
def __init__ (self):
LoggerApi.__init__(self)
self.buffer = cStringIO.StringIO()
def write (self, msg, newline = True):
line = str(msg) + ("\n" if newline else "")
self.buffer.write(line)
#print line
def flush (self):
pass
# main object
def __init__ (self, server = "localhost", sync_port = 4501, async_port = 4500):
self.logger = BasicTestAPI.Logger()
self.client = CTRexStatelessClient(logger = self.logger, sync_port = sync_port, async_port = async_port)
def connect (self):
rc = self.client.connect(mode = "RWF")
self.__verify_rc(rc)
def disconnect (self):
self.client.disconnect()
def __verify_rc (self, rc):
if rc.bad():
raise self.Failure(rc)
def inject_profile (self, filename, rate = "1", duration = None):
cmd = "-f {0} -m {1}".format(filename, rate)
if duration:
cmd += " -d {0}".format(duration)
rc = self.client.cmd_start_line(cmd)
self.__verify_rc(rc)
def wait_while_traffic_on (self, timeout = None):
while self.client.get_active_ports():
time.sleep(0.1)
def get_stats (self):
total_stats = trex_stats.CPortStats(None)
for port in self.client.ports.values():
total_stats += port.port_stats
return total_stats
|
