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