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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
from collections import namedtuple, OrderedDict
from .utils.text_opts import *
from .trex_stl_exceptions import *
import types
RpcCmdData = namedtuple('RpcCmdData', ['method', 'params', 'api_class'])
TupleRC = namedtuple('RCT', ['rc', 'data', 'is_warn'])
class RpcResponseStatus(namedtuple('RpcResponseStatus', ['success', 'id', 'msg'])):
__slots__ = ()
def __str__(self):
return "{id:^3} - {msg} ({stat})".format(id=self.id,
msg=self.msg,
stat="success" if self.success else "fail")
# simple class to represent complex return value
class RC():
def __init__ (self, rc = None, data = None, is_warn = False):
self.rc_list = []
if (rc != None):
self.rc_list.append(TupleRC(rc, data, is_warn))
def __nonzero__ (self):
return self.good()
def __bool__ (self):
return self.good()
def add (self, rc):
self.rc_list += rc.rc_list
def good (self):
return all([x.rc for x in self.rc_list])
def bad (self):
return not self.good()
def warn (self):
return any([x.is_warn for x in self.rc_list])
def data (self):
d = [x.data if x.rc else "" for x in self.rc_list]
return (d if len(d) != 1 else d[0])
def err (self):
e = [x.data if not x.rc else "" for x in self.rc_list]
return (e if len(e) != 1 else e[0])
def __str__ (self):
if self.good():
s = ""
for x in self.rc_list:
if x.data:
s += format_text("\n{0}".format(x.data), 'bold')
return s
else:
show_count = 10
err_list = []
err_count = 0
for x in filter(len, listify(self.err())):
err_count += 1
if len(err_list) < show_count:
err_list.append(format_text(x, 'bold'))
s = ''
if err_count > show_count:
s += format_text('Occurred %s errors, showing first %s:\n' % (err_count, show_count), 'bold')
s += '\n'.join(err_list)
return s
def __iter__(self):
return self.rc_list.__iter__()
def prn_func (self, msg, newline = True):
if newline:
print(msg)
else:
print(msg),
def annotate (self, log_func = None, desc = None, show_status = True):
if not log_func:
log_func = self.prn_func
if desc:
log_func(format_text('\n{:<60}'.format(desc), 'bold'), newline = False)
else:
log_func("")
if self.bad():
# print all the errors
print("")
for x in self.rc_list:
if not x.rc:
log_func(format_text("\n{0}".format(x.data), 'bold'))
print("")
if show_status:
log_func(format_text("[FAILED]\n", 'red', 'bold'))
else:
if show_status:
log_func(format_text("[SUCCESS]\n", 'green', 'bold'))
def RC_OK(data = ""):
return RC(True, data)
def RC_ERR (err = ""):
return RC(False, err)
def RC_WARN (warn):
return RC(True, warn, is_warn = True)
try:
long
long_exists = True
except:
long_exists = False
def is_integer(arg):
if type(arg) is int:
return True
if long_exists and type(arg) is long:
return True
return False
# validate type of arg
# example1: validate_type('somearg', somearg, [int, long])
# example2: validate_type('another_arg', another_arg, str)
def validate_type(arg_name, arg, valid_types):
if long_exists:
if valid_types is int:
valid_types = (int, long)
elif type(valid_types) is list and int in valid_types and long not in valid_types:
valid_types.append(long)
if type(valid_types) is list:
valid_types = tuple(valid_types)
if (type(valid_types) is type or # single type, not array of types
type(valid_types) is tuple or # several valid types as tuple
type(valid_types) is types.ClassType): # old style class
if isinstance(arg, valid_types):
return
raise STLTypeError(arg_name, type(arg), valid_types)
else:
raise STLError('validate_type: valid_types should be type or list or tuple of types')
def validate_choice (arg_name, arg, choices):
if arg is not None and not arg in choices:
raise STLError("validate_choice: argument '{0}' can only be one of '{1}'".format(arg_name, choices))
# throws STLError if not exactly one argument is present
def verify_exclusive_arg (args_list):
if not (len(list(filter(lambda x: x is not None, args_list))) == 1):
raise STLError('exactly one parameter from {0} should be provided'.format(args_list))
def listify (x):
if isinstance(x, list):
return x
else:
return [x]
# shows as 'N/A', but does not let any compares for user to not mistake in automation
class StatNotAvailable(str):
def __new__(cls, value, *args, **kwargs):
cls.stat_name = value
return super(StatNotAvailable, cls).__new__(cls, 'N/A')
def __cmp__(self, *args, **kwargs):
raise Exception("Stat '%s' not available at this setup" % self.stat_name)
class LRU_cache(OrderedDict):
def __init__(self, maxlen = 20, *args, **kwargs):
OrderedDict.__init__(self, *args, **kwargs)
self.maxlen = maxlen
def __setitem__(self, *args, **kwargs):
OrderedDict.__setitem__(self, *args, **kwargs)
if len(self) > self.maxlen:
self.popitem(last = False)
|
