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
 187 class TestUserData(unittest.TestCase):
 188     def test_idle_no_data(self):
 189         ml = GLib.MainLoop()
 190
 191         def cb():
 192             ml.quit()
 193         id = GLib.idle_add(cb)
 194         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 195                          GLib.PRIORITY_DEFAULT_IDLE)
 196         ml.run()
 197
 198     def test_timeout_no_data(self):
 199         ml = GLib.MainLoop()
 200
 201         def cb():
 202             ml.quit()
 203         id = GLib.timeout_add(50, cb)
 204         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 205                          GLib.PRIORITY_DEFAULT)
 206         ml.run()
 207
 208     def test_idle_data(self):
 209         ml = GLib.MainLoop()
 210
 211         def cb(data):
 212             data['called'] = True
 213             ml.quit()
 214         data = {}
 215         id = GLib.idle_add(cb, data)
 216         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 217                          GLib.PRIORITY_DEFAULT_IDLE)
 218         ml.run()
 219         self.assertTrue(data['called'])
 220
 221     def test_idle_multidata(self):
 222         ml = GLib.MainLoop()
 223
 224         def cb(data, data2):
 225             data['called'] = True
 226             data['data2'] = data2
 227             ml.quit()
 228         data = {}
 229         id = GLib.idle_add(cb, data, 'hello')
 230         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 231                          GLib.PRIORITY_DEFAULT_IDLE)
 232         ml.run()
 233         self.assertTrue(data['called'])
 234         self.assertEqual(data['data2'], 'hello')
 235
 236     def test_timeout_data(self):
 237         ml = GLib.MainLoop()
 238
 239         def cb(data):
 240             data['called'] = True
 241             ml.quit()
 242         data = {}
 243         id = GLib.timeout_add(50, cb, data)
 244         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 245                          GLib.PRIORITY_DEFAULT)
 246         ml.run()
 247         self.assertTrue(data['called'])
 248
 249     def test_timeout_multidata(self):
 250         ml = GLib.MainLoop()
 251
 252         def cb(data, data2):
 253             data['called'] = True
 254             data['data2'] = data2
 255             ml.quit()
 256         data = {}
 257         id = GLib.timeout_add(50, cb, data, 'hello')
 258         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 259                          GLib.PRIORITY_DEFAULT)
 260         ml.run()
 261         self.assertTrue(data['called'])
 262         self.assertEqual(data['data2'], 'hello')
 263
 264     def test_idle_no_data_priority(self):
 265         ml = GLib.MainLoop()
 266
 267         def cb():
 268             ml.quit()
 269         id = GLib.idle_add(cb, priority=GLib.PRIORITY_HIGH)
 270         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 271                          GLib.PRIORITY_HIGH)
 272         ml.run()
 273
 274     def test_timeout_no_data_priority(self):
 275         ml = GLib.MainLoop()
 276
 277         def cb():
 278             ml.quit()
 279         id = GLib.timeout_add(50, cb, priority=GLib.PRIORITY_HIGH)
 280         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 281                          GLib.PRIORITY_HIGH)
 282         ml.run()
 283
 284     def test_idle_data_priority(self):
 285         ml = GLib.MainLoop()
 286
 287         def cb(data):
 288             data['called'] = True
 289             ml.quit()
 290         data = {}
 291         id = GLib.idle_add(cb, data, priority=GLib.PRIORITY_HIGH)
 292         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 293                          GLib.PRIORITY_HIGH)
 294         ml.run()
 295         self.assertTrue(data['called'])
 296
 297     def test_timeout_data_priority(self):
 298         ml = GLib.MainLoop()
 299
 300         def cb(data):
 301             data['called'] = True
 302             ml.quit()
 303         data = {}
 304         id = GLib.timeout_add(50, cb, data, priority=GLib.PRIORITY_HIGH)
 305         self.assertEqual(ml.get_context().find_source_by_id(id).priority,
 306                          GLib.PRIORITY_HIGH)
 307         ml.run()
 308         self.assertTrue(data['called'])
 309
 310     def cb_no_data(self):
 311         self.loop.quit()
 312
 313     def test_idle_method_callback_no_data(self):
 314         self.loop = GLib.MainLoop()
 315         GLib.idle_add(self.cb_no_data)
 316         self.loop.run()
 317
 318     def cb_with_data(self, data):
 319         data['called'] = True
 320         self.loop.quit()
 321
 322     def test_idle_method_callback_with_data(self):
 323         self.loop = GLib.MainLoop()
 324         data = {}
 325         GLib.idle_add(self.cb_with_data, data)
 326         self.loop.run()
 327         self.assertTrue(data['called'])
 328
 329
 330 if __name__ == '__main__':
 331     unittest.main()