Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions mocket/decorators/mocketizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,9 @@ def __init__(
self.instance = instance
self.truesocket_recording_dir = truesocket_recording_dir
self.namespace = namespace or str(id(self))
MocketMode().STRICT = strict_mode
MocketMode.STRICT = strict_mode
if strict_mode:
MocketMode().STRICT_ALLOWED = strict_mode_allowed or []
MocketMode.STRICT_ALLOWED = strict_mode_allowed or []
elif strict_mode_allowed:
raise ValueError(
"Allowed locations are only accepted when STRICT mode is active."
Expand Down
26 changes: 21 additions & 5 deletions mocket/mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from typing import NoReturn


class MocketMode:
class _MocketMode:
__shared_state: ClassVar[dict[str, Any]] = {}
STRICT: ClassVar = None
STRICT_ALLOWED: ClassVar = None
Expand All @@ -31,15 +31,31 @@ def is_allowed(self, location: str | tuple[str, int]) -> bool:
return host_allowed or location in self.STRICT_ALLOWED

@staticmethod
def raise_not_allowed() -> NoReturn:
def raise_not_allowed(
address: tuple[str, int] | None = None,
data: bytes | None = None,
) -> NoReturn:
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}"
msg = (
"Mocket tried to use the real `socket` module while STRICT mode was active."
)
if address:
host, port = address
msg += f"\nAttempted address: {host}:{port}"
if data:
from mocket.compat import decode_from_bytes

preview = decode_from_bytes(data).split("\r\n", 1)[0][:200]
msg += f"\nSent data: {preview}"

msg += f"\nRegistered entries:\n{formatted_entries}"
raise StrictMocketException(msg)


MocketMode = _MocketMode()
4 changes: 2 additions & 2 deletions mocket/socket.py
Original file line number Diff line number Diff line change
Expand Up @@ -228,8 +228,8 @@ def recv(self, buffersize: int, flags: int | None = None) -> bytes:
raise exc

def true_sendall(self, data: bytes, *args: Any, **kwargs: Any) -> bytes:
if not MocketMode().is_allowed(self._address):
MocketMode.raise_not_allowed()
if not MocketMode.is_allowed(self._address):
MocketMode.raise_not_allowed(self._address, data)

# try to get the response from recordings
if Mocket._record_storage:
Expand Down
6 changes: 4 additions & 2 deletions tests/test_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,8 @@ def test_strict_mode_error_message():
str(exc_info.value)
== """
Mocket tried to use the real `socket` module while STRICT mode was active.
Attempted address: httpbin.local:80
Sent data: GET /ip HTTP/1.1
Registered entries:
('httpbin.local', 80):
Entry(method='GET', schema='http', location=('httpbin.local', 80), path='/user.agent', query='')
Expand All @@ -67,5 +69,5 @@ def test_strict_mode_false_with_allowed_hosts():
@pytest.mark.parametrize("strict_mode_on", (False, True))
def test_strict_mode_allowed_or_not(strict_mode_on):
with Mocketizer(strict_mode=strict_mode_on):
assert MocketMode().is_allowed("foobar.com") is not strict_mode_on
assert MocketMode().is_allowed(("foobar.com", 443)) is not strict_mode_on
assert MocketMode.is_allowed("foobar.com") is not strict_mode_on
assert MocketMode.is_allowed(("foobar.com", 443)) is not strict_mode_on