Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(161)

Unified Diff: third_party/twisted_8_1/twisted/test/test_epoll.py

Issue 12261012: Remove third_party/twisted_8_1 (Closed) Base URL: svn://svn.chromium.org/chrome/trunk/tools/build
Patch Set: Created 7 years, 10 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: third_party/twisted_8_1/twisted/test/test_epoll.py
diff --git a/third_party/twisted_8_1/twisted/test/test_epoll.py b/third_party/twisted_8_1/twisted/test/test_epoll.py
deleted file mode 100644
index a5912a7b20dac72037092bc3c7fcfaba1fefe097..0000000000000000000000000000000000000000
--- a/third_party/twisted_8_1/twisted/test/test_epoll.py
+++ /dev/null
@@ -1,159 +0,0 @@
-# Copyright (c) 2001-2006 Twisted Matrix Laboratories.
-# See LICENSE for details.
-
-"""
-Tests for epoll wrapper.
-"""
-
-import socket, errno, time
-
-from twisted.trial import unittest
-from twisted.python.util import untilConcludes
-
-try:
- from twisted.python import _epoll
-except ImportError:
- _epoll = None
-
-
-class EPoll(unittest.TestCase):
- """
- Tests for the low-level epoll bindings.
- """
- def setUp(self):
- """
- Create a listening server port and a list with which to keep track
- of created sockets.
- """
- self.serverSocket = socket.socket()
- self.serverSocket.bind(('127.0.0.1', 0))
- self.serverSocket.listen(1)
- self.connections = [self.serverSocket]
-
-
- def tearDown(self):
- """
- Close any sockets which were opened by the test.
- """
- for skt in self.connections:
- skt.close()
-
-
- def _connectedPair(self):
- """
- Return the two sockets which make up a new TCP connection.
- """
- client = socket.socket()
- client.setblocking(False)
- try:
- client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
- except socket.error, e:
- self.assertEquals(e.args[0], errno.EINPROGRESS)
- else:
- raise unittest.FailTest("Connect should have raised EINPROGRESS")
- server, addr = self.serverSocket.accept()
-
- self.connections.extend((client, server))
- return client, server
-
-
- def test_create(self):
- """
- Test the creation of an epoll object.
- """
- try:
- p = _epoll.epoll(16)
- except OSError, e:
- raise unittest.FailTest(str(e))
- else:
- p.close()
-
-
- def test_badCreate(self):
- """
- Test that attempting to create an epoll object with some random
- objects raises a TypeError.
- """
- self.assertRaises(TypeError, _epoll.epoll, 1, 2, 3)
- self.assertRaises(TypeError, _epoll.epoll, 'foo')
- self.assertRaises(TypeError, _epoll.epoll, None)
- self.assertRaises(TypeError, _epoll.epoll, ())
- self.assertRaises(TypeError, _epoll.epoll, ['foo'])
- self.assertRaises(TypeError, _epoll.epoll, {})
- self.assertRaises(TypeError, _epoll.epoll)
-
-
- def test_add(self):
- """
- Test adding a socket to an epoll object.
- """
- server, client = self._connectedPair()
-
- p = _epoll.epoll(2)
- try:
- p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT)
- p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT)
- finally:
- p.close()
-
-
- def test_controlAndWait(self):
- """
- Test waiting on an epoll object which has had some sockets added to
- it.
- """
- client, server = self._connectedPair()
-
- p = _epoll.epoll(16)
- p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT |
- _epoll.ET)
- p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT |
- _epoll.ET)
-
- now = time.time()
- events = untilConcludes(p.wait, 4, 1000)
- then = time.time()
- self.failIf(then - now > 0.01)
-
- events.sort()
- expected = [(client.fileno(), _epoll.OUT),
- (server.fileno(), _epoll.OUT)]
- expected.sort()
-
- self.assertEquals(events, expected)
-
- now = time.time()
- events = untilConcludes(p.wait, 4, 200)
- then = time.time()
- self.failUnless(then - now > 0.1)
- self.failIf(events)
-
- client.send("Hello!")
- server.send("world!!!")
-
- now = time.time()
- events = untilConcludes(p.wait, 4, 1000)
- then = time.time()
- self.failIf(then - now > 0.01)
-
- events.sort()
- expected = [(client.fileno(), _epoll.IN | _epoll.OUT),
- (server.fileno(), _epoll.IN | _epoll.OUT)]
- expected.sort()
-
- self.assertEquals(events, expected)
-
-if _epoll is None:
- EPoll.skip = "_epoll module unavailable"
-else:
- try:
- e = _epoll.epoll(16)
- except IOError, exc:
- if exc.errno == errno.ENOSYS:
- del exc
- EPoll.skip = "epoll support missing from platform"
- else:
- raise
- else:
- e.close()
- del e
« no previous file with comments | « third_party/twisted_8_1/twisted/test/test_enterprise.py ('k') | third_party/twisted_8_1/twisted/test/test_error.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698