Recording: support removing headers from recorded data

mocket/async_mocket.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from mocket.mocketizer import Mocketizer
 from mocket.utils import get_mocketize
 
@@ -7,11 +9,17 @@ async def wrapper(
     truesocket_recording_dir=None,
     strict_mode=False,
     strict_mode_allowed=None,
+    recording_ignored_headers: list[str] | None = None,
     *args,
     **kwargs,
 ):
     async with Mocketizer.factory(
-        test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args
+        test,
+        truesocket_recording_dir,
+        strict_mode,
+        strict_mode_allowed,
+        recording_ignored_headers,
+        args,
     ):
         return await test(*args, **kwargs)
 

mocket/inject.py

@@ -17,6 +17,7 @@
 def enable(
     namespace: str | None = None,
     truesocket_recording_dir: str | None = None,
+    recording_ignored_headers: list[str] | None = None,
 ) -> None:
     from mocket.mocket import Mocket
     from mocket.socket import (
@@ -33,6 +34,7 @@ def enable(
 
     Mocket._namespace = namespace
     Mocket._truesocket_recording_dir = truesocket_recording_dir
+    Mocket._recording_ignored_headers = recording_ignored_headers or []
 
     if truesocket_recording_dir and not os.path.isdir(truesocket_recording_dir):
         # JSON dumps will be saved here

mocket/mocket.py

@@ -22,6 +22,7 @@ class Mocket:
     _requests: ClassVar[list] = []
     _namespace: ClassVar[str] = str(id(_entries))
     _truesocket_recording_dir: ClassVar[str | None] = None
+    _recording_ignored_headers: list[str] | None = None
 
     enable = mocket.inject.enable
     disable = mocket.inject.disable
@@ -96,6 +97,10 @@ def get_namespace(cls) -> str:
     def get_truesocket_recording_dir(cls) -> str | None:
         return cls._truesocket_recording_dir
 
+    @classmethod
+    def get_recording_ignored_headers(cls):
+        return cls._recording_ignored_headers
+
     @classmethod
     def assert_fail_if_entries_not_served(cls) -> None:
         """Mocket checks that all entries have been served at least once."""

mocket/mocketizer.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from mocket.mocket import Mocket
 from mocket.mode import MocketMode
 from mocket.utils import get_mocketize
@@ -11,9 +13,11 @@ def __init__(
         truesocket_recording_dir=None,
         strict_mode=False,
         strict_mode_allowed=None,
+        recording_ignored_headers: list[str] | None = None,
     ):
         self.instance = instance
         self.truesocket_recording_dir = truesocket_recording_dir
+        self.recording_ignored_headers = recording_ignored_headers
         self.namespace = namespace or str(id(self))
         MocketMode().STRICT = strict_mode
         if strict_mode:
@@ -27,6 +31,7 @@ def enter(self):
         Mocket.enable(
             namespace=self.namespace,
             truesocket_recording_dir=self.truesocket_recording_dir,
+            recording_ignored_headers=self.recording_ignored_headers,
         )
         if self.instance:
             self.check_and_call("mocketize_setup")
@@ -57,7 +62,14 @@ def check_and_call(self, method_name):
             method()
 
     @staticmethod
-    def factory(test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args):
+    def factory(
+        test,
+        truesocket_recording_dir,
+        strict_mode,
+        strict_mode_allowed,
+        recording_ignored_headers,
+        args,
+    ):
         instance = args[0] if args else None
         namespace = None
         if truesocket_recording_dir:
@@ -75,6 +87,7 @@ def factory(test, truesocket_recording_dir, strict_mode, strict_mode_allowed, ar
             truesocket_recording_dir=truesocket_recording_dir,
             strict_mode=strict_mode,
             strict_mode_allowed=strict_mode_allowed,
+            recording_ignored_headers=recording_ignored_headers,
         )
 
 
@@ -83,11 +96,17 @@ def wrapper(
     truesocket_recording_dir=None,
     strict_mode=False,
     strict_mode_allowed=None,
+    recording_ignored_headers: list[str] | None = None,
     *args,
     **kwargs,
 ):
     with Mocketizer.factory(
-        test, truesocket_recording_dir, strict_mode, strict_mode_allowed, args
+        test,
+        truesocket_recording_dir,
+        strict_mode,
+        strict_mode_allowed,
+        recording_ignored_headers,
+        args,
     ):
         return test(*args, **kwargs)
 

mocket/socket.py

@@ -7,6 +7,7 @@
 import os
 import select
 import socket
+import re
 from json.decoder import JSONDecodeError
 from types import TracebackType
 from typing import Any, Type
@@ -314,6 +315,11 @@ def true_sendall(self, data: ReadableBuffer, *args: Any, **kwargs: Any) -> int:
 
             # dump the resulting dictionary to a JSON file
             if Mocket.get_truesocket_recording_dir():
+                # filter out unwanted headers
+                for header in Mocket.get_recording_ignored_headers():
+                    header_pattern = rf"{header}: .*\r\n"
+                    req = re.sub(header_pattern, "", req)
+
                 # update the dictionary with request and response lines
                 response_dict["request"] = req
                 response_dict["response"] = hexdump(encoded_response)

tests/test_http.py

@@ -314,6 +314,37 @@ def test_truesendall_with_dump_from_recording(self):
 
         self.assertEqual(len(responses["httpbin.local"]["80"].keys()), 2)
 
+    @mocketize(
+        truesocket_recording_dir=os.path.dirname(__file__),
+        recording_ignored_headers=["Authorization"],
+    )
+    def test_truesendall_with_dump_from_recording_and_ignored_headers(self):
+        requests.get(
+            "http://httpbin.local/ip",
+            headers={
+                "user-agent": "Fake-User-Agent",
+                "Accept-Encoding": "gzip, deflate, zstd",
+                "Authorization": "token",
+            },
+        )
+        requests.get(
+            "http://httpbin.local/gzip",
+            headers={
+                "user-agent": "Fake-User-Agent",
+                "Accept-Encoding": "gzip, deflate, zstd",
+                "Authorization": "token",
+            },
+        )
+
+        dump_filename = os.path.join(
+            Mocket.get_truesocket_recording_dir(), Mocket.get_namespace() + ".json"
+        )
+        with open(dump_filename) as f:
+            responses = json.load(f)
+
+        for key, value in responses["httpbin.local"]["80"].items():
+            self.assertNotIn("Authorization", value["request"])
+
     @mocketize
     def test_post_file_object(self):
         url = "http://github.com/fluidicon.png"