Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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""
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)
def wrap_bio(self, incoming, outcoming, *args, **kwargs):
ssl_obj = MocketSocket()
ssl_obj._host = kwargs["server_hostname"]
return ssl_obj
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
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
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(
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