11884044999

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

Build # 11884044999

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor: introduce state object (#264)

* refactor: move enable- and disable-functions from mocket.mocket to mocket.inject
* refactor: Mocket - add typing and get rid of cyclic import

Run Details

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

1 existing line 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

97.22

/mocket/mocket.py

from __future__ import annotations

import collections
import itertools
import os
from typing import TYPE_CHECKING, ClassVar

import mocket.inject

# NOTE this is here for backwards-compat to keep old import-paths working
# from mocket.socket import MocketSocket as MocketSocket

if TYPE_CHECKING:
    from mocket.entry import MocketEntry
    from mocket.types import Address


class Mocket:
    _socket_pairs: ClassVar[dict[Address, tuple[int, int]]] = {}
    _address: ClassVar[Address] = (None, None)
    _entries: ClassVar[dict[Address, list[MocketEntry]]] = collections.defaultdict(list)
    _requests: ClassVar[list] = []
    _namespace: ClassVar[str] = str(id(_entries))
    _truesocket_recording_dir: ClassVar[str | None] = None

    enable = mocket.inject.enable
    disable = mocket.inject.disable

    @classmethod
    def get_pair(cls, address: Address) -> tuple[int, int] | tuple[None, None]:
        """
        Given the id() of the caller, return a pair of file descriptors
        as a tuple of two integers: (<read_fd>, <write_fd>)
        """
        return cls._socket_pairs.get(address, (None, None))

    @classmethod
    def set_pair(cls, address: Address, pair: tuple[int, int]) -> None:
        """
        Store a pair of file descriptors under the key `id_`
        as a tuple of two integers: (<read_fd>, <write_fd>)
        """
        cls._socket_pairs[address] = pair

    @classmethod
    def register(cls, *entries: MocketEntry) -> None:
        for entry in entries:
            cls._entries[entry.location].append(entry)

    @classmethod
    def get_entry(cls, host: str, port: int, data) -> MocketEntry | None:
        host = host or cls._address[0]
        port = port or cls._address[1]
        entries = cls._entries.get((host, port), [])
        for entry in entries:
            if entry.can_handle(data):
                return entry
        return None

    @classmethod
    def collect(cls, data) -> None:
        cls._requests.append(data)

    @classmethod
    def reset(cls) -> None:
        for r_fd, w_fd in cls._socket_pairs.values():
            os.close(r_fd)
            os.close(w_fd)
        cls._socket_pairs = {}
        cls._entries = collections.defaultdict(list)
        cls._requests = []

    @classmethod
    def last_request(cls):
        if cls.has_requests():
            return cls._requests[-1]

    @classmethod
    def request_list(cls):
        return cls._requests

    @classmethod
    def remove_last_request(cls) -> None:
        if cls.has_requests():
            del cls._requests[-1]

    @classmethod
    def has_requests(cls) -> bool:
        return bool(cls.request_list())

    @classmethod
    def get_namespace(cls) -> str:
        return cls._namespace

    @classmethod
    def get_truesocket_recording_dir(cls) -> str | None:
        return cls._truesocket_recording_dir

    @classmethod
    def assert_fail_if_entries_not_served(cls) -> None:
        """Mocket checks that all entries have been served at least once."""
        if not all(entry._served for entry in itertools.chain(*cls._entries.values())):
            raise AssertionError("Some Mocket entries have not been served")

1	from __future__ import annotations	7✔
2
3	import collections	7✔
4	import itertools	7✔
5	import os	7✔
6	from typing import TYPE_CHECKING, ClassVar	7✔
7
8	import mocket.inject	7✔
9
10	# NOTE this is here for backwards-compat to keep old import-paths working
11	# from mocket.socket import MocketSocket as MocketSocket
12
13	if TYPE_CHECKING:	7✔
NEW 14	from mocket.entry import MocketEntry	×
NEW 15	from mocket.types import Address	×
16
17
18	class Mocket:	7✔
19	_socket_pairs: ClassVar[dict[Address, tuple[int, int]]] = {}	7✔
20	_address: ClassVar[Address] = (None, None)	7✔
21	_entries: ClassVar[dict[Address, list[MocketEntry]]] = collections.defaultdict(list)	7✔
22	_requests: ClassVar[list] = []	7✔
23	_namespace: ClassVar[str] = str(id(_entries))	7✔
24	_truesocket_recording_dir: ClassVar[str \| None] = None	7✔
25
26	enable = mocket.inject.enable	7✔
27	disable = mocket.inject.disable	7✔
28
29	@classmethod	7✔
30	def get_pair(cls, address: Address) -> tuple[int, int] \| tuple[None, None]:	7✔
31	"""
32	Given the id() of the caller, return a pair of file descriptors
33	as a tuple of two integers: (<read_fd>, <write_fd>)
34	"""
35	return cls._socket_pairs.get(address, (None, None))	7✔
36
37	@classmethod	7✔
38	def set_pair(cls, address: Address, pair: tuple[int, int]) -> None:	7✔
39	"""
40	Store a pair of file descriptors under the key `id_`
41	as a tuple of two integers: (<read_fd>, <write_fd>)
42	"""
43	cls._socket_pairs[address] = pair	7✔
44
45	@classmethod	7✔
46	def register(cls, *entries: MocketEntry) -> None:	7✔
47	for entry in entries:	7✔
48	cls._entries[entry.location].append(entry)	7✔
49
50	@classmethod	7✔
51	def get_entry(cls, host: str, port: int, data) -> MocketEntry \| None:	7✔
52	host = host or cls._address[0]	7✔
53	port = port or cls._address[1]	7✔
54	entries = cls._entries.get((host, port), [])	7✔
55	for entry in entries:	7✔
56	if entry.can_handle(data):	7✔
57	return entry	7✔
58	return None	7✔
59
60	@classmethod	7✔
61	def collect(cls, data) -> None:	7✔
62	cls._requests.append(data)	7✔
63
64	@classmethod	7✔
65	def reset(cls) -> None:	7✔
66	for r_fd, w_fd in cls._socket_pairs.values():	7✔
67	os.close(r_fd)	7✔
68	os.close(w_fd)	7✔
69	cls._socket_pairs = {}	7✔
70	cls._entries = collections.defaultdict(list)	7✔
71	cls._requests = []	7✔
72
73	@classmethod	7✔
74	def last_request(cls):	7✔
75	if cls.has_requests():	7✔
76	return cls._requests[-1]	7✔
77
78	@classmethod	7✔
79	def request_list(cls):	7✔
80	return cls._requests	7✔
81
82	@classmethod	7✔
83	def remove_last_request(cls) -> None:	7✔
84	if cls.has_requests():	7✔
85	del cls._requests[-1]	7✔
86
87	@classmethod	7✔
88	def has_requests(cls) -> bool:	7✔
89	return bool(cls.request_list())	7✔
90
91	@classmethod	7✔
92	def get_namespace(cls) -> str:	7✔
93	return cls._namespace	7✔
94
95	@classmethod	7✔
96	def get_truesocket_recording_dir(cls) -> str \| None:	7✔
97	return cls._truesocket_recording_dir	7✔
98
99	@classmethod	7✔
100	def assert_fail_if_entries_not_served(cls) -> None:	7✔
101	"""Mocket checks that all entries have been served at least once."""
102	if not all(entry._served for entry in itertools.chain(*cls._entries.values())):	7✔
103	raise AssertionError("Some Mocket entries have not been served")	7✔

mindflayer / python-mocket / 11884044999

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