Issue #8524: Add a forget() method to socket objects, so as to put the

socket into the closed state without closing the underlying file
descriptor.
This commit is contained in:
Antoine Pitrou 2010-08-08 23:24:50 +00:00
parent ba8a98600e
commit e43f9d0ed6
6 changed files with 50 additions and 3 deletions

View file

@ -79,7 +79,6 @@ from _ssl import (
from socket import getnameinfo as _getnameinfo
from socket import error as socket_error
from socket import dup as _dup
from socket import socket, AF_INET, SOCK_STREAM
import base64 # for DER-to-PEM translation
import traceback
@ -148,7 +147,7 @@ class SSLSocket(socket):
family=sock.family,
type=sock.type,
proto=sock.proto,
fileno=_dup(sock.fileno()))
fileno=sock.fileno())
self.settimeout(sock.gettimeout())
# see if it's connected
try:
@ -158,7 +157,7 @@ class SSLSocket(socket):
raise
else:
connected = True
sock.close()
sock.forget()
elif fileno is not None:
socket.__init__(self, fileno=fileno)
else:

View file

@ -655,6 +655,19 @@ class BasicTCPTest(SocketConnectedTest):
self.serv_conn.send(MSG)
self.serv_conn.shutdown(2)
def testForget(self):
# Testing forget()
f = self.cli_conn.fileno()
self.cli_conn.forget()
self.assertRaises(socket.error, self.cli_conn.recv, 1024)
self.cli_conn.close()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=f)
msg = sock.recv(1024)
self.assertEqual(msg, MSG)
def _testForget(self):
self.serv_conn.send(MSG)
@unittest.skipUnless(thread, 'Threading required for this test.')
class BasicUDPTest(ThreadedUDPSocketTest):