New mocket method for checking if all entries have been served (#137)

* Mocket method for checking if all entries have been served at least once.

Author: mindflayer

mocket/__init__.py

@@ -2,4 +2,4 @@
 
 __all__ = ("mocketize", "Mocket", "MocketEntry", "Mocketizer")
 
-__version__ = "3.9.35"
+__version__ = "3.9.36"

mocket/mocket.py

@@ -3,6 +3,7 @@
 import errno
 import hashlib
 import io
+import itertools
 import json
 import os
 import select
@@ -16,13 +17,7 @@
 from urllib3.util.ssl_ import ssl_wrap_socket as urllib3_ssl_wrap_socket
 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 .compat import basestring, byte_type, decode_from_bytes, encode_to_bytes, text_type
 from .utils import SSL_PROTOCOL, MocketSocketCore, hexdump, hexload, wrap_ssl_socket
 
 xxh32 = None
@@ -517,6 +512,13 @@ def get_namespace(cls):
     def get_truesocket_recording_dir(cls):
         return cls._truesocket_recording_dir
 
+    @classmethod
+    def assert_fail_if_entries_not_served(cls):
+        """ Mocket checks that all entries have been served at least once. """
+        assert all(
+            entry._served for entry in itertools.chain(*cls._entries.values())
+        ), "Some Mocket entries have not been served"
+
 
 class MocketEntry(object):
     class Response(byte_type):
@@ -526,8 +528,11 @@ def data(self):
 
     request_cls = str
     response_cls = Response
+    responses = None
+    _served = None
 
     def __init__(self, location, responses):
+        self._served = False
         self.location = location
         self.response_index = 0
 
@@ -536,19 +541,18 @@ def __init__(self, location, responses):
         ):
             responses = [responses]
 
-        lresponses = []
+        self.responses = []
         for r in responses:
             if isinstance(r, BaseException):
                 pass
             elif not getattr(r, "data", False):
                 if isinstance(r, text_type):
                     r = encode_to_bytes(r)
                 r = self.response_cls(r)
-            lresponses.append(r)
+            self.responses.append(r)
         else:
             if not responses:
-                lresponses = [self.response_cls(encode_to_bytes(""))]
-        self.responses = lresponses
+                self.responses = [self.response_cls(encode_to_bytes(""))]
 
     def can_handle(self, data):
         return True
@@ -562,6 +566,8 @@ def get_response(self):
         if self.response_index < len(self.responses) - 1:
             self.response_index += 1
 
+        self._served = True
+
         if isinstance(response, BaseException):
             raise response
 

tests/main/test_http.py

@@ -363,3 +363,22 @@ def test_sockets(self):
 
         # Proof that worked.
         self.assertEqual(Mocket.last_request().body, '{"hello": "world"}')
+
+    @mocketize
+    def test_fail_because_entry_not_served(self):
+        url = "http://github.com/fluidicon.png"
+        Entry.single_register(Entry.GET, url)
+        Entry.single_register(Entry.GET, "http://github.com/fluidicon.jpg")
+        requests.get(url)
+        with self.assertRaises(AssertionError):
+            Mocket.assert_fail_if_entries_not_served()
+
+    @mocketize
+    def test_does_not_fail_because_all_entries_are_served(self):
+        url = "http://github.com/fluidicon.png"
+        second_url = "http://github.com/fluidicon.jpg"
+        Entry.single_register(Entry.GET, url)
+        Entry.single_register(Entry.GET, second_url)
+        requests.get(url)
+        requests.get(second_url)
+        Mocket.assert_fail_if_entries_not_served()