11988860782

Committed 23 Nov 2024 04:57PM UTC coverage: 98.876% (-0.4%) from 99.255%

Build # 11988860782

Build Type

Pull #268

github

Committed by

web-flow

Commit Message

Merge 199e90344 into 0da27224a

Pull Request Pull Request #268: Improve readability and backwards-compatibility of injection-code

Run Details

101 of 105 new or added lines in 4 files covered. (96.19%)

7 existing lines in 2 files now uncovered.

968 of 979 relevant lines covered (98.88%)

6.87 hits per line

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

95.35

/mocket/inject.py

from __future__ import annotations

import contextlib
import os
from types import ModuleType
from typing import Any

from packaging.version import Version

from mocket.utils import package_version, python_version


def _replace(module: ModuleType, name: str, new_value: Any) -> None:
    module.__dict__[name] = new_value


def _inject_stdlib_socket() -> None:
    import socket

    from mocket.socket import (
        MocketSocket,
        mock_create_connection,
        mock_getaddrinfo,
        mock_gethostbyname,
        mock_gethostname,
        mock_inet_pton,
        mock_socketpair,
    )

    _replace(socket, "socket", MocketSocket)
    _replace(socket, "SocketType", MocketSocket)
    _replace(socket, "create_connection", mock_create_connection)
    _replace(socket, "getaddrinfo", mock_getaddrinfo)
    _replace(socket, "gethostbyname", mock_gethostbyname)
    _replace(socket, "gethostname", mock_gethostname)
    _replace(socket, "inet_pton", mock_inet_pton)
    _replace(socket, "socketpair", mock_socketpair)


def _restore_stdlib_socket() -> None:
    import socket

    from mocket.socket import (
        true_socket_create_connection,
        true_socket_getaddrinfo,
        true_socket_gethostbyname,
        true_socket_gethostname,
        true_socket_inet_pton,
        true_socket_socket,
        true_socket_socket_type,
        true_socket_socketpair,
    )

    _replace(socket, "SocketType", true_socket_socket_type)
    _replace(socket, "create_connection", true_socket_create_connection)
    _replace(socket, "getaddrinfo", true_socket_getaddrinfo)
    _replace(socket, "gethostbyname", true_socket_gethostbyname)
    _replace(socket, "gethostname", true_socket_gethostname)
    _replace(socket, "inet_pton", true_socket_inet_pton)
    _replace(socket, "socket", true_socket_socket)
    _replace(socket, "socketpair", true_socket_socketpair)


def _inject_stdlib_ssl() -> None:
    import ssl

    from mocket.ssl.context import MocketSSLContext

    _replace(ssl, "SSLContext", MocketSSLContext)

    if python_version() < Version("3.12.0"):
        _replace(ssl, "wrap_socket", MocketSSLContext.wrap_socket)


def _restore_stdlib_ssl() -> None:
    import ssl

    from mocket.ssl.context import true_ssl_ssl_context

    _replace(ssl, "SSLContext", true_ssl_ssl_context)

    if python_version() < Version("3.12.0"):
        from mocket.ssl.context import true_ssl_wrap_socket

        _replace(ssl, "wrap_socket", true_ssl_wrap_socket)


def _inject_urllib3() -> None:
    import urllib3

    from mocket.socket import mock_urllib3_match_hostname
    from mocket.ssl.context import MocketSSLContext

    _replace(urllib3.util.ssl_, "ssl_wrap_socket", MocketSSLContext.wrap_socket)
    _replace(urllib3.util, "ssl_wrap_socket", MocketSSLContext.wrap_socket)
    _replace(urllib3.connection, "ssl_wrap_socket", MocketSSLContext.wrap_socket)
    _replace(urllib3.connection, "match_hostname", mock_urllib3_match_hostname)

    if package_version("urllib3") < Version("2.0.0"):
        _replace(urllib3.util.ssl_, "wrap_socket", MocketSSLContext.wrap_socket)

    with contextlib.suppress(ImportError):
        from urllib3.contrib.pyopenssl import extract_from_urllib3

        extract_from_urllib3()


