我们从Python开源项目中，提取了以下9个代码示例，用于说明如何使用socket._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.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()
