Merge pull request #214 from ento/strict-mode-bypass

Allow specifying strict mode exceptions

README.rst

@@ -169,6 +169,20 @@ NEW!!! Sometimes you just want your tests to fail when they attempt to use the n
         with pytest.raises(StrictMocketException):
             requests.get("https://duckduckgo.com/")
 
+You can specify exceptions as a list of hosts or host-port pairs.
+
+.. code-block:: python
+
+    with Mocketizer(strict_mode=True, strict_mode_allowed=["localhost", ("intake.ourmetrics.net", 443)]):
+        ...
+
+    # OR
+
+    @mocketize(strict_mode=True, strict_mode_allowed=["localhost", ("intake.ourmetrics.net", 443)])
+    def test_get():
+        ...
+
+
 How to be sure that all the Entry instances have been served?
 =============================================================
 Add this instruction at the end of the test execution:

mocket/async_mocket.py

@@ -3,9 +3,16 @@
 
 
 async def wrapper(
-    test, truesocket_recording_dir=None, strict_mode=False, *args, **kwargs
+    test,
+    truesocket_recording_dir=None,
+    strict_mode=False,
+    strict_mode_allowed=None,
+    *args,
+    **kwargs
 ):
-    async with Mocketizer.factory(test, truesocket_recording_dir, strict_mode, args):
+    async with Mocketizer.factory(
+        test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args
+    ):
         return await test(*args, **kwargs)
 
 

mocket/mocket.py

@@ -334,7 +334,21 @@ def recv(self, buffersize, flags=None):
 
     def true_sendall(self, data, *args, **kwargs):
         if MocketMode().STRICT:
-            raise StrictMocketException("Mocket tried to use the real `socket` module.")
+            if not MocketMode().allowed((self._host, self._port)):
+                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 is active.\n"
+                    f"Registered entries:\n{formatted_entries}"
+                )
 
         req = decode_from_bytes(data)
         # make request unique again
@@ -642,6 +656,9 @@ def __init__(self, location, responses):
                     r = self.response_cls(r)
                 self.responses.append(r)
 
+    def __repr__(self):
+        return "{}(location={})".format(self.__class__.__name__, self.location)
+
     @staticmethod
     def can_handle(data):
         return True
@@ -670,11 +687,13 @@ def __init__(
         namespace=None,
         truesocket_recording_dir=None,
         strict_mode=False,
+        strict_mode_allowed=None,
     ):
         self.instance = instance
         self.truesocket_recording_dir = truesocket_recording_dir
         self.namespace = namespace or text_type(id(self))
         MocketMode().STRICT = strict_mode
+        MocketMode().STRICT_ALLOWED = strict_mode_allowed
 
     def enter(self):
         Mocket.enable(
@@ -709,7 +728,7 @@ def check_and_call(self, method_name):
             method()
 
     @staticmethod
-    def factory(test, truesocket_recording_dir, strict_mode, args):
+    def factory(test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args):
         instance = args[0] if args else None
         namespace = None
         if truesocket_recording_dir:
@@ -726,11 +745,21 @@ def factory(test, truesocket_recording_dir, strict_mode, args):
             namespace=namespace,
             truesocket_recording_dir=truesocket_recording_dir,
             strict_mode=strict_mode,
+            strict_mode_allowed=strict_mode_allowed,
         )
 
 
-def wrapper(test, truesocket_recording_dir=None, strict_mode=False, *args, **kwargs):
-    with Mocketizer.factory(test, truesocket_recording_dir, strict_mode, args):
+def wrapper(
+    test,
+    truesocket_recording_dir=None,
+    strict_mode=False,
+    strict_mode_allowed=None,
+    *args,
+    **kwargs
+):
+    with Mocketizer.factory(
+        test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args
+    ):
         return test(*args, **kwargs)
 
 

mocket/mockhttp.py

@@ -175,6 +175,18 @@ def __init__(self, uri, method, responses, match_querystring=True):
         self._sent_data = b""
         self._match_querystring = match_querystring
 
+    def __repr__(self):
+        return (
+            "{}(method={!r}, schema={!r}, location={!r}, path={!r}, query={!r})".format(
+                self.__class__.__name__,
+                self.method,
+                self.schema,
+                self.location,
+                self.path,
+                self.query,
+            )
+        )
+
     def collect(self, data):
         consume_response = True
 

mocket/utils.py

@@ -47,6 +47,18 @@ def get_mocketize(wrapper_):
 class MocketMode:
     __shared_state = {}
     STRICT = None
+    STRICT_ALLOWED = None
 
     def __init__(self):
         self.__dict__ = self.__shared_state
+
+    def allowed(self, location):
+        if not self.STRICT_ALLOWED:
+            return False
+        host, port = location
+        for allowed in self.STRICT_ALLOWED:
+            if isinstance(allowed, str) and host == allowed:
+                return True
+            elif location == allowed:
+                return True
+        return False

tests/main/test_mode.py

@@ -3,6 +3,7 @@
 
 from mocket import Mocketizer, mocketize
 from mocket.exceptions import StrictMocketException
+from mocket.mockhttp import Entry, Response
 
 
 @mocketize(strict_mode=True)
@@ -26,3 +27,33 @@ def test_intermittent_strict_mode():
 
     with Mocketizer(strict_mode=False):
         requests.get(url)
+
+
+@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
+def test_strict_mode_exceptions():
+    url = "http://httpbin.local/ip"
+
+    with Mocketizer(strict_mode=True, strict_mode_allowed=["httpbin.local"]):
+        requests.get(url)
+
+    with Mocketizer(strict_mode=True, strict_mode_allowed=[("httpbin.local", 80)]):
+        requests.get(url)
+
+
+def test_strict_mode_error_message():
+    url = "http://httpbin.local/ip"
+
+    Entry.register(Entry.GET, "http://httpbin.local/user.agent", Response(status=404))
+
+    with Mocketizer(strict_mode=True):
+        with pytest.raises(StrictMocketException) as exc_info:
+            requests.get(url)
+        assert (
+            str(exc_info.value)
+            == """
+Mocket tried to use the real `socket` module while strict mode is active.
+Registered entries:
+  ('httpbin.local', 80):
+    Entry(method='GET', schema='http', location=('httpbin.local', 80), path='/user.agent', query='')
+""".strip()
+        )