def test_close(self):
import httplib
# calling .close() on urllib2's response objects should close the
# underlying socket
# delve deep into response to fetch socket._socketobject
response = _urlopen_with_retry("http://www.example.com/")
abused_fileobject = response.fp
self.assertIs(abused_fileobject.__class__, socket._fileobject)
httpresponse = abused_fileobject._sock
self.assertIs(httpresponse.__class__, httplib.HTTPResponse)
fileobject = httpresponse.fp
self.assertIs(fileobject.__class__, socket._fileobject)
self.assertTrue(not fileobject.closed)
response.close()
self.assertTrue(fileobject.closed)
python类_socketobject()的实例源码
def test_close(self):
import httplib
# calling .close() on urllib2's response objects should close the
# underlying socket
# delve deep into response to fetch socket._socketobject
response = _urlopen_with_retry("http://www.example.com/")
abused_fileobject = response.fp
self.assertIs(abused_fileobject.__class__, socket._fileobject)
httpresponse = abused_fileobject._sock
self.assertIs(httpresponse.__class__, httplib.HTTPResponse)
fileobject = httpresponse.fp
self.assertIs(fileobject.__class__, socket._fileobject)
self.assertTrue(not fileobject.closed)
response.close()
self.assertTrue(fileobject.closed)
def test_close(self):
import httplib
# calling .close() on urllib2's response objects should close the
# underlying socket
# delve deep into response to fetch socket._socketobject
response = _urlopen_with_retry("http://www.example.com/")
abused_fileobject = response.fp
self.assertIs(abused_fileobject.__class__, socket._fileobject)
httpresponse = abused_fileobject._sock
self.assertIs(httpresponse.__class__, httplib.HTTPResponse)
fileobject = httpresponse.fp
self.assertIs(fileobject.__class__, socket._fileobject)
self.assertTrue(not fileobject.closed)
response.close()
self.assertTrue(fileobject.closed)
def test_close(self):
import httplib
# calling .close() on urllib2's response objects should close the
# underlying socket
# delve deep into response to fetch socket._socketobject
response = _urlopen_with_retry("http://www.python.org/")
abused_fileobject = response.fp
self.assertTrue(abused_fileobject.__class__ is socket._fileobject)
httpresponse = abused_fileobject._sock
self.assertTrue(httpresponse.__class__ is httplib.HTTPResponse)
fileobject = httpresponse.fp
self.assertTrue(fileobject.__class__ is socket._fileobject)
self.assertTrue(not fileobject.closed)
response.close()
self.assertTrue(fileobject.closed)
def _finishInit(self, whenDone, skt, error, reactor):
"""
Called by subclasses to continue to the stage of initialization where
the socket connect attempt is made.
@param whenDone: A 0-argument callable to invoke once the connection is
set up. This is L{None} if the connection could not be prepared
due to a previous error.
@param skt: The socket object to use to perform the connection.
@type skt: C{socket._socketobject}
@param error: The error to fail the connection with.
@param reactor: The reactor to use for this client.
@type reactor: L{twisted.internet.interfaces.IReactorTime}
"""
if whenDone:
self._commonConnection.__init__(self, skt, None, reactor)
reactor.callLater(0, whenDone)
else:
reactor.callLater(0, self.failIfNotConnected, error)
def patch_socket():
socket.socket = MockSocket
socket.socket = socket.__dict__['socket'] = MockSocket
socket._socketobject = socket.__dict__['_socketobject'] = MockSocket
socket.SocketType = socket.__dict__['SocketType'] = MockSocket
socket.create_connection = socket.__dict__['create_connection'] = create_connection
socket.getaddrinfo = socket.__dict__['getaddrinfo'] = getaddrinfo
socket.gethostname = socket.__dict__['gethostname'] = lambda: 'localhost'
socket.gethostbyname = socket.__dict__['gethostbyname'] = lambda host: '127.0.0.1'
socket.inet_aton = socket.__dict__['inet_aton'] = lambda host: '127.0.0.1'
def disable(cls):
cls._is_enabled = False
socket.socket = old_socket
socket.SocketType = old_SocketType
socket._socketobject = old_socket
socket.create_connection = old_create_connection
socket.gethostname = old_gethostname
socket.gethostbyname = old_gethostbyname
socket.getaddrinfo = old_getaddrinfo
socket.__dict__['socket'] = old_socket
socket.__dict__['_socketobject'] = old_socket
socket.__dict__['SocketType'] = old_SocketType
socket.__dict__['create_connection'] = old_create_connection
socket.__dict__['gethostname'] = old_gethostname
socket.__dict__['gethostbyname'] = old_gethostbyname
socket.__dict__['getaddrinfo'] = old_getaddrinfo
if socks:
socks.socksocket = old_socksocket
socks.__dict__['socksocket'] = old_socksocket
if ssl:
ssl.wrap_socket = old_ssl_wrap_socket
ssl.SSLSocket = old_sslsocket
ssl.__dict__['wrap_socket'] = old_ssl_wrap_socket
ssl.__dict__['SSLSocket'] = old_sslsocket
if not PY3:
ssl.sslwrap_simple = old_sslwrap_simple
ssl.__dict__['sslwrap_simple'] = old_sslwrap_simple
if pyopenssl_override:
# Replace PyOpenSSL Monkeypatching
inject_into_urllib3()
def fake_socket(tmpdir_factory, request):
packets = request.node.callspec.params.get('packets')
# write data to the "socket"; this must be an actual file, because
# select.select()() expects a real file descriptor
filename = str(tmpdir_factory.mktemp('fake-socket').join(str(uuid4())))
f = open(filename, 'w')
request.addfinalizer(f.close)
sockobj = socket._socket.socket if six.PY3 else socket._socketobject
class fakesocket(sockobj):
buff = six.BytesIO()
def _send(self, data):
self.buff.write(data)
return len(data) # number of bytes sent
sock = fakesocket()
sock.f = six.BytesIO(packets)
# socket.socket overrides these at instantiation time to the underlying
# C implementation; set them here so that we can mock send() and recv()
# calls
sock.send = types.MethodType(_send, sock)
sock.recv = types.MethodType(
lambda self, _bytes: self.f.read(_bytes),
sock
)
sock.fileno = types.MethodType(lambda self: f.fileno(), sock)
return sock
def enable(cls):
cls._is_enabled = True
# Some versions of python internally shadowed the
# SocketType variable incorrectly https://bugs.python.org/issue20386
bad_socket_shadow = (socket.socket != socket.SocketType)
socket.socket = fakesock.socket
socket._socketobject = fakesock.socket
if not bad_socket_shadow:
socket.SocketType = fakesock.socket
socket.create_connection = create_fake_connection
socket.gethostname = fake_gethostname
socket.gethostbyname = fake_gethostbyname
socket.getaddrinfo = fake_getaddrinfo
socket.__dict__['socket'] = fakesock.socket
socket.__dict__['_socketobject'] = fakesock.socket
if not bad_socket_shadow:
socket.__dict__['SocketType'] = fakesock.socket
socket.__dict__['create_connection'] = create_fake_connection
socket.__dict__['gethostname'] = fake_gethostname
socket.__dict__['gethostbyname'] = fake_gethostbyname
socket.__dict__['getaddrinfo'] = fake_getaddrinfo
if socks:
socks.socksocket = fakesock.socket
socks.__dict__['socksocket'] = fakesock.socket
if ssl:
ssl.wrap_socket = fake_wrap_socket
ssl.SSLSocket = FakeSSLSocket
ssl.__dict__['wrap_socket'] = fake_wrap_socket
ssl.__dict__['SSLSocket'] = FakeSSLSocket
if not PY3:
ssl.sslwrap_simple = fake_wrap_socket
ssl.__dict__['sslwrap_simple'] = fake_wrap_socket
if pyopenssl_override:
# Remove PyOpenSSL monkeypatch - use the default implementation
extract_from_urllib3()