How to use the mocket.mocket.MocketSocket function in mocket

To help you get started, we’ve selected a few mocket examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github mindflayer / python-mocket / mocket / mocket.py View on Github external
try:
            try:
                encoded_response = hexdump.dehex(response_dict["response"])
            except TypeError:  # pragma: no cover
                # Python 2
                encoded_response = hexdump.restore(
                    encode_to_bytes(response_dict["response"])
                )
        # if not available, call the real sendall
        except KeyError:
            host, port = Mocket._address
            host = true_gethostbyname(host)

            if isinstance(self.true_socket, true_socket) and self._secure_socket:
                try:
                    self = MocketSocket(sock=self)
                except TypeError:
                    ssl_context = self.kwargs.get("ssl_context")
                    server_hostname = self.kwargs.get("server_hostname")
                    self.true_socket = true_ssl_context.wrap_socket(
                        self=ssl_context,
                        sock=self.true_socket,
                        server_hostname=server_hostname,
                    )

            try:
                self.true_socket.connect((host, port))
            except (OSError, socket.error, ValueError):
                # already connected
                pass
            self.true_socket.sendall(data, *args, **kwargs)
            encoded_response = b""
github mindflayer / python-mocket / mocket / mocket.py View on Github external
def __init__(self, sock=None, server_hostname=None, _context=None, *args, **kwargs):
        if isinstance(sock, MocketSocket):
            self.sock = sock
            self.sock._host = server_hostname
            if true_ssl_context:
                self.sock.true_socket = true_ssl_socket(
                    sock=self.sock.true_socket,
                    server_hostname=server_hostname,
                    _context=true_ssl_context(protocol=SSL_PROTOCOL),
                )
            else:  # Python 2.
                self.sock.true_socket = true_ssl_socket(sock=self.sock.true_socket)
        elif isinstance(sock, int) and true_ssl_context:
            self.context = true_ssl_context(sock)
github mindflayer / python-mocket / mocket / mocket.py View on Github external
def wrap_bio(self, incoming, outcoming, *args, **kwargs):
        ssl_obj = MocketSocket()
        ssl_obj._host = kwargs["server_hostname"]
        return ssl_obj
github lyft / amundsenmetadatalibrary / metadata_service / proxy / aws4authwebsocket / transport.py View on Github external
def monkey_patch_mocket() -> None:
    # monkey patch this method since the existing behavior is really broken
    T = TypeVar('T', bound=mocket.mocket.MocketSocket)

    def monkey_fileno(self: T) -> int:
        if mocket.mocket.Mocket.r_fd is None and mocket.mocket.Mocket.w_fd is None:
            mocket.mocket.Mocket.r_fd, mocket.mocket.Mocket.w_fd = os.pipe()
        return mocket.mocket.Mocket.r_fd

    mocket.mocket.MocketSocket.fileno = monkey_fileno
github lyft / amundsenmetadatalibrary / metadata_service / proxy / aws4authwebsocket / transport.py View on Github external
def monkey_patch_mocket() -> None:
    # monkey patch this method since the existing behavior is really broken
    T = TypeVar('T', bound=mocket.mocket.MocketSocket)

    def monkey_fileno(self: T) -> int:
        if mocket.mocket.Mocket.r_fd is None and mocket.mocket.Mocket.w_fd is None:
            mocket.mocket.Mocket.r_fd, mocket.mocket.Mocket.w_fd = os.pipe()
        return mocket.mocket.Mocket.r_fd

    mocket.mocket.MocketSocket.fileno = monkey_fileno
github mindflayer / python-mocket / mocket / mocket.py View on Github external
def enable(namespace=None, truesocket_recording_dir=None):
        Mocket._namespace = namespace
        Mocket._truesocket_recording_dir = truesocket_recording_dir

        if truesocket_recording_dir:
            # JSON dumps will be saved here
            assert os.path.isdir(truesocket_recording_dir)

        socket.socket = socket.__dict__["socket"] = MocketSocket
        socket._socketobject = socket.__dict__["_socketobject"] = MocketSocket
        socket.SocketType = socket.__dict__["SocketType"] = MocketSocket
        socket.create_connection = socket.__dict__[
            "create_connection"
        ] = create_connection
        socket.gethostname = socket.__dict__["gethostname"] = lambda: "localhost"
        socket.gethostbyname = socket.__dict__[
            "gethostbyname"
        ] = lambda host: "127.0.0.1"
        socket.getaddrinfo = socket.__dict__[
            "getaddrinfo"
        ] = lambda host, port, family=None, socktype=None, proto=None, flags=None: [
            (2, 1, 6, "", (host, port))
        ]
        ssl.wrap_socket = ssl.__dict__["wrap_socket"] = FakeSSLContext.wrap_socket
        # ssl.SSLSocket = ssl.__dict__["SSLSocket"] = MocketSocket
        ssl.SSLContext = ssl.__dict__["SSLContext"] = FakeSSLContext
        socket.inet_pton = socket.__dict__["inet_pton"] = lambda family, ip: byte_type(
github mindflayer / python-mocket / mocket / mocket.py View on Github external
def enable(namespace=None, truesocket_recording_dir=None):
        Mocket._namespace = namespace
        Mocket._truesocket_recording_dir = truesocket_recording_dir

        if truesocket_recording_dir:
            # JSON dumps will be saved here
            assert os.path.isdir(truesocket_recording_dir)

        socket.socket = socket.__dict__["socket"] = MocketSocket
        socket._socketobject = socket.__dict__["_socketobject"] = MocketSocket
        socket.SocketType = socket.__dict__["SocketType"] = MocketSocket
        socket.create_connection = socket.__dict__[
            "create_connection"
        ] = create_connection
        socket.gethostname = socket.__dict__["gethostname"] = lambda: "localhost"
        socket.gethostbyname = socket.__dict__[
            "gethostbyname"
        ] = lambda host: "127.0.0.1"
        socket.getaddrinfo = socket.__dict__[
            "getaddrinfo"
        ] = lambda host, port, family=None, socktype=None, proto=None, flags=None: [
            (2, 1, 6, "", (host, port))
        ]
        ssl.wrap_socket = ssl.__dict__["wrap_socket"] = FakeSSLContext.wrap_socket
        # ssl.SSLSocket = ssl.__dict__["SSLSocket"] = MocketSocket