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
|
import os
import re
import sys
import time
import copy
import thread
import socket
import threading
import logging
import inspect
import argparse
import telnetlib
import redis
import random
import redis
import json
import glob
import commands
from collections import defaultdict
from argparse import RawTextHelpFormatter
from string import Template
PWD = os.path.dirname(os.path.realpath(__file__))
WORKDIR = os.path.join(PWD, '../')
def getenv(key, default):
if key in os.environ:
return os.environ[key]
return default
logfile = getenv('T_LOGFILE', 'log/t.log')
if logfile == '-':
logging.basicConfig(level=logging.DEBUG,
format="%(asctime)-15s [%(threadName)s] [%(levelname)s] %(message)s")
else:
logging.basicConfig(filename=logfile, level=logging.DEBUG,
format="%(asctime)-15s [%(threadName)s] [%(levelname)s] %(message)s")
logging.info("test running")
def strstr(s1, s2):
return s1.find(s2) != -1
def lets_sleep(SLEEP_TIME = 0.1):
time.sleep(SLEEP_TIME)
def TT(template, args): #todo: modify all
return Template(template).substitute(args)
def TTCMD(template, args): #todo: modify all
'''
Template for cmd (we will replace all spaces)
'''
ret = TT(template, args)
return re.sub(' +', ' ', ret)
def nothrow(ExceptionToCheck=Exception, logger=None):
def deco_retry(f):
def f_retry(*args, **kwargs):
try:
return f(*args, **kwargs)
except ExceptionToCheck, e:
if logger:
logger.info(e)
else:
print str(e)
return f_retry # true decorator
return deco_retry
@nothrow(Exception)
def test_nothrow():
raise Exception('exception: xx')
def json_encode(j):
return json.dumps(j, indent=4, cls=MyEncoder)
def json_decode(j):
return json.loads(j)
#commands dose not work on windows..
def system(cmd, log_fun=logging.info):
if log_fun: log_fun(cmd)
r = commands.getoutput(cmd)
return r
def shorten(s, l=80):
if len(s)<=l:
return s
return s[:l-3] + '...'
def assert_true(a):
assert a, 'assert fail: except true, got %s' % a
def assert_equal(a, b):
assert a == b, 'assert fail: %s vs %s' % (shorten(str(a)), shorten(str(b)))
def assert_raises(exception_cls, callable, *args, **kwargs):
try:
callable(*args, **kwargs)
except exception_cls as e:
return e
except Exception as e:
assert False, 'assert_raises %s but raised: %s' % (exception_cls, e)
assert False, 'assert_raises %s but nothing raise' % (exception_cls)
def assert_fail(err_response, callable, *args, **kwargs):
try:
callable(*args, **kwargs)
except Exception as e:
assert re.search(err_response, str(e)), \
'assert "%s" but got "%s"' % (err_response, e)
return
assert False, 'assert_fail %s but nothing raise' % (err_response)
if __name__ == "__main__":
test_nothrow()
|
