blob: 947bec60decff3fdd48467842c677cf813049e62 (
plain)
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
|
from __future__ import absolute_import
from aiogreen import socketpair
import tests
class AddReaderTests(tests.TestCase):
def test_add_reader(self):
result = {'received': None}
rsock, wsock = socketpair()
self.addCleanup(rsock.close)
self.addCleanup(wsock.close)
def reader():
data = rsock.recv(100)
result['received'] = data
self.loop.remove_reader(rsock)
self.loop.stop()
def writer():
self.loop.remove_writer(wsock)
self.loop.call_soon(wsock.send, b'abc')
self.loop.add_reader(rsock, reader)
self.loop.add_writer(wsock, writer)
self.loop.run_forever()
self.assertEqual(result['received'], b'abc')
if __name__ == '__main__':
import unittest
unittest.main()
|
