Mercurial > ~darius > hgwebdir.cgi > epro

comparison velib_python/test/mock_gobject.py @ 8:9c0435a617db

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Import velib_python
author Daniel O'Connor <darius@dons.net.au>
date Sun, 05 Dec 2021 14:35:36 +1030
parents
children
comparison
equal deleted inserted replaced
5:982eeffe9d95 8:9c0435a617db
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)