12013336921

Committed 25 Nov 2024 03:28PM UTC coverage: 99.025%. Remained the same

Build # 12013336921

Build Type

Pull #273

github

Committed by

web-flow

Commit Message

Merge 5405e2ea4 into a5b5e34b8

Pull Request Pull Request #273: Target `make safetest` got broken

Run Details

914 of 923 relevant lines covered (99.02%)

6.88 hits per line

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

97.5

/mocket/inject.py

from __future__ import annotations

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

import urllib3

_patches_restore: dict[tuple[ModuleType, str], Any] = {}


def _patch(module: ModuleType, name: str, patched_value: Any) -> None:
    with contextlib.suppress(KeyError):
        original_value, module.__dict__[name] = module.__dict__[name], patched_value
        _patches_restore[(module, name)] = original_value


def _restore(module: ModuleType, name: str) -> None:
    if original_value := _patches_restore.pop((module, name)):
        module.__dict__[name] = original_value


def enable(
    namespace: str | None = None,
    truesocket_recording_dir: str | None = None,
) -> None:
    from mocket.socket import (
        MocketSocket,
        mock_create_connection,
        mock_getaddrinfo,
        mock_gethostbyname,
        mock_gethostname,
        mock_inet_pton,
        mock_socketpair,
    )
    from mocket.ssl.context import MocketSSLContext, mock_wrap_socket
    from mocket.urllib3 import (
        mock_match_hostname as mock_urllib3_match_hostname,
    )
    from mocket.urllib3 import (
        mock_ssl_wrap_socket as mock_urllib3_ssl_wrap_socket,
    )

    patches = {
        # stdlib: socket
        (socket, "socket"): MocketSocket,
        (socket, "create_connection"): mock_create_connection,
        (socket, "getaddrinfo"): mock_getaddrinfo,
        (socket, "gethostbyname"): mock_gethostbyname,
        (socket, "gethostname"): mock_gethostname,
        (socket, "inet_pton"): mock_inet_pton,
        (socket, "SocketType"): MocketSocket,
        (socket, "socketpair"): mock_socketpair,
        # stdlib: ssl
        (ssl, "SSLContext"): MocketSSLContext,
        (ssl, "wrap_socket"): mock_wrap_socket,  # python < 3.12.0
        # urllib3
        (urllib3.connection, "match_hostname"): mock_urllib3_match_hostname,
        (urllib3.connection, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
        (urllib3.util, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
        (urllib3.util.ssl_, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
        (urllib3.util.ssl_, "wrap_socket"): mock_urllib3_ssl_wrap_socket,  # urllib3 < 2
    }

    for (module, name), new_value in patches.items():
        _patch(module, name, new_value)

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

        extract_from_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:
    for module, name in list(_patches_restore.keys()):
        _restore(module, name)

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

        inject_into_urllib3()

    from mocket.mocket import Mocket

    Mocket.reset()

1	from __future__ import annotations	7✔
2
3	import contextlib	7✔
4	import os	7✔
5	import socket	7✔
6	import ssl	7✔
7	from types import ModuleType	7✔
8	from typing import Any	7✔
9
10	import urllib3	7✔
11
12	_patches_restore: dict[tuple[ModuleType, str], Any] = {}	7✔
13
14
15	def _patch(module: ModuleType, name: str, patched_value: Any) -> None:	7✔
16	with contextlib.suppress(KeyError):	7✔
17	original_value, module.__dict__[name] = module.__dict__[name], patched_value	7✔
18	_patches_restore[(module, name)] = original_value	7✔
19
20
21	def _restore(module: ModuleType, name: str) -> None:	7✔
22	if original_value := _patches_restore.pop((module, name)):	7✔
23	module.__dict__[name] = original_value	7✔
24
25
26	def enable(	7✔
27	namespace: str \| None = None,
28	truesocket_recording_dir: str \| None = None,
29	) -> None:
30	from mocket.socket import (	7✔
31	MocketSocket,
32	mock_create_connection,
33	mock_getaddrinfo,
34	mock_gethostbyname,
35	mock_gethostname,
36	mock_inet_pton,
37	mock_socketpair,
38	)
39	from mocket.ssl.context import MocketSSLContext, mock_wrap_socket	7✔
40	from mocket.urllib3 import (	7✔
41	mock_match_hostname as mock_urllib3_match_hostname,
42	)
43	from mocket.urllib3 import (	7✔
44	mock_ssl_wrap_socket as mock_urllib3_ssl_wrap_socket,
45	)
46
47	patches = {	7✔
48	# stdlib: socket
49	(socket, "socket"): MocketSocket,
50	(socket, "create_connection"): mock_create_connection,
51	(socket, "getaddrinfo"): mock_getaddrinfo,
52	(socket, "gethostbyname"): mock_gethostbyname,
53	(socket, "gethostname"): mock_gethostname,
54	(socket, "inet_pton"): mock_inet_pton,
55	(socket, "SocketType"): MocketSocket,
56	(socket, "socketpair"): mock_socketpair,
57	# stdlib: ssl
58	(ssl, "SSLContext"): MocketSSLContext,
59	(ssl, "wrap_socket"): mock_wrap_socket, # python < 3.12.0
60	# urllib3
61	(urllib3.connection, "match_hostname"): mock_urllib3_match_hostname,
62	(urllib3.connection, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
63	(urllib3.util, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
64	(urllib3.util.ssl_, "ssl_wrap_socket"): mock_urllib3_ssl_wrap_socket,
65	(urllib3.util.ssl_, "wrap_socket"): mock_urllib3_ssl_wrap_socket, # urllib3 < 2
66	}
67
68	for (module, name), new_value in patches.items():	7✔
69	_patch(module, name, new_value)	7✔
70
71	with contextlib.suppress(ImportError):	7✔
72	from urllib3.contrib.pyopenssl import extract_from_urllib3	7✔
73
74	extract_from_urllib3()	1✔
75
76	from mocket.mocket import Mocket	7✔
77
78	Mocket._namespace = namespace	7✔
79	Mocket._truesocket_recording_dir = truesocket_recording_dir	7✔
80	if truesocket_recording_dir and not os.path.isdir(truesocket_recording_dir):	7✔
81	# JSON dumps will be saved here
82	raise AssertionError	×
83
84
85	def disable() -> None:	7✔
86	for module, name in list(_patches_restore.keys()):	7✔
87	_restore(module, name)	7✔
88
89	with contextlib.suppress(ImportError):	7✔
90	from urllib3.contrib.pyopenssl import inject_into_urllib3	7✔
91
92	inject_into_urllib3()	1✔
93
94	from mocket.mocket import Mocket	7✔
95
96	Mocket.reset()	7✔

mindflayer / python-mocket / 12013336921

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