path: root/tests/test_source.py

# -*- Mode: Python -*-

import unittest
import warnings

from gi.repository import GLib, GObject
from gi import PyGIDeprecationWarning


class Idle(GLib.Idle):
    def __init__(self, loop):
        GLib.Idle.__init__(self)
        self.count = 0
        self.set_callback(self.callback, loop)

    def callback(self, loop):
        self.count += 1
        return True


class MySource(GLib.Source):
    def __init__(self):
        GLib.Source.__init__(self)

    def prepare(self):
        return True, 0

    def check(self):
        return True

    def dispatch(self, callback, args):
        return callback(*args)


class TestSource(unittest.TestCase):
    def timeout_callback(self, loop):
        loop.quit()

    def my_callback(self, loop):
        self.pos += 1
        return True

    def setup_timeout(self, loop):
        timeout = GLib.Timeout(500)
        timeout.set_callback(self.timeout_callback, loop)
        timeout.attach()

    def test_sources(self):
        loop = GLib.MainLoop()

        self.setup_timeout(loop)

        idle = Idle(loop)
        self.assertEqual(idle.get_context(), None)
        idle.attach()
        self.assertEqual(idle.get_context(), GLib.main_context_default())

        self.pos = 0

        m = MySource()
        self.assertEqual(m.get_context(), None)
        m.set_callback(self.my_callback, loop)
        m.attach()
        self.assertEqual(m.get_context(), GLib.main_context_default())

        loop.run()

        m.destroy()
        idle.destroy()

        self.assertGreater(self.pos, 0)
        self.assertGreaterEqual(idle.count, 0)
        self.assertTrue(m.is_destroyed())
        self.assertTrue(idle.is_destroyed())

    def test_source_prepare(self):
        # this test may not terminate if prepare() is wrapped incorrectly
        dispatched = [False]
        loop = GLib.MainLoop()

        class CustomTimeout(GLib.Source):
            def prepare(self):
                return (False, 10)

            def check(self):
                return True

            def dispatch(self, callback, args):
                dispatched[0] = True

                loop.quit()

                return False

        source = CustomTimeout()

        source.attach()
        source.set_callback(dir)

        loop.run()

        assert dispatched[0]

    def test_is_destroyed_simple(self):
        s = GLib.Source()

        self.assertFalse(s.is_destroyed())
        s.destroy()
        self.assertTrue(s.is_destroyed())

        c = GLib.MainContext()
        s = GLib.Source()
        s.attach(c)
        self.assertFalse(s.is_destroyed())
        s.destroy()
        self.assertTrue(s.is_destroyed())

    def test_is_destroyed_context(self):
        def f():
            c = GLib.MainContext()
            s = GLib.Source()
            s.attach(c)
            return s

        s = f()
        self.assertTrue(s.is_destroyed())

    def test_remove(self):
        s = GLib.idle_add(dir)
        self.assertEqual(GLib.source_remove(s), True)
        # s is now removed, should fail now
        self.assertEqual(GLib.source_remove(s), False)

        # accepts large source IDs (they are unsigned)
        self.assertEqual(GLib.source_remove(GObject.G_MAXINT32), False)
        self.assertEqual(GLib.source_remove(GObject.G_MAXINT32 + 1), False)
        self.assertEqual(GLib.source_remove(GObject.G_MAXUINT32), False)

    def test_recurse_property(self):
        s = GLib.Idle()
        self.assertTrue(s.can_recurse in [False, True])
        s.can_recurse = False
        self.assertFalse(s.can_recurse)

    def test_priority(self):
        s = GLib.Idle()
        self.assertEqual(s.priority, GLib.PRIORITY_DEFAULT_IDLE)
        s.priority = GLib.PRIORITY_HIGH
        self.assertEqual(s.priority, GLib.PRIORITY_HIGH)

        s = GLib.Idle(GLib.PRIORITY_LOW)
        self.assertEqual(s.priority, GLib.PRIORITY_LOW)

        s = GLib.Timeout(1, GLib.PRIORITY_LOW)
        self.assertEqual(s.priority, GLib.PRIORITY_LOW)

        s = GLib.Source()
        self.assertEqual(s.priority, GLib.PRIORITY_DEFAULT)

    def test_get_current_time(self):
        # Note, deprecated API
        s = GLib.Idle()
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter('always')
            time = s.get_current_time()
            self.assertTrue(issubclass(w[0].category, PyGIDeprecationWarning))

        self.assertTrue(isinstance(time, float))
        # plausibility check, and check magnitude of result
        self.assertGreater(time, 1300000000.0)
        self.assertLess(time, 2000000000.0)

    def test_add_remove_poll(self):
        # FIXME: very shallow test, only verifies the API signature
        pollfd = GLib.PollFD(99, GLib.IOCondition.IN | GLib.IOCondition.HUP)
        self.assertEqual(pollfd.fd, 99)
        source = GLib.Source()
        source.add_poll(pollfd)
        source.remove_poll(pollfd)

    def test_out_of_scope_before_dispatch(self):
        # https://bugzilla.gnome.org/show_bug.cgi?id=504337
        GLib.Timeout(20)
        GLib.Idle()

    def test_finalize(self):
        self.dispatched = False
        self.finalized = False

        class S(GLib.Source):
            def prepare(s):
                return (True, 1)

            def dispatch(s, callback, args):
                self.dispatched = True
                return False

            def finalize(s):
                self.finalized = True

        source = S()
        id = source.attach()
        print('source id:', id)
        self.assertFalse(self.finalized)
        self.assertFalse(source.is_destroyed())

        while source.get_context().iteration(False):
            pass

        source.destroy()
        self.assertTrue(self.dispatched)
        self.assertFalse(self.finalized)
        self.assertTrue(source.is_destroyed())
        del source
        self.assertTrue(self.finalized)

    def test_extra_init_args(self):
        class SourceWithInitArgs(GLib.Source):
            def __init__(self, arg, kwarg=None):
                super(SourceWithInitArgs, self).__init__()
                self.arg = arg
                self.kwarg = kwarg

        source = SourceWithInitArgs(1, kwarg=2)
        self.assertEqual(source.arg, 1)
        self.assertEqual(source.kwarg, 2)


