def send(self, *args):
return os.write(self.fd, *args)
+ def getsockopt(self, level, optname, buflen=None):
+ if (level == socket.SOL_SOCKET and
+ optname == socket.SO_ERROR and
+ not buflen):
+ return 0
+ raise NotImplementedError("Only asyncore specific behaviour "
+ "implemented.")
+
read = recv
write = send
w.close()
self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
+ @unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
+ 'asyncore.file_dispatcher required')
+ def test_dispatcher(self):
+ fd = os.open(TESTFN, os.O_RDONLY)
+ data = []
+ class FileDispatcher(asyncore.file_dispatcher):
+ def handle_read(self):
+ data.append(self.recv(29))
+ s = FileDispatcher(fd)
+ os.close(fd)
+ asyncore.loop(timeout=0.01, use_poll=True, count=2)
+ self.assertEqual(b"".join(data), self.d)
+
class BaseTestHandler(asyncore.dispatcher):
Library
-------
+- Issue #9354: Provide getsockopt() in asyncore's file_wrapper.
+
- Issue #4108: In urllib.robotparser, if there are multiple 'User-agent: *'
entries, consider the first one.
Library
-------
+
- Issue #1555570: email no longer inserts extra blank lines when a \r\n
combo crosses an 8192 byte boundary.