1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
import os
import unittest
import platform
import urwid
from urwid.compat import PYTHON3
class EventLoopTestMixin(object):
def test_event_loop(self):
rd, wr = os.pipe()
evl = self.evl
out = []
def step1():
out.append("writing")
os.write(wr, "hi".encode('ascii'))
def step2():
out.append(os.read(rd, 2).decode('ascii'))
raise urwid.ExitMainLoop
handle = evl.alarm(0, step1)
handle = evl.watch_file(rd, step2)
evl.run()
self.assertEqual(out, ["writing", "hi"])
def test_remove_alarm(self):
evl = self.evl
handle = evl.alarm(50, lambda: None)
self.assertTrue(evl.remove_alarm(handle))
self.assertFalse(evl.remove_alarm(handle))
def test_remove_watch_file(self):
evl = self.evl
handle = evl.watch_file(5, lambda: None)
self.assertTrue(evl.remove_watch_file(handle))
self.assertFalse(evl.remove_watch_file(handle))
def test_run(self):
evl = self.evl
out = []
rd, wr = os.pipe()
self.assertEqual(os.write(wr, "data".encode('ascii')), 4)
def say_hello():
out.append("hello")
def say_waiting():
out.append("waiting")
def exit_clean():
out.append("clean exit")
raise urwid.ExitMainLoop
def exit_error():
1/0
handle = evl.alarm(0.01, exit_clean)
handle = evl.alarm(0.005, say_hello)
self.assertEqual(evl.enter_idle(say_waiting), 1)
evl.run()
self.assertTrue("hello" in out, out)
self.assertTrue("clean exit"in out, out)
handle = evl.watch_file(rd, exit_clean)
del out[:]
evl.run()
self.assertEqual(out, ["clean exit"])
self.assertTrue(evl.remove_watch_file(handle))
handle = evl.alarm(0, exit_error)
self.assertRaises(ZeroDivisionError, evl.run)
handle = evl.watch_file(rd, exit_error)
self.assertRaises(ZeroDivisionError, evl.run)
class SelectEventLoopTest(unittest.TestCase, EventLoopTestMixin):
def setUp(self):
self.evl = urwid.SelectEventLoop()
try:
import gi.repository
except ImportError:
pass
else:
class GLibEventLoopTest(unittest.TestCase, EventLoopTestMixin):
def setUp(self):
self.evl = urwid.GLibEventLoop()
try:
import tornado
except ImportError:
pass
else:
class TornadoEventLoopTest(unittest.TestCase, EventLoopTestMixin):
def setUp(self):
from tornado.ioloop import IOLoop
self.evl = urwid.TornadoEventLoop(IOLoop())
try:
import twisted
except ImportError:
pass
else:
class TwistedEventLoopTest(unittest.TestCase, EventLoopTestMixin):
def setUp(self):
self.evl = urwid.TwistedEventLoop()
# can't restart twisted reactor, so use shortened tests
def test_event_loop(self):
pass
def test_run(self):
evl = self.evl
out = []
rd, wr = os.pipe()
self.assertEqual(os.write(wr, "data".encode('ascii')), 4)
def step2():
out.append(os.read(rd, 2).decode('ascii'))
def say_hello():
out.append("hello")
def say_waiting():
out.append("waiting")
def exit_clean():
out.append("clean exit")
raise urwid.ExitMainLoop
def exit_error():
1/0
handle = evl.watch_file(rd, step2)
handle = evl.alarm(0.01, exit_clean)
handle = evl.alarm(0.005, say_hello)
self.assertEqual(evl.enter_idle(say_waiting), 1)
evl.run()
self.assertTrue("da" in out, out)
self.assertTrue("ta" in out, out)
self.assertTrue("hello" in out, out)
self.assertTrue("clean exit" in out, out)
|