class TestUserData(unittest.TestCase):
    def test_idle_no_data(self):
        ml = GLib.MainLoop()

        def cb():
            ml.quit()
        id = GLib.idle_add(cb)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT_IDLE)
        ml.run()

    def test_timeout_no_data(self):
        ml = GLib.MainLoop()

        def cb():
            ml.quit()
        id = GLib.timeout_add(50, cb)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT)
        ml.run()

    def test_idle_data(self):
        ml = GLib.MainLoop()

        def cb(data):
            data['called'] = True
            ml.quit()
        data = {}
        id = GLib.idle_add(cb, data)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT_IDLE)
        ml.run()
        self.assertTrue(data['called'])

    def test_idle_multidata(self):
        ml = GLib.MainLoop()

        def cb(data, data2):
            data['called'] = True
            data['data2'] = data2
            ml.quit()
        data = {}
        id = GLib.idle_add(cb, data, 'hello')
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT_IDLE)
        ml.run()
        self.assertTrue(data['called'])
        self.assertEqual(data['data2'], 'hello')

    def test_timeout_data(self):
        ml = GLib.MainLoop()

        def cb(data):
            data['called'] = True
            ml.quit()
        data = {}
        id = GLib.timeout_add(50, cb, data)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT)
        ml.run()
        self.assertTrue(data['called'])

    def test_timeout_multidata(self):
        ml = GLib.MainLoop()

        def cb(data, data2):
            data['called'] = True
            data['data2'] = data2
            ml.quit()
        data = {}
        id = GLib.timeout_add(50, cb, data, 'hello')
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_DEFAULT)
        ml.run()
        self.assertTrue(data['called'])
        self.assertEqual(data['data2'], 'hello')

    def test_idle_no_data_priority(self):
        ml = GLib.MainLoop()

        def cb():
            ml.quit()
        id = GLib.idle_add(cb, priority=GLib.PRIORITY_HIGH)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_HIGH)
        ml.run()

    def test_timeout_no_data_priority(self):
        ml = GLib.MainLoop()

        def cb():
            ml.quit()
        id = GLib.timeout_add(50, cb, priority=GLib.PRIORITY_HIGH)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_HIGH)
        ml.run()

    def test_idle_data_priority(self):
        ml = GLib.MainLoop()

        def cb(data):
            data['called'] = True
            ml.quit()
        data = {}
        id = GLib.idle_add(cb, data, priority=GLib.PRIORITY_HIGH)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_HIGH)
        ml.run()
        self.assertTrue(data['called'])

    def test_timeout_data_priority(self):
        ml = GLib.MainLoop()

        def cb(data):
            data['called'] = True
            ml.quit()
        data = {}
        id = GLib.timeout_add(50, cb, data, priority=GLib.PRIORITY_HIGH)
        self.assertEqual(ml.get_context().find_source_by_id(id).priority,
                         GLib.PRIORITY_HIGH)
        ml.run()
        self.assertTrue(data['called'])

    def cb_no_data(self):
        self.loop.quit()

    def test_idle_method_callback_no_data(self):
        self.loop = GLib.MainLoop()
        GLib.idle_add(self.cb_no_data)
        self.loop.run()

    def cb_with_data(self, data):
        data['called'] = True
        self.loop.quit()

    def test_idle_method_callback_with_data(self):
        self.loop = GLib.MainLoop()
        data = {}
        GLib.idle_add(self.cb_with_data, data)
        self.loop.run()
        self.assertTrue(data['called'])


if __name__ == '__main__':
    unittest.main()