3 # Thierry Parmentelat <thierry.parmentelat@inria.fr>
4 # Copyright (C) 2015 INRIA
7 from datetime import datetime, timedelta
11 ### more generic code for waiting for any number of things
12 ### within a given timeframe i.e. given some timeout/silent/period
13 ### takes in argument a list of tasks that are instances
14 ### of a CompleterTask subclass
16 def __init__ (self, tasks, verbose=True, message=None):
18 self.verbose = verbose
19 self.message = "({})".format(message) if message else ""
20 def run (self, timeout_timedelta, silent_timedelta, period):
21 begin = datetime.now()
22 timeout = begin+timeout_timedelta
23 timeout_seconds = timeout_timedelta.total_seconds()
24 timeout_minutes = timeout_seconds/60
25 graceout = datetime.now()+silent_timedelta
26 silent_seconds = silent_timedelta.total_seconds()
27 silent_minutes = silent_seconds/60
28 period_seconds = int(period.total_seconds())
30 if timeout_seconds >= 120:
31 utils.header("Completer [{} tasks]: max timeout is {} minutes, "
32 "silent for {} minutes (period is {} s)"\
33 .format(len(self.tasks), timeout_minutes,
34 silent_minutes, period_seconds))
36 utils.header("Completer [{} tasks]: max timeout is {} seconds, "
37 "silent for {} seconds (period is {} s)"\
38 .format(len(self.tasks), timeout_seconds,
39 silent_seconds, period_seconds))
44 success = task.run (silent=datetime.now() <= graceout)
51 duration = datetime.now() - begin
52 print("total completer {} {}s".format(self.message,
53 int(duration.total_seconds())))
55 if datetime.now() > timeout:
57 task.failure_epilogue()
60 print('{}s..'.format(period_seconds), end=' ')
61 time.sleep(period_seconds)
66 #################### CompleterTask
67 ### . run(silent) (return True or False)
68 ### silent is an input boolean indicating if we're within the silent period
69 ### . failure() (print a message)
71 ########## expectations (+ first arg self)
72 # failure() (to describe which went wrong once it's over)
80 def run (self, silent):
81 result = self.actual_run()
83 print('+' if result else '.', end=' ')
86 print(self.message(), "->", "OK" if result else "KO")
90 return "you-need-to-redefine-message"
92 def failure_epilogue (self):
93 print("you-need-to-redefine-failure_epilogue")
96 class TaskTest (CompleterTask):
98 def __init__ (self,max):
100 self.counter = TaskTest.counter
101 TaskTest.counter += 1
102 self.delay = random.random()*max
103 self.fire = datetime.now() + timedelta(seconds=self.delay)
104 def actual_run(self):
105 return datetime.now() >= self.fire
107 return "Task {} - delay was {}s".format(self.counter, self.delay)
109 def failure_epilogue (self):
110 print("BOTTOM LINE: FAILURE with task ({})".format(self.counter))
114 if len(sys.argv) != 6:
115 print("Usage: <command> number_tasks max_random timeout_s silent_s period_s")
117 [number, max, timeout, silent, period] = [ int(x) for x in sys.argv[1:]]
118 tasks = [ TaskTest(max) for i in range(number)]
119 success = Completer(tasks,verbose=True).run(timedelta(seconds=timeout),
120 timedelta(seconds=silent),
121 timedelta(seconds=period))
122 print("OVERALL",success)
124 if __name__ == '__main__':