def _restore_urllib3() -> None:
    import urllib3

    from mocket.socket import true_urllib3_match_hostname
    from mocket.ssl.context import true_urllib3_ssl_wrap_socket

    _replace(urllib3.connection, "match_hostname", true_urllib3_match_hostname)
    _replace(urllib3.util.ssl_, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)
    _replace(urllib3.util, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)
    _replace(urllib3.connection, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)

    if package_version("urllib3") < Version("2.0.0"):
        from mocket.ssl.context import true_urllib3_wrap_socket

        _replace(urllib3.util.ssl_, "wrap_socket", true_urllib3_wrap_socket)

    with contextlib.suppress(ImportError):
        from urllib3.contrib.pyopenssl import inject_into_urllib3

        inject_into_urllib3()


def enable(
    namespace: str | None = None,
    truesocket_recording_dir: str | None = None,
) -> None:
    _inject_stdlib_socket()
    _inject_stdlib_ssl()
    _inject_urllib3()

    from mocket.mocket import Mocket

    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


def disable() -> None:
    _restore_stdlib_socket()
    _restore_stdlib_ssl()
    _restore_urllib3()

    from mocket.mocket import Mocket

    Mocket.reset()

1	from __future__ import annotations	7✔
2
3	import contextlib	7✔
4	import os	7✔
5	from types import ModuleType	7✔
6	from typing import Any	7✔
7
8	from packaging.version import Version	7✔
9
10	from mocket.utils import package_version, python_version	7✔
11
12
13	def _replace(module: ModuleType, name: str, new_value: Any) -> None:	7✔
14	module.__dict__[name] = new_value	7✔
15
16
17	def _inject_stdlib_socket() -> None:	7✔
18	import socket	7✔
19
20	from mocket.socket import (	7✔
21	MocketSocket,
22	mock_create_connection,
23	mock_getaddrinfo,
24	mock_gethostbyname,
25	mock_gethostname,
26	mock_inet_pton,
27	mock_socketpair,
28	)
29
30	_replace(socket, "socket", MocketSocket)	7✔
31	_replace(socket, "SocketType", MocketSocket)	7✔
32	_replace(socket, "create_connection", mock_create_connection)	7✔
33	_replace(socket, "getaddrinfo", mock_getaddrinfo)	7✔
34	_replace(socket, "gethostbyname", mock_gethostbyname)	7✔
35	_replace(socket, "gethostname", mock_gethostname)	7✔
36	_replace(socket, "inet_pton", mock_inet_pton)	7✔
37	_replace(socket, "socketpair", mock_socketpair)	7✔
38
39
40	def _restore_stdlib_socket() -> None:	7✔
41	import socket	7✔
42
43	from mocket.socket import (	7✔
44	true_socket_create_connection,
45	true_socket_getaddrinfo,
46	true_socket_gethostbyname,
47	true_socket_gethostname,
48	true_socket_inet_pton,
49	true_socket_socket,
50	true_socket_socket_type,
51	true_socket_socketpair,
52	)
53
54	_replace(socket, "SocketType", true_socket_socket_type)	7✔
55	_replace(socket, "create_connection", true_socket_create_connection)	7✔
56	_replace(socket, "getaddrinfo", true_socket_getaddrinfo)	7✔
57	_replace(socket, "gethostbyname", true_socket_gethostbyname)	7✔
58	_replace(socket, "gethostname", true_socket_gethostname)	7✔
59	_replace(socket, "inet_pton", true_socket_inet_pton)	7✔
60	_replace(socket, "socket", true_socket_socket)	7✔
61	_replace(socket, "socketpair", true_socket_socketpair)	7✔
62
63
64	def _inject_stdlib_ssl() -> None:	7✔
65	import ssl	7✔
66
67	from mocket.ssl.context import MocketSSLContext	7✔
68
69	_replace(ssl, "SSLContext", MocketSSLContext)	7✔
70
71	if python_version() < Version("3.12.0"):	7✔
72	_replace(ssl, "wrap_socket", MocketSSLContext.wrap_socket)	5✔
73
74
75	def _restore_stdlib_ssl() -> None:	7✔
76	import ssl	7✔
77
78	from mocket.ssl.context import true_ssl_ssl_context	7✔
79
80	_replace(ssl, "SSLContext", true_ssl_ssl_context)	7✔
81
82	if python_version() < Version("3.12.0"):	7✔
83	from mocket.ssl.context import true_ssl_wrap_socket	5✔
84
85	_replace(ssl, "wrap_socket", true_ssl_wrap_socket)	5✔
86
87
88	def _inject_urllib3() -> None:	7✔
89	import urllib3	7✔
90
91	from mocket.socket import mock_urllib3_match_hostname	7✔
92	from mocket.ssl.context import MocketSSLContext	7✔
93
94	_replace(urllib3.util.ssl_, "ssl_wrap_socket", MocketSSLContext.wrap_socket)	7✔
95	_replace(urllib3.util, "ssl_wrap_socket", MocketSSLContext.wrap_socket)	7✔
96	_replace(urllib3.connection, "ssl_wrap_socket", MocketSSLContext.wrap_socket)	7✔
97	_replace(urllib3.connection, "match_hostname", mock_urllib3_match_hostname)	7✔
98
99	if package_version("urllib3") < Version("2.0.0"):	7✔
NEW 100	_replace(urllib3.util.ssl_, "wrap_socket", MocketSSLContext.wrap_socket)	×
101
102	with contextlib.suppress(ImportError):	7✔
103	from urllib3.contrib.pyopenssl import extract_from_urllib3	7✔
104
105	extract_from_urllib3()	1✔
106
107
108	def _restore_urllib3() -> None:	7✔
109	import urllib3	7✔
110
111	from mocket.socket import true_urllib3_match_hostname	7✔
112	from mocket.ssl.context import true_urllib3_ssl_wrap_socket	7✔
113
114	_replace(urllib3.connection, "match_hostname", true_urllib3_match_hostname)	7✔
115	_replace(urllib3.util.ssl_, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)	7✔
116	_replace(urllib3.util, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)	7✔
117	_replace(urllib3.connection, "ssl_wrap_socket", true_urllib3_ssl_wrap_socket)	7✔
118
119	if package_version("urllib3") < Version("2.0.0"):	7✔
NEW 120	from mocket.ssl.context import true_urllib3_wrap_socket	×
121
NEW 122	_replace(urllib3.util.ssl_, "wrap_socket", true_urllib3_wrap_socket)	×
123
124	with contextlib.suppress(ImportError):	7✔
125	from urllib3.contrib.pyopenssl import inject_into_urllib3	7✔
126
127	inject_into_urllib3()	1✔
128
129
130	def enable(	7✔
131	namespace: str \| None = None,
132	truesocket_recording_dir: str \| None = None,
133	) -> None:
134	_inject_stdlib_socket()	7✔
135	_inject_stdlib_ssl()	7✔
136	_inject_urllib3()	7✔
137
138	from mocket.mocket import Mocket	7✔
139
140	Mocket._namespace = namespace	7✔
141	Mocket._truesocket_recording_dir = truesocket_recording_dir	7✔
142	if truesocket_recording_dir and not os.path.isdir(truesocket_recording_dir):	7✔
143	# JSON dumps will be saved here
144	raise AssertionError	×
145
146
147	def disable() -> None:	7✔
148	_restore_stdlib_socket()	7✔
149	_restore_stdlib_ssl()	7✔
150	_restore_urllib3()	7✔
151
152	from mocket.mocket import Mocket	7✔
153
154	Mocket.reset()	7✔

mindflayer / python-mocket / 11988860782

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