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.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()