|
1 import unittest |
|
2 |
|
3 from qmsk.net.lib.event2.base import event_base |
|
4 from qmsk.net.lib.event2.event import event |
|
5 from qmsk.net.lib.event2.constants import * |
|
6 |
|
7 import os |
|
8 |
|
9 class MyEvent (event) : |
|
10 def __init__ (self, base, fd, ev) : |
|
11 event.__init__(self, base, fd, ev) |
|
12 |
|
13 self.ev_in = ev |
|
14 self.ev_out = None |
|
15 |
|
16 self.res = None |
|
17 |
|
18 def __call__ (self, fd, mask) : |
|
19 self.ev_out = mask |
|
20 |
|
21 self.res = (mask == self.ev_in) |
|
22 |
|
23 class TestEventBase (unittest.TestCase) : |
|
24 """ |
|
25 Simple event_base bits |
|
26 """ |
|
27 |
|
28 def setUp (self) : |
|
29 self.ev_base = event_base() |
|
30 |
|
31 def test_loop_empty (self) : |
|
32 """ |
|
33 loop() on an empty event_base returns False and doesn't block |
|
34 """ |
|
35 |
|
36 self.assertFalse(self.ev_base.loop(nonblock=False, once=False)) |
|
37 |
|
38 def test_loop_empty_nb (self) : |
|
39 """ |
|
40 loop(nonblock=False) on an empty event_base returns False |
|
41 """ |
|
42 |
|
43 self.assertFalse(self.ev_base.loop(nonblock=True)) |
|
44 |
|
45 |
|
46 class TestEventPipe (unittest.TestCase) : |
|
47 """ |
|
48 Test basic event stuff using a pipe. |
|
49 """ |
|
50 |
|
51 def setUp (self) : |
|
52 self.ev_base = event_base() |
|
53 self.fd_read, self.fd_write = os.pipe() |
|
54 |
|
55 def test_event (self) : |
|
56 """ |
|
57 Constructing an event works |
|
58 """ |
|
59 |
|
60 self.assertTrue(event(self.ev_base, self.fd_read, EV_READ)) |
|
61 |
|
62 def test_read (self) : |
|
63 """ |
|
64 Read event on pipe with data fires |
|
65 """ |
|
66 |
|
67 ev = MyEvent(self.ev_base, self.fd_read, EV_READ) |
|
68 ev.add(0.1) # 100ms |
|
69 |
|
70 # trigger |
|
71 os.write(self.fd_write, "foo") |
|
72 |
|
73 # loop |
|
74 self.ev_base.loop(nonblock=True) |
|
75 |
|
76 # test |
|
77 self.assertEquals(ev.ev_out, EV_READ) |
|
78 |
|
79 def test_read_timeout (self) : |
|
80 """ |
|
81 Read event on empty pipe timeouts |
|
82 """ |
|
83 |
|
84 ev = MyEvent(self.ev_base, self.fd_read, EV_READ) |
|
85 ev.add(0.0) |
|
86 |
|
87 # loop |
|
88 self.ev_base.loop(once=True) |
|
89 |
|
90 # test |
|
91 self.assertEquals(ev.ev_out, EV_TIMEOUT) |
|
92 |
|
93 def test_read_nonblock (self) : |
|
94 """ |
|
95 Read event on empty pipe doesn't block loop with nonblock=True |
|
96 """ |
|
97 |
|
98 ev = MyEvent(self.ev_base, self.fd_read, EV_READ) |
|
99 ev.add() |
|
100 |
|
101 # loop once |
|
102 self.ev_base.loop(nonblock=True) |
|
103 |
|
104 # test |
|
105 self.assertEquals(ev.ev_out, None) |
|
106 |
|
107 if __name__ == '__main__' : |
|
108 unittest.main() |