Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 1 | #!/usr/bin/env python |
| 2 | |
| 3 | import sched |
| 4 | import time |
| 5 | import unittest |
| 6 | from test import support |
| 7 | |
| 8 | |
| 9 | class TestCase(unittest.TestCase): |
| 10 | |
| 11 | def test_enter(self): |
| 12 | l = [] |
| 13 | fun = lambda x: l.append(x) |
| 14 | scheduler = sched.scheduler(time.time, time.sleep) |
Charles-François Natali | f670ca5 | 2012-02-16 19:49:48 +0100 | [diff] [blame] | 15 | for x in [0.5, 0.4, 0.3, 0.2, 0.1]: |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 16 | z = scheduler.enter(x, 1, fun, (x,)) |
| 17 | scheduler.run() |
Charles-François Natali | f670ca5 | 2012-02-16 19:49:48 +0100 | [diff] [blame] | 18 | self.assertEqual(l, [0.1, 0.2, 0.3, 0.4, 0.5]) |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 19 | |
| 20 | def test_enterabs(self): |
| 21 | l = [] |
| 22 | fun = lambda x: l.append(x) |
| 23 | scheduler = sched.scheduler(time.time, time.sleep) |
| 24 | for x in [0.05, 0.04, 0.03, 0.02, 0.01]: |
| 25 | z = scheduler.enterabs(x, 1, fun, (x,)) |
| 26 | scheduler.run() |
| 27 | self.assertEqual(l, [0.01, 0.02, 0.03, 0.04, 0.05]) |
| 28 | |
| 29 | def test_priority(self): |
| 30 | l = [] |
| 31 | fun = lambda x: l.append(x) |
| 32 | scheduler = sched.scheduler(time.time, time.sleep) |
| 33 | for priority in [1, 2, 3, 4, 5]: |
Charles-François Natali | f670ca5 | 2012-02-16 19:49:48 +0100 | [diff] [blame] | 34 | z = scheduler.enterabs(0.01, priority, fun, (priority,)) |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 35 | scheduler.run() |
| 36 | self.assertEqual(l, [1, 2, 3, 4, 5]) |
| 37 | |
| 38 | def test_cancel(self): |
| 39 | l = [] |
| 40 | fun = lambda x: l.append(x) |
| 41 | scheduler = sched.scheduler(time.time, time.sleep) |
Charles-François Natali | f670ca5 | 2012-02-16 19:49:48 +0100 | [diff] [blame] | 42 | now = time.time() |
| 43 | event1 = scheduler.enterabs(now + 0.01, 1, fun, (0.01,)) |
| 44 | event2 = scheduler.enterabs(now + 0.02, 1, fun, (0.02,)) |
| 45 | event3 = scheduler.enterabs(now + 0.03, 1, fun, (0.03,)) |
| 46 | event4 = scheduler.enterabs(now + 0.04, 1, fun, (0.04,)) |
| 47 | event5 = scheduler.enterabs(now + 0.05, 1, fun, (0.05,)) |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 48 | scheduler.cancel(event1) |
| 49 | scheduler.cancel(event5) |
| 50 | scheduler.run() |
| 51 | self.assertEqual(l, [0.02, 0.03, 0.04]) |
| 52 | |
| 53 | def test_empty(self): |
| 54 | l = [] |
| 55 | fun = lambda x: l.append(x) |
| 56 | scheduler = sched.scheduler(time.time, time.sleep) |
| 57 | self.assertTrue(scheduler.empty()) |
| 58 | for x in [0.05, 0.04, 0.03, 0.02, 0.01]: |
| 59 | z = scheduler.enterabs(x, 1, fun, (x,)) |
| 60 | self.assertFalse(scheduler.empty()) |
| 61 | scheduler.run() |
| 62 | self.assertTrue(scheduler.empty()) |
| 63 | |
| 64 | def test_queue(self): |
| 65 | l = [] |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 66 | fun = lambda x: l.append(x) |
| 67 | scheduler = sched.scheduler(time.time, time.sleep) |
Charles-François Natali | 4a72ebe | 2012-02-16 19:51:45 +0100 | [diff] [blame] | 68 | now = time.time() |
| 69 | e5 = scheduler.enterabs(now + 0.05, 1, fun) |
| 70 | e1 = scheduler.enterabs(now + 0.01, 1, fun) |
| 71 | e2 = scheduler.enterabs(now + 0.02, 1, fun) |
| 72 | e4 = scheduler.enterabs(now + 0.04, 1, fun) |
| 73 | e3 = scheduler.enterabs(now + 0.03, 1, fun) |
Giampaolo Rodola' | 6a5dcd4 | 2011-11-26 12:17:42 +0100 | [diff] [blame] | 74 | # queue property is supposed to return an order list of |
| 75 | # upcoming events |
| 76 | self.assertEqual(list(scheduler.queue), [e1, e2, e3, e4, e5]) |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 77 | |
Giampaolo Rodola' | be55d99 | 2011-11-22 13:33:34 +0100 | [diff] [blame] | 78 | def test_args_kwargs(self): |
| 79 | flag = [] |
| 80 | |
| 81 | def fun(*a, **b): |
| 82 | flag.append(None) |
| 83 | self.assertEqual(a, (1,2,3)) |
| 84 | self.assertEqual(b, {"foo":1}) |
| 85 | |
| 86 | scheduler = sched.scheduler(time.time, time.sleep) |
| 87 | z = scheduler.enterabs(0.01, 1, fun, argument=(1,2,3), kwargs={"foo":1}) |
| 88 | scheduler.run() |
| 89 | self.assertEqual(flag, [None]) |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 90 | |
Giampaolo Rodola' | 556ba04 | 2011-12-14 14:38:45 +0100 | [diff] [blame] | 91 | def test_run_non_blocking(self): |
| 92 | l = [] |
| 93 | fun = lambda x: l.append(x) |
| 94 | scheduler = sched.scheduler(time.time, time.sleep) |
| 95 | for x in [10, 9, 8, 7, 6]: |
| 96 | scheduler.enter(x, 1, fun, (x,)) |
| 97 | scheduler.run(blocking=False) |
| 98 | self.assertEqual(l, []) |
| 99 | |
| 100 | |
Giampaolo Rodolà | b5c2376 | 2010-08-04 09:28:05 +0000 | [diff] [blame] | 101 | def test_main(): |
| 102 | support.run_unittest(TestCase) |
| 103 | |
| 104 | if __name__ == "__main__": |
| 105 | test_main() |