12095558571

Committed 30 Nov 2024 10:11AM UTC coverage: 97.727% (-1.0%) from 98.68%

Build # 12095558571

Build Type

Pull #276

github

Committed by

web-flow

Commit Message

Merge a69d07eba into 9cfad4cdb

Pull Request Pull Request #276: refactor: rework Entry, Request and Response classes

Run Details

732 of 752 new or added lines in 30 files covered. (97.34%)

1 existing line in 1 file now uncovered.

1204 of 1232 relevant lines covered (97.73%)

6.8 hits per line

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

94.57

/mocket/core/mocket.py

from __future__ import annotations

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

import mocket.core.inject
from mocket.core.recording import MocketRecordStorage

if TYPE_CHECKING:
    from mocket.core.entry import MocketBaseEntry, MocketBaseRequest
    from mocket.core.types import Address


class Mocket:
    _socket_pairs: ClassVar[dict[Address, tuple[int, int]]] = {}
    _address: ClassVar[Address] = (None, None)
    _entries: ClassVar[dict[Address, list[MocketBaseEntry]]] = collections.defaultdict(
        list
    )
    _last_entry: ClassVar[MocketBaseEntry | None] = None  # TODO review how this is used
    _requests: ClassVar[list[MocketBaseRequest]] = []
    _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.core.inject.enable()

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

        mocket.core.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: MocketBaseEntry) -> None:
        for entry in entries:
            cls._entries[entry.address].append(entry)

    @classmethod
    def get_entry(cls, host: str, port: int, data) -> MocketBaseEntry | 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) -> MocketBaseRequest | None:
        if cls.has_requests():
            return cls._requests[-1]
        return None

    @classmethod
    def request_list(cls) -> list[MocketBaseRequest]:
        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_response 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.core.inject	7✔
10	from mocket.core.recording import MocketRecordStorage	7✔
11
12	if TYPE_CHECKING:	7✔
NEW 13	from mocket.core.entry import MocketBaseEntry, MocketBaseRequest	×
NEW 14	from mocket.core.types import Address	×
15
16
17	class Mocket:	7✔
18	_socket_pairs: ClassVar[dict[Address, tuple[int, int]]] = {}	7✔
19	_address: ClassVar[Address] = (None, None)	7✔
20	_entries: ClassVar[dict[Address, list[MocketBaseEntry]]] = collections.defaultdict(	7✔
21	list
22	)
23	_last_entry: ClassVar[MocketBaseEntry \| None] = None # TODO review how this is used	7✔
24	_requests: ClassVar[list[MocketBaseRequest]] = []	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.core.inject.enable()	7✔
49
50	@classmethod	7✔
51	def disable(cls) -> None:	7✔
52	cls.reset()	7✔
53
54	mocket.core.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: MocketBaseEntry) -> None:	7✔
74	for entry in entries:	7✔
75	cls._entries[entry.address].append(entry)	7✔
76
77	@classmethod	7✔
78	def get_entry(cls, host: str, port: int, data) -> MocketBaseEntry \| 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) -> MocketBaseRequest \| None:	7✔
103	if cls.has_requests():	7✔
104	return cls._requests[-1]	7✔
105	return None	7✔
106
107	@classmethod	7✔
108	def request_list(cls) -> list[MocketBaseRequest]:	7✔
109	return cls._requests	7✔
110
111	@classmethod	7✔
112	def remove_last_request(cls) -> None:	7✔
113	if cls.has_requests():	7✔
114	del cls._requests[-1]	7✔
115
116	@classmethod	7✔
117	def has_requests(cls) -> bool:	7✔
118	return bool(cls.request_list())	7✔
119
120	@classmethod	7✔
121	def get_namespace(cls) -> str \| None:	7✔
122	if not cls._record_storage:	7✔
NEW 123	return None	×
124	return cls._record_storage.namespace	7✔
125
126	@classmethod	7✔
127	def get_truesocket_recording_dir(cls) -> str \| None:	7✔
128	if not cls._record_storage:	7✔
NEW 129	return None	×
130	return str(cls._record_storage.directory)	7✔
131
132	@classmethod	7✔
133	def assert_fail_if_entries_not_served(cls) -> None:	7✔
134	"""Mocket checks that all entries have been served at least once."""
135	if not all(	7✔
136	entry.served_response for entry in itertools.chain(*cls._entries.values())
137	):
138	raise AssertionError("Some Mocket entries have not been served")	7✔

mindflayer / python-mocket / 12095558571

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