11881415972

Committed 17 Nov 2024 06:59PM UTC coverage: 99.324% (-0.2%) from 99.545%

Build # 11881415972

Build Type

Pull #264

github

Committed by

web-flow

Commit Message

Merge 6291c31b8 into 2beb8f151

Pull Request Pull Request #264: Refactor: introduce state object

Run Details

104 of 107 new or added lines in 9 files covered. (97.2%)

4 existing lines in 1 file now uncovered.

881 of 887 relevant lines covered (99.32%)

6.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.48

/mocket/inject.py

from __future__ import annotations

import os
import socket
import ssl

import urllib3
from urllib3.connection import match_hostname as urllib3_match_hostname
from urllib3.util.ssl_ import ssl_wrap_socket as urllib3_ssl_wrap_socket

try:
    from urllib3.util.ssl_ import wrap_socket as urllib3_wrap_socket
except ImportError:
    urllib3_wrap_socket = None


try:  # pragma: no cover
    from urllib3.contrib.pyopenssl import extract_from_urllib3, inject_into_urllib3

    pyopenssl_override = True
except ImportError:
    pyopenssl_override = False

true_socket = socket.socket
true_create_connection = socket.create_connection
true_gethostbyname = socket.gethostbyname
true_gethostname = socket.gethostname
true_getaddrinfo = socket.getaddrinfo
true_socketpair = socket.socketpair
true_ssl_wrap_socket = getattr(
    ssl, "wrap_socket", None
)  # from Py3.12 it's only under SSLContext
true_ssl_socket = ssl.SSLSocket
true_ssl_context = ssl.SSLContext
true_inet_pton = socket.inet_pton
true_urllib3_wrap_socket = urllib3_wrap_socket
true_urllib3_ssl_wrap_socket = urllib3_ssl_wrap_socket
true_urllib3_match_hostname = urllib3_match_hostname


def enable(
    namespace: str | None = None,
    truesocket_recording_dir: str | None = None,
) -> None:
    from mocket.mocket import Mocket
    from mocket.socket import MocketSocket, create_connection, socketpair
    from mocket.ssl import FakeSSLContext

    Mocket._namespace = namespace
    Mocket._truesocket_recording_dir = truesocket_recording_dir

    if truesocket_recording_dir and not os.path.isdir(truesocket_recording_dir):
        # JSON dumps will be saved here
        raise AssertionError

    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))
        ]
    )
    socket.socketpair = socket.__dict__["socketpair"] = socketpair
    ssl.wrap_socket = ssl.__dict__["wrap_socket"] = FakeSSLContext.wrap_socket
    ssl.SSLContext = ssl.__dict__["SSLContext"] = FakeSSLContext
    socket.inet_pton = socket.__dict__["inet_pton"] = lambda family, ip: bytes(
        "\x7f\x00\x00\x01", "utf-8"
    )
    urllib3.util.ssl_.wrap_socket = urllib3.util.ssl_.__dict__["wrap_socket"] = (
        FakeSSLContext.wrap_socket
    )
    urllib3.util.ssl_.ssl_wrap_socket = urllib3.util.ssl_.__dict__[
        "ssl_wrap_socket"
    ] = FakeSSLContext.wrap_socket
    urllib3.util.ssl_wrap_socket = urllib3.util.__dict__["ssl_wrap_socket"] = (
        FakeSSLContext.wrap_socket
    )
    urllib3.connection.ssl_wrap_socket = urllib3.connection.__dict__[
        "ssl_wrap_socket"
    ] = FakeSSLContext.wrap_socket
    urllib3.connection.match_hostname = urllib3.connection.__dict__[
        "match_hostname"
    ] = lambda *args: None
    if pyopenssl_override:  # pragma: no cover
        # Take out the pyopenssl version - use the default implementation
        extract_from_urllib3()


