tests/test_source.py

   1 # -*- Mode: Python -*-
   2
   3 import unittest
   4 import warnings
   5
   6 from gi.repository import GLib, GObject
   7 from gi import PyGIDeprecationWarning
   8
   9
  10 class Idle(GLib.Idle):
  11     def __init__(self, loop):
  12         GLib.Idle.__init__(self)
  13         self.count = 0
  14         self.set_callback(self.callback, loop)
  15
  16     def callback(self, loop):
  17         self.count += 1
  18         return True
  19
  20
  21 class MySource(GLib.Source):
  22     def __init__(self):
  23         GLib.Source.__init__(self)
  24
  25     def prepare(self):
  26         return True, 0
  27
  28     def check(self):
  29         return True
  30
  31     def dispatch(self, callback, args):
  32         return callback(*args)
  33
  34
  35 class TestSource(unittest.TestCase):
  36     def timeout_callback(self, loop):
  37         loop.quit()
  38
  39     def my_callback(self, loop):
  40         self.pos += 1
  41         return True
  42
  43     def setup_timeout(self, loop):
  44         timeout = GLib.Timeout(500)
  45         timeout.set_callback(self.timeout_callback, loop)
  46         timeout.attach()
  47
  48     def test_sources(self):
  49         loop = GLib.MainLoop()
  50
  51         self.setup_timeout(loop)
  52
  53         idle = Idle(loop)
  54         self.assertEqual(idle.get_context(), None)
  55         idle.attach()
  56         self.assertEqual(idle.get_context(), GLib.main_context_default())
  57
  58         self.pos = 0
  59
  60         m = MySource()
  61         self.assertEqual(m.get_context(), None)
  62         m.set_callback(self.my_callback, loop)
  63         m.attach()
  64         self.assertEqual(m.get_context(), GLib.main_context_default())
  65
  66         loop.run()
  67
  68         m.destroy()
  69         idle.destroy()
  70
  71         self.assertGreater(self.pos, 0)
  72         self.assertGreaterEqual(idle.count, 0)
  73         self.assertTrue(m.is_destroyed())
  74         self.assertTrue(idle.is_destroyed())
  75
  76     def test_source_prepare(self):
  77         # this test may not terminate if prepare() is wrapped incorrectly
  78         dispatched = [False]
  79         loop = GLib.MainLoop()
  80
  81         class CustomTimeout(GLib.Source):
  82             def prepare(self):
  83                 return (False, 10)
  84
  85             def check(self):
  86                 return True
  87
  88             def dispatch(self, callback, args):
  89                 dispatched[0] = True
  90
  91                 loop.quit()
  92
  93                 return False
  94
  95         source = CustomTimeout()
  96
  97         source.attach()
  98         source.set_callback(dir)
  99
 100         loop.run()
 101
 102         assert dispatched[0]
 103
 104     def test_is_destroyed_simple(self):
 105         s = GLib.Source()
 106
 107         self.assertFalse(s.is_destroyed())
 108         s.destroy()
 109         self.assertTrue(s.is_destroyed())
 110
 111         c = GLib.MainContext()
 112         s = GLib.Source()
 113         s.attach(c)
 114         self.assertFalse(s.is_destroyed())
 115         s.destroy()
 116         self.assertTrue(s.is_destroyed())
 117
 118     def test_is_destroyed_context(self):
 119         def f():
 120             c = GLib.MainContext()
 121             s = GLib.Source()
 122             s.attach(c)
 123             return s
 124
 125         s = f()
 126         self.assertTrue(s.is_destroyed())
 127
 128     def test_remove(self):
 129         s = GLib.idle_add(dir)
 130         self.assertEqual(GLib.source_remove(s), True)
 131         # s is now removed, should fail now
 132         self.assertEqual(GLib.source_remove(s), False)
 133
 134         # accepts large source IDs (they are unsigned)
 135         self.assertEqual(GLib.source_remove(GObject.G_MAXINT32), False)
 136         self.assertEqual(GLib.source_remove(GObject.G_MAXINT32 + 1), False)
 137         self.assertEqual(GLib.source_remove(GObject.G_MAXUINT32), False)
 138
 139     def test_recurse_property(self):
 140         s = GLib.Idle()
 141         self.assertTrue(s.can_recurse in [False, True])
 142         s.can_recurse = False
 143         self.assertFalse(s.can_recurse)
 144
 145     def test_priority(self):
 146         s = GLib.Idle()
 147         self.assertEqual(s.priority, GLib.PRIORITY_DEFAULT_IDLE)
 148         s.priority = GLib.PRIORITY_HIGH
 149         self.assertEqual(s.priority, GLib.PRIORITY_HIGH)
 150
 151         s = GLib.Idle(GLib.PRIORITY_LOW)
 152         self.assertEqual(s.priority, GLib.PRIORITY_LOW)
 153
 154         s = GLib.Timeout(1, GLib.PRIORITY_LOW)
 155         self.assertEqual(s.priority, GLib.PRIORITY_LOW)
 156
 157         s = GLib.Source()
 158         self.assertEqual(s.priority, GLib.PRIORITY_DEFAULT)
 159
 160     def test_get_current_time(self):
 161         # Note, deprecated API
 162         s = GLib.Idle()
 163         with warnings.catch_warnings(record=True) as w:
 164             warnings.simplefilter('always')
 165             time = s.get_current_time()
 166             self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))
 167
 168         self.assertTrue(isinstance(time, float))
 169         # plausibility check, and check magnitude of result
 170         self.assertGreater(time, 1300000000.0)
 171         self.assertLess(time, 2000000000.0)
 172
 173     def test_add_remove_poll(self):
 174         # FIXME: very shallow test, only verifies the API signature
 175         pollfd = GLib.PollFD(99, GLib.IOCondition.IN | GLib.IOCondition.HUP)
 176         self.assertEqual(pollfd.fd, 99)
 177         source = GLib.Source()
 178         source.add_poll(pollfd)
 179         source.remove_poll(pollfd)
 180
 181     def test_out_of_scope_before_dispatch(self):
 182         # https://bugzilla.gnome.org/show_bug.cgi?id=504337
 183         GLib.Timeout(20)
 184         GLib.Idle()
 185
 186     def test_finalize(self):
 187         self.dispatched = False
 188         self.finalized = False
 189
 190         class S(GLib.Source):
 191             def prepare(s):
 192                 return (True, 1)
 193
 194             def dispatch(s, callback, args):
 195                 self.dispatched = True
 196                 return False
 197
 198             def finalize(s):
 199                 self.finalized = True
 200
 201         source = S()
 202         id = source.attach()
 203         print('source id:', id)
 204         self.assertFalse(self.finalized)
 205         self.assertFalse(source.is_destroyed())
 206
 207         while source.get_context().iteration(False):
 208             pass
 209
 210         source.destroy()
 211         self.assertTrue(self.dispatched)
 212         self.assertFalse(self.finalized)
 213         self.assertTrue(source.is_destroyed())
 214         del source
 215         self.assertTrue(self.finalized)
 216
 217
 218 class TestUserData(unittest.TestCase):
 219     def test_idle_no_data(self):
 220         ml = GLib.MainLoop()
 221
 222         def cb():
 223             ml.quit()
 224         id = GLib.idle_add(cb)
 225         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 226                          GLib.PRIORITY_DEFAULT_IDLE)
 227         ml.run()
 228
 229     def test_timeout_no_data(self):
 230         ml = GLib.MainLoop()
 231
 232         def cb():
 233             ml.quit()
 234         id = GLib.timeout_add(50, cb)
 235         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 236                          GLib.PRIORITY_DEFAULT)
 237         ml.run()
 238
 239     def test_idle_data(self):
 240         ml = GLib.MainLoop()
 241
 242         def cb(data):
 243             data['called'] = True
 244             ml.quit()
 245         data = {}
 246         id = GLib.idle_add(cb, data)
 247         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 248                          GLib.PRIORITY_DEFAULT_IDLE)
 249         ml.run()
 250         self.assertTrue(data['called'])
 251
 252     def test_idle_multidata(self):
 253         ml = GLib.MainLoop()
 254
 255         def cb(data, data2):
 256             data['called'] = True
 257             data['data2'] = data2
 258             ml.quit()
 259         data = {}
 260         id = GLib.idle_add(cb, data, 'hello')
 261         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 262                          GLib.PRIORITY_DEFAULT_IDLE)
 263         ml.run()
 264         self.assertTrue(data['called'])
 265         self.assertEqual(data['data2'], 'hello')
 266
 267     def test_timeout_data(self):
 268         ml = GLib.MainLoop()
 269
 270         def cb(data):
 271             data['called'] = True
 272             ml.quit()
 273         data = {}
 274         id = GLib.timeout_add(50, cb, data)
 275         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 276                          GLib.PRIORITY_DEFAULT)
 277         ml.run()
 278         self.assertTrue(data['called'])
 279
 280     def test_timeout_multidata(self):
 281         ml = GLib.MainLoop()
 282
 283         def cb(data, data2):
 284             data['called'] = True
 285             data['data2'] = data2
 286             ml.quit()
 287         data = {}
 288         id = GLib.timeout_add(50, cb, data, 'hello')
 289         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 290                          GLib.PRIORITY_DEFAULT)
 291         ml.run()
 292         self.assertTrue(data['called'])
 293         self.assertEqual(data['data2'], 'hello')
 294
 295     def test_idle_no_data_priority(self):
 296         ml = GLib.MainLoop()
 297
 298         def cb():
 299             ml.quit()
 300         id = GLib.idle_add(cb, priority=GLib.PRIORITY_HIGH)
 301         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 302                          GLib.PRIORITY_HIGH)
 303         ml.run()
 304
 305     def test_timeout_no_data_priority(self):
 306         ml = GLib.MainLoop()
 307
 308         def cb():
 309             ml.quit()
 310         id = GLib.timeout_add(50, cb, priority=GLib.PRIORITY_HIGH)
 311         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 312                          GLib.PRIORITY_HIGH)
 313         ml.run()
 314
 315     def test_idle_data_priority(self):
 316         ml = GLib.MainLoop()
 317
 318         def cb(data):
 319             data['called'] = True
 320             ml.quit()
 321         data = {}
 322         id = GLib.idle_add(cb, data, priority=GLib.PRIORITY_HIGH)
 323         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 324                          GLib.PRIORITY_HIGH)
 325         ml.run()
 326         self.assertTrue(data['called'])
 327
 328     def test_timeout_data_priority(self):
 329         ml = GLib.MainLoop()
 330
 331         def cb(data):
 332             data['called'] = True
 333             ml.quit()
 334         data = {}
 335         id = GLib.timeout_add(50, cb, data, priority=GLib.PRIORITY_HIGH)
 336         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 337                          GLib.PRIORITY_HIGH)
 338         ml.run()
 339         self.assertTrue(data['called'])
 340
 341     def cb_no_data(self):
 342         self.loop.quit()
 343
 344     def test_idle_method_callback_no_data(self):
 345         self.loop = GLib.MainLoop()
 346         GLib.idle_add(self.cb_no_data)
 347         self.loop.run()
 348
 349     def cb_with_data(self, data):
 350         data['called'] = True
 351         self.loop.quit()
 352
 353     def test_idle_method_callback_with_data(self):
 354         self.loop = GLib.MainLoop()
 355         data = {}
 356         GLib.idle_add(self.cb_with_data, data)
 357         self.loop.run()
 358         self.assertTrue(data['called'])
 359
 360
 361 if __name__ == '__main__':
 362     unittest.main()