home *** CD-ROM | disk | FTP | other *** search
/ Maximum CD 2011 July / maximum-cd-2011-07.iso / DiscContents / LibO_3.3.2_Win_x86_install_multi.exe / libreoffice1.cab / test_epoll.py < prev    next >
Encoding:
Python Source  |  2011-03-15  |  6.1 KB  |  196 lines
  1. # Copyright (c) 2001-2006 Twisted Matrix Laboratories.
  2. #
  3. # Permission is hereby granted, free of charge, to any person obtaining
  4. # a copy of this software and associated documentation files (the
  5. # "Software"), to deal in the Software without restriction, including
  6. # without limitation the rights to use, copy, modify, merge, publish,
  7. # distribute, sublicense, and/or sell copies of the Software, and to
  8. # permit persons to whom the Software is furnished to do so, subject to
  9. # the following conditions:
  10. #
  11. # The above copyright notice and this permission notice shall be
  12. # included in all copies or substantial portions of the Software.
  13. #
  14. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  15. # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16. # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  17. # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  18. # LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  19. # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  20. # WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  21. """
  22. Tests for epoll wrapper.
  23. """
  24. import os
  25. import socket
  26. import errno
  27. import time
  28. import select
  29. import tempfile
  30. import unittest
  31.  
  32. from test import test_support
  33. if not hasattr(select, "epoll"):
  34.     raise test_support.TestSkipped("test works only on Linux 2.6")
  35.  
  36. try:
  37.     select.epoll()
  38. except IOError, e:
  39.     if e.errno == errno.ENOSYS:
  40.         raise test_support.TestSkipped("kernel doesn't support epoll()")
  41.  
  42. class TestEPoll(unittest.TestCase):
  43.  
  44.     def setUp(self):
  45.         self.serverSocket = socket.socket()
  46.         self.serverSocket.bind(('127.0.0.1', 0))
  47.         self.serverSocket.listen(1)
  48.         self.connections = [self.serverSocket]
  49.  
  50.  
  51.     def tearDown(self):
  52.         for skt in self.connections:
  53.             skt.close()
  54.  
  55.     def _connected_pair(self):
  56.         client = socket.socket()
  57.         client.setblocking(False)
  58.         try:
  59.             client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
  60.         except socket.error, e:
  61.             self.assertEquals(e.args[0], errno.EINPROGRESS)
  62.         else:
  63.             raise AssertionError("Connect should have raised EINPROGRESS")
  64.         server, addr = self.serverSocket.accept()
  65.  
  66.         self.connections.extend((client, server))
  67.         return client, server
  68.  
  69.     def test_create(self):
  70.         try:
  71.             ep = select.epoll(16)
  72.         except OSError, e:
  73.             raise AssertionError(str(e))
  74.         self.assert_(ep.fileno() > 0, ep.fileno())
  75.         self.assert_(not ep.closed)
  76.         ep.close()
  77.         self.assert_(ep.closed)
  78.         self.assertRaises(ValueError, ep.fileno)
  79.  
  80.     def test_badcreate(self):
  81.         self.assertRaises(TypeError, select.epoll, 1, 2, 3)
  82.         self.assertRaises(TypeError, select.epoll, 'foo')
  83.         self.assertRaises(TypeError, select.epoll, None)
  84.         self.assertRaises(TypeError, select.epoll, ())
  85.         self.assertRaises(TypeError, select.epoll, ['foo'])
  86.         self.assertRaises(TypeError, select.epoll, {})
  87.  
  88.     def test_add(self):
  89.         server, client = self._connected_pair()
  90.  
  91.         ep = select.epoll(2)
  92.         try:
  93.             ep.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
  94.             ep.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
  95.         finally:
  96.             ep.close()
  97.  
  98.     def test_fromfd(self):
  99.         server, client = self._connected_pair()
  100.  
  101.         ep = select.epoll(2)
  102.         ep2 = select.epoll.fromfd(ep.fileno())
  103.  
  104.         ep2.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
  105.         ep2.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
  106.  
  107.         events = ep.poll(1, 4)
  108.         events2 = ep2.poll(0.9, 4)
  109.         self.assertEqual(len(events), 2)
  110.         self.assertEqual(len(events2), 2)
  111.  
  112.         ep.close()
  113.         try:
  114.             ep2.poll(1, 4)
  115.         except IOError, e:
  116.             self.failUnlessEqual(e.args[0], errno.EBADF, e)
  117.         else:
  118.             self.fail("epoll on closed fd didn't raise EBADF")
  119.  
  120.     def test_control_and_wait(self):
  121.         client, server = self._connected_pair()
  122.  
  123.         ep = select.epoll(16)
  124.         ep.register(server.fileno(),
  125.                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
  126.         ep.register(client.fileno(),
  127.                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
  128.  
  129.         now = time.time()
  130.         events = ep.poll(1, 4)
  131.         then = time.time()
  132.         self.failIf(then - now > 0.1, then - now)
  133.  
  134.         events.sort()
  135.         expected = [(client.fileno(), select.EPOLLOUT),
  136.                     (server.fileno(), select.EPOLLOUT)]
  137.         expected.sort()
  138.  
  139.         self.assertEquals(events, expected)
  140.         self.failIf(then - now > 0.01, then - now)
  141.  
  142.         now = time.time()
  143.         events = ep.poll(timeout=2.1, maxevents=4)
  144.         then = time.time()
  145.         self.failIf(events)
  146.  
  147.         client.send("Hello!")
  148.         server.send("world!!!")
  149.  
  150.         now = time.time()
  151.         events = ep.poll(1, 4)
  152.         then = time.time()
  153.         self.failIf(then - now > 0.01)
  154.  
  155.         events.sort()
  156.         expected = [(client.fileno(), select.EPOLLIN | select.EPOLLOUT),
  157.                     (server.fileno(), select.EPOLLIN | select.EPOLLOUT)]
  158.         expected.sort()
  159.  
  160.         self.assertEquals(events, expected)
  161.  
  162.         ep.unregister(client.fileno())
  163.         ep.modify(server.fileno(), select.EPOLLOUT)
  164.         now = time.time()
  165.         events = ep.poll(1, 4)
  166.         then = time.time()
  167.         self.failIf(then - now > 0.01)
  168.  
  169.         expected = [(server.fileno(), select.EPOLLOUT)]
  170.         self.assertEquals(events, expected)
  171.  
  172.     def test_errors(self):
  173.         self.assertRaises(ValueError, select.epoll, -2)
  174.         self.assertRaises(ValueError, select.epoll().register, -1,
  175.                           select.EPOLLIN)
  176.  
  177.     def test_unregister_closed(self):
  178.         server, client = self._connected_pair()
  179.         fd = server.fileno()
  180.         ep = select.epoll(16)
  181.         ep.register(server)
  182.  
  183.         now = time.time()
  184.         events = ep.poll(1, 4)
  185.         then = time.time()
  186.         self.failIf(then - now > 0.01)
  187.  
  188.         server.close()
  189.         ep.unregister(fd)
  190.  
  191. def test_main():
  192.     test_support.run_unittest(TestEPoll)
  193.  
  194. if __name__ == "__main__":
  195.     test_main()
  196.