• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#encoding=utf-8
2'''
3======================================================================================
4版权 (C) 2015-2020, Huawei Technologies Co., HUTAF xDevice
5========================================================================================
6@FileName:    time_info.py
7@Function:    时间格式化方法,以及timeout注释
8@Author:
9@Date:
10======================================================================================
11'''
12
13import datetime
14import sys
15import threading
16import time
17from util.log_info import logger
18thead = []
19
20
21def get_timestamp():
22    return time.time()
23
24def get_time_str_info(time_stamp):
25    st = time.localtime(time_stamp)
26    result = time.strftime("%Y-%m-%d %H:%M:%S", st)
27    return result
28
29def time_info_to_timestamp(timeinfo):
30    st = time.strptime(timeinfo, "%Y-%m-%d %H:%M:%S")
31    return time.mktime(st)
32
33def get_now_time_str_info():
34    now = datetime.datetime.now()
35    time_str = now.strftime("%Y-%m-%d %H:%M:%S")
36    microsecond = now.microsecond
37    result = "%s,%s" % (time_str, microsecond)
38    return result
39
40def get_now_time_info():
41    now = datetime.datetime.now()
42    time_str = now.strftime("%Y-%m-%d %H:%M:%S")
43    return time_str
44
45class KThread(threading.Thread):
46    def __init__(self, *args, **kwargs):
47        threading.Thread.__init__(self, *args, **kwargs)
48        self.killed = False
49
50    def start(self):
51        self.__run_backup = self.run
52        self.run = self.__run      # Force the Thread to install our trace.
53        threading.Thread.start(self)
54
55    def __run(self):
56        sys.settrace(self.global_trace)
57        self.__run_backup()
58        self.run = self.__run_backup
59
60    def global_trace(self, frame, why, arg):
61        if why == "call":
62            return self.local_trace
63        else:
64            return None
65
66    def local_trace(self, frame, why, arg):
67        if self.killed:
68            if why == "line":
69                raise SystemExit()
70        return self.local_trace
71
72    def kill(self):
73        self.killed = True
74
75class Timeout(Exception):
76    def __init__(self, arg):
77        for i in thead:
78            if i is not None:
79                i.kill()
80                i = None
81        super(Timeout, self).__init__(arg)
82
83def timeout(seconds = 300):
84    def timeout_decorator(func):
85        def _new_func(oldfunc, result, oldfunc_args, oldfunc_kwargs):
86            result.append(oldfunc(*oldfunc_args, **oldfunc_kwargs))
87
88        def _(*args, **kwargs):
89            result = []
90            new_kwargs = { # create new args for _new_func, because we want to get the func return val to result list
91                "oldfunc": func,
92                "result": result,
93                "oldfunc_args": args,
94                "oldfunc_kwargs": kwargs
95            }
96            thd = KThread(target=_new_func, args=(), kwargs=new_kwargs)
97            thd.setDaemon(True)
98            thd.start()
99            thd.join(seconds)
100            alive = thd.isAlive()
101            thd.kill()
102            if alive:
103                raise Timeout(u"function run too long, timeout %d seconds." % seconds)
104            else:
105                if len(result) == 0:
106                    logger.info(u"function run too long, timeout %d seconds." % seconds)
107                    raise Exception("TESTCASE FAILED")
108                else:
109                    return result[0]
110        _.__name__ = func.__name__
111        _.__doc__ = func.__doc__
112        return _
113    return timeout_decorator