9000428178

Committed 08 May 2024 10:38AM UTC coverage: 98.59% (-0.2%) from 98.818%

Build # 9000428178

Build Type

push

github

Committed by

web-flow

Commit Message

Begin mypy type-checking (#229)

* mypy config
* narrow initial type-checking
* add type stubs
* allow any generics
* compat typing
* make types
* add type check step to `make test`
* pre-commit fixes
* remove explicit override

this would require depending on typing_extensions

* exceptions
* implicity rexport
* type ignores
* remove unused
* formatting

---------

Co-authored-by: Giorgio Salluzzo <giorgio.salluzzo@gmail.com>

Run Details

24 of 26 new or added lines in 2 files covered. (92.31%)

839 of 851 relevant lines covered (98.59%)

4.89 hits per line

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

95.65

/mocket/utils.py

from __future__ import annotations

import binascii
import io
import os
import ssl
from typing import TYPE_CHECKING, Any, Callable, ClassVar

from .compat import decode_from_bytes, encode_to_bytes
from .exceptions import StrictMocketException

if TYPE_CHECKING:
    from _typeshed import ReadableBuffer
    from typing_extensions import NoReturn

SSL_PROTOCOL = ssl.PROTOCOL_TLSv1_2


class MocketSocketCore(io.BytesIO):
    def write(  # type: ignore[override] # BytesIO returns int
        self,
        content: ReadableBuffer,
    ) -> None:
        super(MocketSocketCore, self).write(content)

        from mocket import Mocket

        if Mocket.r_fd and Mocket.w_fd:
            os.write(Mocket.w_fd, content)


def hexdump(binary_string: bytes) -> str:
    r"""
    >>> hexdump(b"bar foobar foo") == decode_from_bytes(encode_to_bytes("62 61 72 20 66 6F 6F 62 61 72 20 66 6F 6F"))
    True
    """
    bs = decode_from_bytes(binascii.hexlify(binary_string).upper())
    return " ".join(a + b for a, b in zip(bs[::2], bs[1::2]))


def hexload(string: str) -> bytes:
    r"""
    >>> hexload("62 61 72 20 66 6F 6F 62 61 72 20 66 6F 6F") == encode_to_bytes("bar foobar foo")
    True
    """
    string_no_spaces = "".join(string.split())
    return encode_to_bytes(binascii.unhexlify(string_no_spaces))


def get_mocketize(wrapper_: Callable) -> Callable:
    import decorator

    if decorator.__version__ < "5":  # type: ignore[attr-defined] # pragma: no cover
        return decorator.decorator(wrapper_)
    return decorator.decorator(  # type: ignore[call-arg] # kwsyntax
        wrapper_,
        kwsyntax=True,
    )


class MocketMode:
    __shared_state: ClassVar[dict[str, Any]] = {}
    STRICT: ClassVar = None
    STRICT_ALLOWED: ClassVar = None

    def __init__(self) -> None:
        self.__dict__ = self.__shared_state

    def is_allowed(self, location: str | tuple[str, int]) -> bool:
        """
        Checks if (`host`, `port`) or at least `host`
        are allowed locations to perform real `socket` calls
        """
        if not self.STRICT:
            return True

        host_allowed = False
        if isinstance(location, tuple):
            host_allowed = location[0] in self.STRICT_ALLOWED
        return host_allowed or location in self.STRICT_ALLOWED

    @staticmethod
    def raise_not_allowed() -> NoReturn:
        from .mocket import Mocket

        current_entries = [
            (location, "\n    ".join(map(str, entries)))
            for location, entries in Mocket._entries.items()
        ]
        formatted_entries = "\n".join(
            [f"  {location}:\n    {entries}" for location, entries in current_entries]
        )
        raise StrictMocketException(
            "Mocket tried to use the real `socket` module while STRICT mode was active.\n"
            f"Registered entries:\n{formatted_entries}"
        )

1	from __future__ import annotations	5✔
2
3	import binascii	5✔
4	import io	5✔
5	import os	5✔
6	import ssl	5✔
7	from typing import TYPE_CHECKING, Any, Callable, ClassVar	5✔
8
9	from .compat import decode_from_bytes, encode_to_bytes	5✔
10	from .exceptions import StrictMocketException	5✔
11
12	if TYPE_CHECKING:	5✔
NEW 13	from _typeshed import ReadableBuffer	×
NEW 14	from typing_extensions import NoReturn	×
15
16	SSL_PROTOCOL = ssl.PROTOCOL_TLSv1_2	5✔
17
18
19	class MocketSocketCore(io.BytesIO):	5✔
20	def write( # type: ignore[override] # BytesIO returns int	5✔
21	self,
22	content: ReadableBuffer,
23	) -> None:
24	super(MocketSocketCore, self).write(content)	5✔
25
26	from mocket import Mocket	5✔
27
28	if Mocket.r_fd and Mocket.w_fd:	5✔
29	os.write(Mocket.w_fd, content)	5✔
30
31
32	def hexdump(binary_string: bytes) -> str:	5✔
33	r"""
34	>>> hexdump(b"bar foobar foo") == decode_from_bytes(encode_to_bytes("62 61 72 20 66 6F 6F 62 61 72 20 66 6F 6F"))
35	True
36	"""
37	bs = decode_from_bytes(binascii.hexlify(binary_string).upper())	5✔
38	return " ".join(a + b for a, b in zip(bs[::2], bs[1::2]))	5✔
39
40
41	def hexload(string: str) -> bytes:	5✔
42	r"""
43	>>> hexload("62 61 72 20 66 6F 6F 62 61 72 20 66 6F 6F") == encode_to_bytes("bar foobar foo")
44	True
45	"""
46	string_no_spaces = "".join(string.split())	5✔
47	return encode_to_bytes(binascii.unhexlify(string_no_spaces))	5✔
48
49
50	def get_mocketize(wrapper_: Callable) -> Callable:	5✔
51	import decorator	5✔
52
53	if decorator.__version__ < "5": # type: ignore[attr-defined] # pragma: no cover
54	return decorator.decorator(wrapper_)
55	return decorator.decorator( # type: ignore[call-arg] # kwsyntax	5✔
56	wrapper_,
57	kwsyntax=True,
58	)
59
60
61	class MocketMode:	5✔
62	__shared_state: ClassVar[dict[str, Any]] = {}	5✔
63	STRICT: ClassVar = None	5✔
64	STRICT_ALLOWED: ClassVar = None	5✔
65
66	def __init__(self) -> None:	5✔
67	self.__dict__ = self.__shared_state	5✔
68
69	def is_allowed(self, location: str \| tuple[str, int]) -> bool:	5✔
70	"""
71	Checks if (`host`, `port`) or at least `host`
72	are allowed locations to perform real `socket` calls
73	"""
74	if not self.STRICT:	5✔
75	return True	5✔
76
77	host_allowed = False	5✔
78	if isinstance(location, tuple):	5✔
79	host_allowed = location[0] in self.STRICT_ALLOWED	5✔
80	return host_allowed or location in self.STRICT_ALLOWED	5✔
81
82	@staticmethod	5✔
83	def raise_not_allowed() -> NoReturn:	5✔
84	from .mocket import Mocket	5✔
85
86	current_entries = [	5✔
87	(location, "\n ".join(map(str, entries)))
88	for location, entries in Mocket._entries.items()
89	]
90	formatted_entries = "\n".join(	5✔
91	[f" {location}:\n {entries}" for location, entries in current_entries]
92	)
93	raise StrictMocketException(	5✔
94	"Mocket tried to use the real `socket` module while STRICT mode was active.\n"
95	f"Registered entries:\n{formatted_entries}"
96	)

mindflayer / python-mocket / 9000428178

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