def disable() -> None:
    from mocket.mocket import Mocket

    socket.socket = socket.__dict__["socket"] = true_socket
    socket._socketobject = socket.__dict__["_socketobject"] = true_socket
    socket.SocketType = socket.__dict__["SocketType"] = true_socket
    socket.create_connection = socket.__dict__["create_connection"] = (
        true_create_connection
    )
    socket.gethostname = socket.__dict__["gethostname"] = true_gethostname
    socket.gethostbyname = socket.__dict__["gethostbyname"] = true_gethostbyname
    socket.getaddrinfo = socket.__dict__["getaddrinfo"] = true_getaddrinfo
    socket.socketpair = socket.__dict__["socketpair"] = true_socketpair
    if true_ssl_wrap_socket:
        ssl.wrap_socket = ssl.__dict__["wrap_socket"] = true_ssl_wrap_socket
    ssl.SSLContext = ssl.__dict__["SSLContext"] = true_ssl_context
    socket.inet_pton = socket.__dict__["inet_pton"] = true_inet_pton
    urllib3.util.ssl_.wrap_socket = urllib3.util.ssl_.__dict__["wrap_socket"] = (
        true_urllib3_wrap_socket
    )
    urllib3.util.ssl_.ssl_wrap_socket = urllib3.util.ssl_.__dict__[
        "ssl_wrap_socket"
    ] = true_urllib3_ssl_wrap_socket
    urllib3.util.ssl_wrap_socket = urllib3.util.__dict__["ssl_wrap_socket"] = (
        true_urllib3_ssl_wrap_socket
    )
    urllib3.connection.ssl_wrap_socket = urllib3.connection.__dict__[
        "ssl_wrap_socket"
    ] = true_urllib3_ssl_wrap_socket
    urllib3.connection.match_hostname = urllib3.connection.__dict__[
        "match_hostname"
    ] = true_urllib3_match_hostname
    Mocket.reset()
    if pyopenssl_override:  # pragma: no cover
        # Put the pyopenssl version back in place
        inject_into_urllib3()

