1#!/usr/bin/env python3
2# This file is part of Xpra.
3# Copyright (C) 2019-2020 Antoine Martin <antoine@xpra.org>
4# Xpra is released under the terms of the GNU GPL v2, or, at your option, any
5# later version. See the file COPYING for details.
6
7import unittest
8
9from xpra import queue_scheduler
10QueueScheduler = queue_scheduler.QueueScheduler
11
12
13class QueueSchedulerTest(unittest.TestCase):
14
15    def test_idle_timeout(self):
16        qs = QueueScheduler()
17        calls = []
18        def idle_add(arg):
19            calls.append(arg)
20        qs.idle_add(idle_add, True)
21        def timeout_add(arg):
22            calls.append(arg)
23        qs.timeout_add(100, timeout_add, True)
24        qs.timeout_add(500, qs.stop)
25        qs.run()
26        assert len(calls)==2
27
28    def test_idle_repeat(self):
29        qs = QueueScheduler()
30        calls = []
31        def idle_add(arg):
32            calls.append(arg)
33            return len(calls)<10
34        qs.idle_add(idle_add, True)
35        qs.timeout_add(500, qs.stop)
36        qs.run()
37        assert len(calls)==10
38
39    def test_idle_cancel(self):
40        qs = QueueScheduler()
41        calls = []
42        def idle_add():
43            calls.append(True)
44            return True
45        tid = qs.idle_add(idle_add)
46        copy = []
47        def cancel():
48            qs.source_remove(tid)
49            copy[:] = list(calls)
50            return False
51        qs.timeout_add(500, cancel)
52        qs.timeout_add(1000, qs.stop)
53        qs.run()
54        assert len(calls)==len(copy), "idle_add continued to run!"
55
56    def test_timeout_cancel(self):
57        qs = QueueScheduler()
58        calls = []
59        def timeout_add():
60            calls.append(True)
61            return True
62        tid = qs.timeout_add(10, timeout_add)
63        copy = []
64        def cancel():
65            qs.source_remove(tid)
66            copy[:] = list(calls)
67            return False
68        qs.timeout_add(500, cancel)
69        qs.timeout_add(1000, qs.stop)
70        qs.run()
71        assert len(calls)==len(copy), "idle_add continued to run!"
72
73    def test_source_remove(self):
74        qs = QueueScheduler()
75        calls = []
76        def timeout_add(arg):
77            calls.append(arg)
78            return True
79        t = qs.timeout_add(100, timeout_add, True)
80        qs.source_remove(t)
81        qs.timeout_add(500, qs.stop)
82        qs.run()
83        assert not calls
84
85    def test_invalid_remove(self):
86        qs = QueueScheduler()
87        qs.source_remove(-1)
88
89    def test_idle_raises_exception(self):
90        qs = QueueScheduler()
91        def raise_exception():
92            raise Exception("test scheduler error handling")
93        from unit.test_util import silence_error
94        with silence_error(queue_scheduler):
95            qs.idle_add(raise_exception)
96            qs.timeout_add(500, qs.stop)
97            qs.run()
98
99    def test_stop_queue(self):
100        qs = QueueScheduler()
101        qs.idle_add(qs.stop_main_queue)
102        qs.run()
103
104    def test_timer_repeat(self):
105        times = list(range(10))
106        qs = QueueScheduler()
107        def timer_fn():
108            times.pop()
109            if not times:
110                qs.stop()
111                return False
112            return True
113        qs.timeout_add(1, timer_fn)
114        qs.timeout_add(2000, qs.stop)
115        qs.run()
116        assert not times, "items remain in list: %s" % (times,)
117
118def main():
119    unittest.main()
120
121if __name__ == '__main__':
122    main()
123