|
8
|
1 # This module contains mock functions for some of the functionality in gobject.
|
|
|
2 # You can use this to create unit tests on code using gobject timers without having to wait for those timer.
|
|
|
3 # Use the patch functions to replace the original gobject functions. The timer_manager object defined here
|
|
|
4 # allows you to set a virtual time stamp, which will invoke all timers that would normally run in the
|
|
|
5 # specified interval.
|
|
|
6
|
|
|
7 from datetime import datetime as dt
|
|
|
8 import time
|
|
|
9
|
|
|
10 class MockTimer(object):
|
|
|
11 def __init__(self, start, timeout, callback, *args, **kwargs):
|
|
|
12 self._timeout = timeout
|
|
|
13 self._next = start + timeout
|
|
|
14 self._callback = callback
|
|
|
15 self._args = args
|
|
|
16 self._kwargs = kwargs
|
|
|
17
|
|
|
18 def run(self):
|
|
|
19 self._next += self._timeout
|
|
|
20 return self._callback(*self._args, **self._kwargs)
|
|
|
21
|
|
|
22 @property
|
|
|
23 def next(self):
|
|
|
24 return self._next
|
|
|
25
|
|
|
26
|
|
|
27 class MockTimerManager(object):
|
|
|
28 def __init__(self, start_time=None):
|
|
|
29 self._resources = []
|
|
|
30 self._time = 0
|
|
|
31 self._id = 0
|
|
|
32 self._timestamp = start_time or time.time()
|
|
|
33
|
|
|
34 def add_timer(self, timeout, callback, *args, **kwargs):
|
|
|
35 return self._add_resource(MockTimer(self._time, timeout, callback, *args, **kwargs))
|
|
|
36
|
|
|
37 def add_idle(self, callback, *args, **kwargs):
|
|
|
38 return self.add_timer(self._time, callback, *args, **kwargs)
|
|
|
39
|
|
|
40 def remove_resouce(self, id):
|
|
|
41 for rid, rr in self._resources:
|
|
|
42 if rid == id:
|
|
|
43 self._resources.remove((rid, rr))
|
|
|
44 return
|
|
|
45 raise Exception('Resource not found: {}'.format(id))
|
|
|
46
|
|
|
47 def _add_resource(self, resource):
|
|
|
48 self._id += 1
|
|
|
49 self._resources.append((self._id, resource))
|
|
|
50 return self._id
|
|
|
51
|
|
|
52 def _terminate(self):
|
|
|
53 raise StopIteration()
|
|
|
54
|
|
|
55 @property
|
|
|
56 def time(self):
|
|
|
57 return self._time
|
|
|
58
|
|
|
59 @property
|
|
|
60 def datetime(self):
|
|
|
61 return dt.fromtimestamp(self._timestamp + self._time / 1000.0)
|
|
|
62
|
|
|
63 def run(self, interval=None):
|
|
|
64 '''
|
|
|
65 Simulate the given interval. Starting from the current (mock) time until time + interval, all timers
|
|
|
66 will be triggered. The timers will be triggered in chronological order. Timer removal (calling
|
|
|
67 source_remove or a False/None return value) and addition within the callback function is supported.
|
|
|
68 If interval is None or not supplied, the function will run until there are no timers left.
|
|
|
69 '''
|
|
|
70 if interval != None:
|
|
|
71 self.add_timer(interval, self._terminate)
|
|
|
72 try:
|
|
|
73 while True:
|
|
|
74 next_timer = None
|
|
|
75 next_id = None
|
|
|
76 for id,t in self._resources:
|
|
|
77 if next_timer == None or t.next < next_timer.next:
|
|
|
78 next_timer = t
|
|
|
79 next_id = id
|
|
|
80 if next_timer == None:
|
|
|
81 return
|
|
|
82 self._time = next_timer.next
|
|
|
83 if not next_timer.run():
|
|
|
84 self._resources.remove((next_id, next_timer))
|
|
|
85 except StopIteration:
|
|
|
86 self._resources.remove((next_id, next_timer))
|
|
|
87 pass
|
|
|
88
|
|
|
89 def reset(self):
|
|
|
90 self._resources = []
|
|
|
91 self._time = 0
|
|
|
92
|
|
|
93
|
|
|
94 timer_manager = MockTimerManager()
|
|
|
95
|
|
|
96
|
|
|
97 def idle_add(callback, *args, **kwargs):
|
|
|
98 return timer_manager.add_idle(callback, *args, **kwargs)
|
|
|
99
|
|
|
100
|
|
|
101 def timeout_add(timeout, callback, *args, **kwargs):
|
|
|
102 return timer_manager.add_timer(timeout, callback, *args, **kwargs)
|
|
|
103
|
|
|
104
|
|
|
105 def timeout_add_seconds(timeout, callback, *args, **kwargs):
|
|
|
106 return timeout_add(timeout * 1000, callback, *args, **kwargs)
|
|
|
107
|
|
|
108
|
|
|
109 class datetime(object):
|
|
|
110 @staticmethod
|
|
|
111 def now():
|
|
|
112 return timer_manager.datetime
|
|
|
113
|
|
|
114 @staticmethod
|
|
|
115 def strptime(*args, **kwargs):
|
|
|
116 return dt.strptime(*args, **kwargs)
|
|
|
117
|
|
|
118
|
|
|
119 def source_remove(id):
|
|
|
120 timer_manager.remove_resouce(id)
|
|
|
121
|
|
|
122
|
|
|
123 def test_function(m, name):
|
|
|
124 print(m.time, m.datetime, name)
|
|
|
125 return True
|
|
|
126
|
|
|
127
|
|
|
128 def patch_gobject(dest):
|
|
|
129 '''
|
|
|
130 Use this function to replace the original gobject/GLib functions with the
|
|
|
131 mocked versions in this file. Suppose your source files being tested uses
|
|
|
132 'from gi.repository import GLib' and the unit test uses 'import tested' you
|
|
|
133 should call path(tested.GLib).
|
|
|
134 '''
|
|
|
135 dest.timeout_add = timeout_add
|
|
|
136 dest.timeout_add_seconds = timeout_add_seconds
|
|
|
137 dest.idle_add = idle_add
|
|
|
138 dest.source_remove = source_remove
|
|
|
139
|
|
|
140
|
|
|
141 def patch_datetime(dest):
|
|
|
142 dest.datetime = datetime
|
|
|
143
|
|
|
144
|
|
|
145 if __name__ == '__main__':
|
|
|
146 m = MockTimerManager()
|
|
|
147 id1 = m.add_timer(100, test_function, m, 'F1')
|
|
|
148 id2 = m.add_timer(30, test_function, m, 'F2')
|
|
|
149 m.run(5000)
|
|
|
150 m.remove_resouce(id1)
|
|
|
151 m.run(2000)
|
|
|
152 m.remove_resouce(id2)
|
|
|
153 m.run(2000)
|