1	from __future__ import annotations	7✔
2
3	import os	7✔
4	import socket	7✔
5	import ssl	7✔
6
7	import urllib3	7✔
8	from urllib3.connection import match_hostname as urllib3_match_hostname	7✔
9	from urllib3.util.ssl_ import ssl_wrap_socket as urllib3_ssl_wrap_socket	7✔
10
11	try:	7✔
12	from urllib3.util.ssl_ import wrap_socket as urllib3_wrap_socket	7✔
13	except ImportError:
14	urllib3_wrap_socket = None
15
16
17	try: # pragma: no cover
18	from urllib3.contrib.pyopenssl import extract_from_urllib3, inject_into_urllib3
19
20	pyopenssl_override = True
21	except ImportError:
22	pyopenssl_override = False
23
24	true_socket = socket.socket	7✔
25	true_create_connection = socket.create_connection	7✔
26	true_gethostbyname = socket.gethostbyname	7✔
27	true_gethostname = socket.gethostname	7✔
28	true_getaddrinfo = socket.getaddrinfo	7✔
29	true_socketpair = socket.socketpair	7✔
30	true_ssl_wrap_socket = getattr(	7✔
31	ssl, "wrap_socket", None
32	) # from Py3.12 it's only under SSLContext
33	true_ssl_socket = ssl.SSLSocket	7✔
34	true_ssl_context = ssl.SSLContext	7✔
35	true_inet_pton = socket.inet_pton	7✔
36	true_urllib3_wrap_socket = urllib3_wrap_socket	7✔
37	true_urllib3_ssl_wrap_socket = urllib3_ssl_wrap_socket	7✔
38	true_urllib3_match_hostname = urllib3_match_hostname	7✔
39
40
41	def enable(	7✔
42	namespace: str \| None = None,
43	truesocket_recording_dir: str \| None = None,
44	) -> None:
45	from mocket.mocket import Mocket	7✔
46	from mocket.socket import MocketSocket, create_connection, socketpair	7✔
47	from mocket.ssl import FakeSSLContext	7✔
48
49	Mocket._namespace = namespace	7✔
50	Mocket._truesocket_recording_dir = truesocket_recording_dir	7✔
51
52	if truesocket_recording_dir and not os.path.isdir(truesocket_recording_dir):	7✔
53	# JSON dumps will be saved here
NEW 54	raise AssertionError	×
55
56	socket.socket = socket.__dict__["socket"] = MocketSocket	7✔
57	socket._socketobject = socket.__dict__["_socketobject"] = MocketSocket	7✔
58	socket.SocketType = socket.__dict__["SocketType"] = MocketSocket	7✔
59	socket.create_connection = socket.__dict__["create_connection"] = create_connection	7✔
60	socket.gethostname = socket.__dict__["gethostname"] = lambda: "localhost"	7✔
61	socket.gethostbyname = socket.__dict__["gethostbyname"] = lambda host: "127.0.0.1"	7✔
62	socket.getaddrinfo = socket.__dict__["getaddrinfo"] = (	7✔
63	lambda host, port, family=None, socktype=None, proto=None, flags=None: [
64	(2, 1, 6, "", (host, port))
65	]
66	)
67	socket.socketpair = socket.__dict__["socketpair"] = socketpair	7✔
68	ssl.wrap_socket = ssl.__dict__["wrap_socket"] = FakeSSLContext.wrap_socket	7✔
69	ssl.SSLContext = ssl.__dict__["SSLContext"] = FakeSSLContext	7✔
70	socket.inet_pton = socket.__dict__["inet_pton"] = lambda family, ip: bytes(	7✔
71	"\x7f\x00\x00\x01", "utf-8"
72	)
73	urllib3.util.ssl_.wrap_socket = urllib3.util.ssl_.__dict__["wrap_socket"] = (	7✔
74	FakeSSLContext.wrap_socket
75	)
76	urllib3.util.ssl_.ssl_wrap_socket = urllib3.util.ssl_.__dict__[	7✔
77	"ssl_wrap_socket"
78	] = FakeSSLContext.wrap_socket
79	urllib3.util.ssl_wrap_socket = urllib3.util.__dict__["ssl_wrap_socket"] = (	7✔
80	FakeSSLContext.wrap_socket
81	)
82	urllib3.connection.ssl_wrap_socket = urllib3.connection.__dict__[	7✔
83	"ssl_wrap_socket"
84	] = FakeSSLContext.wrap_socket
85	urllib3.connection.match_hostname = urllib3.connection.__dict__[	7✔
86	"match_hostname"
87	] = lambda *args: None
88	if pyopenssl_override: # pragma: no cover
89	# Take out the pyopenssl version - use the default implementation
90	extract_from_urllib3()
91
92
93	def disable() -> None:	7✔
94	from mocket.mocket import Mocket	7✔
95
96	socket.socket = socket.__dict__["socket"] = true_socket	7✔
97	socket._socketobject = socket.__dict__["_socketobject"] = true_socket	7✔
98	socket.SocketType = socket.__dict__["SocketType"] = true_socket	7✔
99	socket.create_connection = socket.__dict__["create_connection"] = (	7✔
100	true_create_connection
101	)
102	socket.gethostname = socket.__dict__["gethostname"] = true_gethostname	7✔
103	socket.gethostbyname = socket.__dict__["gethostbyname"] = true_gethostbyname	7✔
104	socket.getaddrinfo = socket.__dict__["getaddrinfo"] = true_getaddrinfo	7✔
105	socket.socketpair = socket.__dict__["socketpair"] = true_socketpair	7✔
106	if true_ssl_wrap_socket:	7✔
107	ssl.wrap_socket = ssl.__dict__["wrap_socket"] = true_ssl_wrap_socket	5✔
108	ssl.SSLContext = ssl.__dict__["SSLContext"] = true_ssl_context	7✔
109	socket.inet_pton = socket.__dict__["inet_pton"] = true_inet_pton	7✔
110	urllib3.util.ssl_.wrap_socket = urllib3.util.ssl_.__dict__["wrap_socket"] = (	7✔
111	true_urllib3_wrap_socket
112	)
113	urllib3.util.ssl_.ssl_wrap_socket = urllib3.util.ssl_.__dict__[	7✔
114	"ssl_wrap_socket"
115	] = true_urllib3_ssl_wrap_socket
116	urllib3.util.ssl_wrap_socket = urllib3.util.__dict__["ssl_wrap_socket"] = (	7✔
117	true_urllib3_ssl_wrap_socket
118	)
119	urllib3.connection.ssl_wrap_socket = urllib3.connection.__dict__[	7✔
120	"ssl_wrap_socket"
121	] = true_urllib3_ssl_wrap_socket
122	urllib3.connection.match_hostname = urllib3.connection.__dict__[	7✔
123	"match_hostname"
124	] = true_urllib3_match_hostname
125	Mocket.reset()	7✔
126	if pyopenssl_override: # pragma: no cover
127	# Put the pyopenssl version back in place
128	inject_into_urllib3()

mindflayer / python-mocket / 11881415972

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous