| Index: third_party/twisted_8_1/twisted/test/test_fdesc.py
|
| diff --git a/third_party/twisted_8_1/twisted/test/test_fdesc.py b/third_party/twisted_8_1/twisted/test/test_fdesc.py
|
| deleted file mode 100644
|
| index a88803cf5cd2e4082591a61d5efd0cedeae779b8..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/test/test_fdesc.py
|
| +++ /dev/null
|
| @@ -1,176 +0,0 @@
|
| -# Copyright (c) 2007 Twisted Matrix Laboratories.
|
| -# See LICENSE for details.
|
| -
|
| -"""
|
| -Tests for L{twisted.internet.fdesc}.
|
| -"""
|
| -
|
| -import os
|
| -import errno
|
| -
|
| -from twisted.trial import unittest
|
| -
|
| -try:
|
| - from twisted.internet import fdesc
|
| -except ImportError:
|
| - fdesc = None
|
| -
|
| -
|
| -class ReadWriteTestCase(unittest.TestCase):
|
| - """
|
| - Tests for fdesc.readFromFD, fdesc.writeToFD.
|
| - """
|
| -
|
| - def setUp(self):
|
| - """
|
| - Create two non-blocking pipes that can be used in tests.
|
| - """
|
| - self.r, self.w = os.pipe()
|
| - fdesc.setNonBlocking(self.r)
|
| - fdesc.setNonBlocking(self.w)
|
| -
|
| -
|
| - def tearDown(self):
|
| - """
|
| - Close pipes.
|
| - """
|
| - try:
|
| - os.close(self.w)
|
| - except OSError:
|
| - pass
|
| - try:
|
| - os.close(self.r)
|
| - except OSError:
|
| - pass
|
| -
|
| -
|
| - def write(self, d):
|
| - """
|
| - Write data to the pipe.
|
| - """
|
| - return fdesc.writeToFD(self.w, d)
|
| -
|
| -
|
| - def read(self):
|
| - """
|
| - Read data from the pipe.
|
| - """
|
| - l = []
|
| - res = fdesc.readFromFD(self.r, l.append)
|
| - if res is None:
|
| - if l:
|
| - return l[0]
|
| - else:
|
| - return ""
|
| - else:
|
| - return res
|
| -
|
| -
|
| - def test_writeAndRead(self):
|
| - """
|
| - Test that the number of bytes L{fdesc.writeToFD} reports as written
|
| - with its return value are seen by L{fdesc.readFromFD}.
|
| - """
|
| - n = self.write("hello")
|
| - self.failUnless(n > 0)
|
| - s = self.read()
|
| - self.assertEquals(len(s), n)
|
| - self.assertEquals("hello"[:n], s)
|
| -
|
| -
|
| - def test_writeAndReadLarge(self):
|
| - """
|
| - Similar to L{test_writeAndRead}, but use a much larger string to verify
|
| - the behavior for that case.
|
| - """
|
| - orig = "0123456879" * 10000
|
| - written = self.write(orig)
|
| - self.failUnless(written > 0)
|
| - result = []
|
| - resultlength = 0
|
| - i = 0
|
| - while resultlength < written or i < 50:
|
| - result.append(self.read())
|
| - resultlength += len(result[-1])
|
| - # Increment a counter to be sure we'll exit at some point
|
| - i += 1
|
| - result = "".join(result)
|
| - self.assertEquals(len(result), written)
|
| - self.assertEquals(orig[:written], result)
|
| -
|
| -
|
| - def test_readFromEmpty(self):
|
| - """
|
| - Verify that reading from a file descriptor with no data does not raise
|
| - an exception and does not result in the callback function being called.
|
| - """
|
| - l = []
|
| - result = fdesc.readFromFD(self.r, l.append)
|
| - self.assertEquals(l, [])
|
| - self.assertEquals(result, None)
|
| -
|
| -
|
| - def test_readFromCleanClose(self):
|
| - """
|
| - Test that using L{fdesc.readFromFD} on a cleanly closed file descriptor
|
| - returns a connection done indicator.
|
| - """
|
| - os.close(self.w)
|
| - self.assertEquals(self.read(), fdesc.CONNECTION_DONE)
|
| -
|
| -
|
| - def test_writeToClosed(self):
|
| - """
|
| - Verify that writing with L{fdesc.writeToFD} when the read end is closed
|
| - results in a connection lost indicator.
|
| - """
|
| - os.close(self.r)
|
| - self.assertEquals(self.write("s"), fdesc.CONNECTION_LOST)
|
| -
|
| -
|
| - def test_readFromInvalid(self):
|
| - """
|
| - Verify that reading with L{fdesc.readFromFD} when the read end is
|
| - closed results in a connection lost indicator.
|
| - """
|
| - os.close(self.r)
|
| - self.assertEquals(self.read(), fdesc.CONNECTION_LOST)
|
| -
|
| -
|
| - def test_writeToInvalid(self):
|
| - """
|
| - Verify that writing with L{fdesc.writeToFD} when the write end is
|
| - closed results in a connection lost indicator.
|
| - """
|
| - os.close(self.w)
|
| - self.assertEquals(self.write("s"), fdesc.CONNECTION_LOST)
|
| -
|
| -
|
| - def test_writeErrors(self):
|
| - """
|
| - Test error path for L{fdesc.writeTod}.
|
| - """
|
| - oldOsWrite = os.write
|
| - def eagainWrite(fd, data):
|
| - err = OSError()
|
| - err.errno = errno.EAGAIN
|
| - raise err
|
| - os.write = eagainWrite
|
| - try:
|
| - self.assertEquals(self.write("s"), 0)
|
| - finally:
|
| - os.write = oldOsWrite
|
| -
|
| - def eintrWrite(fd, data):
|
| - err = OSError()
|
| - err.errno = errno.EINTR
|
| - raise err
|
| - os.write = eintrWrite
|
| - try:
|
| - self.assertEquals(self.write("s"), 0)
|
| - finally:
|
| - os.write = oldOsWrite
|
| -
|
| -
|
| -if fdesc is None:
|
| - ReadWriteTestCase.skip = "not supported on this platform"
|
|
|