12028314747

Committed 26 Nov 2024 10:09AM UTC coverage: 98.682% (-0.3%) from 99.025%

Build # 12028314747

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor introduce recording storage (#274)

* refactor: separate injection and enable/disable logic
* refactor: add class that handles request records

Run Details

135 of 142 new or added lines in 5 files covered. (95.07%)

3 existing lines in 2 files now uncovered.

973 of 986 relevant lines covered (98.68%)

6.85 hits per line

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

94.44

/mocket/mocket.py

from __future__ import annotations

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

import mocket.inject
from mocket.recording import MocketRecordStorage

# 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] = []
    _record_storage: ClassVar[MocketRecordStorage | None] = None

    @classmethod
    def enable(
        cls,
        namespace: str | None = None,
        truesocket_recording_dir: str | None = None,
    ) -> None:
        if namespace is None:
            namespace = str(id(cls._entries))

        if truesocket_recording_dir is not None:
            recording_dir = Path(truesocket_recording_dir)

            if not recording_dir.is_dir():
                # JSON dumps will be saved here
                raise AssertionError

            cls._record_storage = MocketRecordStorage(
                directory=recording_dir,
                namespace=namespace,
            )

        mocket.inject.enable()

    @classmethod
    def disable(cls) -> None:
        cls.reset()

        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 = []
        cls._record_storage = None

    @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 | None:
        if not cls._record_storage:
            return None
        return cls._record_storage.namespace

    @classmethod
    def get_truesocket_recording_dir(cls) -> str | None:
        if not cls._record_storage:
            return None
        return str(cls._record_storage.directory)

    @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 pathlib import Path	7✔
7	from typing import TYPE_CHECKING, ClassVar	7✔
8
9	import mocket.inject	7✔
10	from mocket.recording import MocketRecordStorage	7✔
11
12	# NOTE this is here for backwards-compat to keep old import-paths working
13	# from mocket.socket import MocketSocket as MocketSocket
14
15	if TYPE_CHECKING:	7✔
16	from mocket.entry import MocketEntry	×
17	from mocket.types import Address	×
18
19
20	class Mocket:	7✔
21	_socket_pairs: ClassVar[dict[Address, tuple[int, int]]] = {}	7✔
22	_address: ClassVar[Address] = (None, None)	7✔
23	_entries: ClassVar[dict[Address, list[MocketEntry]]] = collections.defaultdict(list)	7✔
24	_requests: ClassVar[list] = []	7✔
25	_record_storage: ClassVar[MocketRecordStorage \| None] = None	7✔
26
27	@classmethod	7✔
28	def enable(	7✔
29	cls,
30	namespace: str \| None = None,
31	truesocket_recording_dir: str \| None = None,
32	) -> None:
33	if namespace is None:	7✔
34	namespace = str(id(cls._entries))	7✔
35
36	if truesocket_recording_dir is not None:	7✔
37	recording_dir = Path(truesocket_recording_dir)	7✔
38
39	if not recording_dir.is_dir():	7✔
40	# JSON dumps will be saved here
NEW 41	raise AssertionError	×
42
43	cls._record_storage = MocketRecordStorage(	7✔
44	directory=recording_dir,
45	namespace=namespace,
46	)
47
48	mocket.inject.enable()	7✔
49
50	@classmethod	7✔
51	def disable(cls) -> None:	7✔
52	cls.reset()	7✔
53
54	mocket.inject.disable()	7✔
55
56	@classmethod	7✔
57	def get_pair(cls, address: Address) -> tuple[int, int] \| tuple[None, None]:	7✔
58	"""
59	Given the id() of the caller, return a pair of file descriptors
60	as a tuple of two integers: (<read_fd>, <write_fd>)
61	"""
62	return cls._socket_pairs.get(address, (None, None))	7✔
63
64	@classmethod	7✔
65	def set_pair(cls, address: Address, pair: tuple[int, int]) -> None:	7✔
66	"""
67	Store a pair of file descriptors under the key `id_`
68	as a tuple of two integers: (<read_fd>, <write_fd>)
69	"""
70	cls._socket_pairs[address] = pair	7✔
71
72	@classmethod	7✔
73	def register(cls, *entries: MocketEntry) -> None:	7✔
74	for entry in entries:	7✔
75	cls._entries[entry.location].append(entry)	7✔
76
77	@classmethod	7✔
78	def get_entry(cls, host: str, port: int, data) -> MocketEntry \| None:	7✔
79	host = host or cls._address[0]	7✔
80	port = port or cls._address[1]	7✔
81	entries = cls._entries.get((host, port), [])	7✔
82	for entry in entries:	7✔
83	if entry.can_handle(data):	7✔
84	return entry	7✔
85	return None	7✔
86
87	@classmethod	7✔
88	def collect(cls, data) -> None:	7✔
89	cls._requests.append(data)	7✔
90
91	@classmethod	7✔
92	def reset(cls) -> None:	7✔
93	for r_fd, w_fd in cls._socket_pairs.values():	7✔
94	os.close(r_fd)	7✔
95	os.close(w_fd)	7✔
96	cls._socket_pairs = {}	7✔
97	cls._entries = collections.defaultdict(list)	7✔
98	cls._requests = []	7✔
99	cls._record_storage = None	7✔
100
101	@classmethod	7✔
102	def last_request(cls):	7✔
103	if cls.has_requests():	7✔
104	return cls._requests[-1]	7✔
105
106	@classmethod	7✔
107	def request_list(cls):	7✔
108	return cls._requests	7✔
109
110	@classmethod	7✔
111	def remove_last_request(cls) -> None:	7✔
112	if cls.has_requests():	7✔
113	del cls._requests[-1]	7✔
114
115	@classmethod	7✔
116	def has_requests(cls) -> bool:	7✔
117	return bool(cls.request_list())	7✔
118
119	@classmethod	7✔
120	def get_namespace(cls) -> str \| None:	7✔
121	if not cls._record_storage:	7✔
NEW 122	return None	×
123	return cls._record_storage.namespace	7✔
124
125	@classmethod	7✔
126	def get_truesocket_recording_dir(cls) -> str \| None:	7✔
127	if not cls._record_storage:	7✔
NEW 128	return None	×
129	return str(cls._record_storage.directory)	7✔
130
131	@classmethod	7✔
132	def assert_fail_if_entries_not_served(cls) -> None:	7✔
133	"""Mocket checks that all entries have been served at least once."""
134	if not all(entry._served for entry in itertools.chain(*cls._entries.values())):	7✔
135	raise AssertionError("Some Mocket entries have not been served")	7✔

mindflayer / python-mocket / 12028314747

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