summaryrefslogtreecommitdiff
path: root/tests/test_add_reader.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_add_reader.py')
-rw-r--r--tests/test_add_reader.py81
1 files changed, 0 insertions, 81 deletions
diff --git a/tests/test_add_reader.py b/tests/test_add_reader.py
deleted file mode 100644
index 12de90b..0000000
--- a/tests/test_add_reader.py
+++ /dev/null
@@ -1,81 +0,0 @@
-from __future__ import absolute_import
-from aioeventlet import socketpair
-import eventlet
-import tests
-socket = eventlet.patcher.original('socket')
-
-
-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')
-
- def check_add_replace(self, event):
- selector = self.loop._selector
- if event == 'reader':
- add_sock = self.loop.add_reader
- remove_sock = self.loop.remove_reader
- def get_handle(fileobj):
- return selector.get_key(fileobj).data[0]
- else:
- add_sock = self.loop.add_writer
- remove_sock = self.loop.remove_writer
- def get_handle(fileobj):
- return selector.get_key(fileobj).data[1]
-
- sock = socket.socket()
- self.addCleanup(sock.close)
-
- def func():
- pass
-
- def func2():
- pass
-
- self.assertRaises(KeyError, get_handle, sock)
-
- add_sock(sock, func)
- handle1 = get_handle(sock)
- self.assertFalse(handle1._cancelled)
-
- add_sock(sock, func2)
- handle2 = get_handle(sock)
- self.assertIsNot(handle1, handle2)
- self.assertTrue(handle1._cancelled)
- self.assertFalse(handle2._cancelled)
-
- removed = remove_sock(sock)
- self.assertTrue(removed)
- self.assertTrue(handle2._cancelled)
-
- removed = remove_sock(sock)
- self.assertFalse(removed)
-
- def test_add_reader_replace(self):
- self.check_add_replace("reader")
-
- def test_add_writer_replace(self):
- self.check_add_replace("writer")
-
-
-if __name__ == '__main__':
- import unittest
- unittest.main()