Add Mocket "strict mode" (#171)

* Add Mocket "strict mode".

Author: mindflayer

README.rst

@@ -141,9 +141,25 @@ Let's fire our example test::
 
     $ py.test example.py
 
+How to make Mocket fail when it tries to write to a real `socket`?
+==================================================================
+NEW!!! Sometimes you just want your tests to fail when they attempt to use the network.
 
-How to be sure that all the calls are properly mocked?
-======================================================
+.. code-block:: python
+
+    with Mocketizer(strict_mode=True):
+        with pytest.raises(StrictMocketException):
+            requests.get("https://duckduckgo.com/")
+
+    # OR
+
+    @mocketize(strict_mode=True)
+    def test_get():
+        with pytest.raises(StrictMocketException):
+            requests.get("https://duckduckgo.com/")
+
+How to be sure that all the Entry instances have been served?
+=============================================================
 Add this instruction at the end of the test execution:
 
 .. code-block:: python

mocket/__init__.py

@@ -3,4 +3,4 @@
 
 __all__ = ("async_mocketize", "mocketize", "Mocket", "MocketEntry", "Mocketizer")
 
-__version__ = "3.10.3"
+__version__ = "3.10.4"

mocket/async_mocket.py

@@ -2,16 +2,10 @@
 from .utils import get_mocketize
 
 
-async def wrapper(test, cls=Mocketizer, truesocket_recording_dir=None, *args, **kwargs):
-    instance = args[0] if args else None
-    namespace = None
-    if truesocket_recording_dir:
-        namespace = Mocketizer.get_namespace(test, instance)
-    async with cls(
-        instance,
-        namespace=namespace,
-        truesocket_recording_dir=truesocket_recording_dir,
-    ):
+async def wrapper(
+    test, truesocket_recording_dir=None, strict_mode=False, *args, **kwargs
+):
+    async with Mocketizer.factory(test, truesocket_recording_dir, strict_mode, args):
         return await test(*args, **kwargs)
 
 

mocket/exceptions.py

@@ -0,0 +1,6 @@
+class MocketException(Exception):
+    pass
+
+
+class StrictMocketException(MocketException):
+    pass

mocket/mocket.py

@@ -18,7 +18,15 @@
 from urllib3.util.ssl_ import wrap_socket as urllib3_wrap_socket
 
 from .compat import basestring, byte_type, decode_from_bytes, encode_to_bytes, text_type
-from .utils import SSL_PROTOCOL, MocketSocketCore, get_mocketize, hexdump, hexload
+from .exceptions import StrictMocketException
+from .utils import (
+    SSL_PROTOCOL,
+    MocketMode,
+    MocketSocketCore,
+    get_mocketize,
+    hexdump,
+    hexload,
+)
 
 xxh32 = None
 try:
@@ -286,6 +294,9 @@ def recv(self, buffersize, flags=None):
         raise exc
 
     def true_sendall(self, data, *args, **kwargs):
+        if MocketMode().STRICT:
+            raise StrictMocketException("Mocket tried to use the real `socket` module.")
+
         req = decode_from_bytes(data)
         # make request unique again
         req_signature = _hash_request(hasher, req)
@@ -597,20 +608,17 @@ def get_response(self):
 
 
 class Mocketizer:
-    def __init__(self, instance=None, namespace=None, truesocket_recording_dir=None):
+    def __init__(
+        self,
+        instance=None,
+        namespace=None,
+        truesocket_recording_dir=None,
+        strict_mode=False,
+    ):
         self.instance = instance
         self.truesocket_recording_dir = truesocket_recording_dir
         self.namespace = namespace or text_type(id(self))
-
-    @staticmethod
-    def get_namespace(test, instance):
-        return ".".join(
-            (
-                instance.__class__.__module__,
-                instance.__class__.__name__,
-                test.__name__,
-            )
-        )
+        MocketMode().STRICT = strict_mode
 
     def enter(self):
         Mocket.enable(
@@ -639,22 +647,34 @@ async def __aenter__(self, *args, **kwargs):
     async def __aexit__(self, *args, **kwargs):
         self.exit()
 
-    def check_and_call(self, method):
-        method = getattr(self.instance, method, None)
+    def check_and_call(self, method_name):
+        method = getattr(self.instance, method_name, None)
         if callable(method):
             method()
 
+    @staticmethod
+    def factory(test, truesocket_recording_dir, strict_mode, args):
+        instance = args[0] if args else None
+        namespace = None
+        if truesocket_recording_dir:
+            namespace = ".".join(
+                (
+                    instance.__class__.__module__,
+                    instance.__class__.__name__,
+                    test.__name__,
+                )
+            )
 
-def wrapper(test, cls=Mocketizer, truesocket_recording_dir=None, *args, **kwargs):
-    instance = args[0] if args else None
-    namespace = None
-    if truesocket_recording_dir:
-        namespace = Mocketizer.get_namespace(test, instance)
-    with cls(
-        instance,
-        namespace=namespace,
-        truesocket_recording_dir=truesocket_recording_dir,
-    ):
+        return Mocketizer(
+            instance,
+            namespace=namespace,
+            truesocket_recording_dir=truesocket_recording_dir,
+            strict_mode=strict_mode,
+        )
+
+
+def wrapper(test, truesocket_recording_dir=None, strict_mode=False, *args, **kwargs):
+    with Mocketizer.factory(test, truesocket_recording_dir, strict_mode, args):
         return test(*args, **kwargs)
 
 

mocket/utils.py

@@ -42,3 +42,11 @@ def get_mocketize(wrapper_):
     if decorator.__version__ < "5":  # pragma: no cover
         return decorator.decorator(wrapper_)
     return decorator.decorator(wrapper_, kwsyntax=True)
+
+
+class MocketMode:
+    __shared_state = {}
+    STRICT = None
+
+    def __init__(self):
+        self.__dict__ = self.__shared_state

tests/main/test_mode.py

@@ -0,0 +1,28 @@
+import pytest
+import requests
+
+from mocket import Mocketizer, mocketize
+from mocket.exceptions import StrictMocketException
+
+
+@mocketize(strict_mode=True)
+def test_strict_mode_fails():
+    url = "https://httpbin.org/ip"
+
+    with pytest.raises(StrictMocketException):
+        requests.get(url)
+
+
+@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
+def test_intermittent_strict_mode():
+    url = "https://httpbin.org/ip"
+
+    with Mocketizer(strict_mode=False):
+        requests.get(url)
+
+    with Mocketizer(strict_mode=True):
+        with pytest.raises(StrictMocketException):
+            requests.get(url)
+
+    with Mocketizer(strict_mode=False):
+        requests.get(url)