Add opt-in case-insensitive header value matcher (#524)

Author: 00200200

doc/api.rst

@@ -69,6 +69,12 @@ HeaderValueMatcher
     .. autoclass:: HeaderValueMatcher
         :members:
 
+CaseInsensitiveHeaderValueMatcher
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+    .. autoclass:: CaseInsensitiveHeaderValueMatcher
+        :members:
+
 URIPattern
 ~~~~~~~~~~
 

doc/howto.rst

@@ -202,6 +202,47 @@ In case you don't want to change the defaults, you can provide the
 .. literalinclude :: ../tests/examples/test_howto_header_value_matcher.py
    :language: python
 
+Case-insensitive comparator names
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+``HeaderValueMatcher`` selects comparators by exact header name and remains the
+default. To select them regardless of header-name casing, explicitly pass a
+``CaseInsensitiveHeaderValueMatcher`` to an expectation or use ``bake()``:
+
+.. code:: python
+
+    from pytest_httpserver import CaseInsensitiveHeaderValueMatcher
+
+    server = httpserver.bake(header_value_matcher=CaseInsensitiveHeaderValueMatcher())
+    server.expect_request(
+        "/", headers={"authorization": 'Digest a="1", b="2"'}
+    ).respond_with_data("OK")
+
+This parses ``Authorization`` values for any casing of that header name. Other
+header values still use exact, case-sensitive equality. Only expectations
+registered through ``server`` use this matcher; the original ``httpserver``
+retains its defaults.
+
+Custom comparator keys are also case-insensitive:
+
+.. code:: python
+
+    def compare_values(actual: str | None, expected: str) -> bool:
+        return actual is not None and actual.lower() == expected.lower()
+
+
+    matcher = CaseInsensitiveHeaderValueMatcher({"x-CUSTOM": compare_values})
+    httpserver.expect_request(
+        "/custom", headers={"X-Custom": "value"}, header_value_matcher=matcher
+    ).respond_with_data("OK")
+
+The supplied mapping replaces the built-in comparators and is copied when the
+matcher is created. If keys differ only in case, the last entry wins. A plain
+mapping raises ``NoMethodFoundForMatchingHeaderValueError`` for unregistered
+names; a ``defaultdict`` can supply a fallback comparator without caching those
+names. Create a new matcher to use a changed mapping. Instances do not read or
+modify ``HeaderValueMatcher.DEFAULT_MATCHERS`` and do not share a registry.
+
 Using custom request handler
 ----------------------------
 In the case the response is not static, for example it depends on the request,

pytest_httpserver/__init__.py

@@ -9,6 +9,7 @@
     "BakedHTTPServer",
     "BlockingHTTPServer",
     "BlockingRequestHandler",
+    "CaseInsensitiveHeaderValueMatcher",
     "Error",
     "HTTPServer",
     "HTTPServerError",
@@ -26,6 +27,7 @@
 from .blocking_httpserver import BlockingRequestHandler
 from .httpserver import METHOD_ALL
 from .httpserver import URI_DEFAULT
+from .httpserver import CaseInsensitiveHeaderValueMatcher
 from .httpserver import Error
 from .httpserver import HeaderValueMatcher
 from .httpserver import HTTPServer

pytest_httpserver/blocking_httpserver.py

@@ -5,9 +5,9 @@
 from typing import TYPE_CHECKING
 from typing import Any
 
+from pytest_httpserver.httpserver import HVMATCHER_T
 from pytest_httpserver.httpserver import METHOD_ALL
 from pytest_httpserver.httpserver import UNDEFINED
-from pytest_httpserver.httpserver import HeaderValueMatcher
 from pytest_httpserver.httpserver import HTTPServerBase
 from pytest_httpserver.httpserver import QueryMatcher
 from pytest_httpserver.httpserver import RequestHandlerBase
@@ -77,7 +77,7 @@ def assert_request(
         data_encoding: str = "utf-8",
         headers: Mapping[str, str] | None = None,
         query_string: QueryMatcher | str | bytes | Mapping[str, str] | None = None,
-        header_value_matcher: HeaderValueMatcher | None = None,
+        header_value_matcher: HVMATCHER_T | None = None,
         json: Any = UNDEFINED,
         timeout: int = 30,
     ) -> BlockingRequestHandler:
@@ -104,7 +104,7 @@ def assert_request(
             specified in the request. If multiple values specified for a given key, the first
             value will be used. If multiple values needed to be handled, use ``MultiDict``
             object from werkzeug.
-        :param header_value_matcher: :py:class:`HeaderValueMatcher` that matches values of headers.
+        :param header_value_matcher: callable that matches values of headers.
         :param json: a python object (eg. a dict) whose value will be compared to the request body after it
             is loaded as json. If load fails, this matcher will be failed also. *Content-Type* is not checked.
             If that's desired, add it to the headers parameter.

pytest_httpserver/httpserver.py

@@ -177,6 +177,43 @@ def __call__(self, header_name: str, actual: str | None, expected: str) -> bool:
 )
 
 
+class CaseInsensitiveHeaderValueMatcher:
+    """
+    Match header values using case-insensitive comparator names.
+
+    This matcher must be passed explicitly as ``header_value_matcher``. By
+    default, it parses Authorization values and compares other values exactly,
+    independently of ``HeaderValueMatcher.DEFAULT_MATCHERS``.
+
+    :param matchers: optional mapping of header names to comparator functions,
+        replacing the built-in comparators. Keys are copied and lowercased at
+        construction; the last entry wins if names differ only in case. Later
+        changes to the input mapping do not affect this instance. A defaultdict's
+        factory supplies comparators for missing names without caching them;
+        otherwise missing names raise NoMethodFoundForMatchingHeaderValueError.
+    """
+
+    def __init__(self, matchers: Mapping[str, Callable[[str | None, str], bool]] | None = None) -> None:
+        if matchers is None:
+            matchers = defaultdict(
+                lambda: HeaderValueMatcher.default_header_value_matcher,
+                {"Authorization": HeaderValueMatcher.authorization_header_value_matcher},
+            )
+        self._matchers = {name.lower(): matcher for name, matcher in matchers.items()}
+        self._default_factory = matchers.default_factory if isinstance(matchers, defaultdict) else None
+
+    def __call__(self, header_name: str, actual: str | None, expected: str) -> bool:
+        try:
+            matcher = self._matchers[header_name.lower()]
+        except KeyError:
+            if self._default_factory is None:
+                raise NoMethodFoundForMatchingHeaderValueError(
+                    "No method found for matching header value: {}".format(header_name)
+                )
+            matcher = self._default_factory()
+        return matcher(actual, expected)
+
+
 class QueryMatcher(abc.ABC):
     """
     Abstract class for QueryMatchers

tests/test_blocking_httpserver.py

@@ -7,6 +7,7 @@
 import requests
 
 from pytest_httpserver import BlockingHTTPServer
+from pytest_httpserver import CaseInsensitiveHeaderValueMatcher
 
 
 @contextmanager
@@ -68,6 +69,22 @@ def test_behave_workflow(httpserver: BlockingHTTPServer):
         then_the_response_is_got_from(server_connection, response)
 
 
+def test_case_insensitive_header_matcher(httpserver: BlockingHTTPServer):
+    request = dict(
+        method="GET",
+        url=httpserver.url_for("/"),
+        headers={"Authorization": 'Digest a="1", b="2"'},
+    )
+    with when_a_request_is_being_sent_to_the_server(request) as server_connection:
+        client_connection = httpserver.assert_request(
+            uri="/",
+            headers={"authorization": 'Digest b="2", a="1"'},
+            header_value_matcher=CaseInsensitiveHeaderValueMatcher(),
+        )
+        client_connection.respond_with_data("OK")
+        assert server_connection.get(timeout=9).text == "OK"
+
+
 def test_raises_assertion_error_when_request_does_not_match(httpserver: BlockingHTTPServer):
     request = dict(
         method="GET",

tests/test_headers.py

@@ -1,15 +1,126 @@
 import http.client
+from collections import defaultdict
 
+import pytest
 import requests
 from werkzeug.datastructures import Headers
 from werkzeug.http import parse_dict_header
 
+from pytest_httpserver import CaseInsensitiveHeaderValueMatcher
 from pytest_httpserver import HTTPServer
+from pytest_httpserver import RequestMatcher
 from pytest_httpserver.httpserver import HeaderValueMatcher
+from pytest_httpserver.httpserver import NoMethodFoundForMatchingHeaderValueError
 
 
-def test_custom_headers(httpserver: HTTPServer):
-    headers_with_values_in_direct_order = {"Custom": 'Scheme key0="value0", key1="value1"'}
+def test_case_insensitive_header_matcher_fallback_does_not_cache_header_names():
+    comparators = iter(
+        [HeaderValueMatcher.default_header_value_matcher, lambda actual, expected: actual == expected.upper()]
+    )
+    matchers = defaultdict(lambda: next(comparators))
+    matcher = CaseInsensitiveHeaderValueMatcher(matchers)
+
+    assert matcher("x-custom", "value", "value")
+    assert matcher("x-custom", "VALUE", "value")
+    assert not matchers
+
+
+@pytest.mark.parametrize("mapping_class", [dict, defaultdict])
+def test_case_insensitive_header_matcher_without_default(mapping_class):
+    matchers = mapping_class()
+    matchers["x-custom"] = HeaderValueMatcher.default_header_value_matcher
+    matcher = CaseInsensitiveHeaderValueMatcher(matchers)
+
+    assert matcher("X-CUSTOM", "value", "value")
+    assert not matcher("X-CUSTOM", "VALUE", "value")
+    with pytest.raises(NoMethodFoundForMatchingHeaderValueError, match="X-Missing"):
+        matcher("X-Missing", "value", "value")
+
+
+def test_case_insensitive_header_matcher_copies_comparators():
+    matchers = {"X-Custom": HeaderValueMatcher.default_header_value_matcher}
+    first = CaseInsensitiveHeaderValueMatcher(matchers)
+    matchers["X-Custom"] = lambda actual, expected: actual == expected.upper()
+    second = CaseInsensitiveHeaderValueMatcher(matchers)
+
+    assert first("X-CUSTOM", "value", "value")
+    assert not first("X-CUSTOM", "VALUE", "value")
+    assert second("x-custom", "VALUE", "value")
+    assert not second("x-custom", "value", "value")
+
+
+def test_case_insensitive_header_matcher_duplicate_names():
+    matcher = CaseInsensitiveHeaderValueMatcher(
+        {
+            "X-Custom": HeaderValueMatcher.default_header_value_matcher,
+            "x-CUSTOM": lambda actual, expected: actual == expected.upper(),
+        }
+    )
+
+    assert matcher("x-custom", "VALUE", "value")
+    assert not matcher("X-Custom", "value", "value")
+
+
+def test_case_insensitive_header_matcher_independent_defaults(monkeypatch):
+    legacy_matchers = {"X-Custom": lambda actual, expected: actual == expected.upper()}
+    monkeypatch.setattr(HeaderValueMatcher, "DEFAULT_MATCHERS", legacy_matchers)
+    customized = CaseInsensitiveHeaderValueMatcher(legacy_matchers)
+
+    assert customized("x-custom", "VALUE", "value")
+    for matcher in (CaseInsensitiveHeaderValueMatcher(), CaseInsensitiveHeaderValueMatcher()):
+        assert matcher("x-custom", "value", "value")
+        assert not matcher("x-custom", "VALUE", "value")
+        assert not matcher("x-custom", None, "value")
+        assert matcher("authorization", 'Digest a="1", b="2"', 'Digest b="2", a="1"')
+        assert not matcher("AUTHORIZATION", None, 'Digest a="1"')
+
+    assert list(legacy_matchers) == ["X-Custom"]
+    default_matcher = RequestMatcher("/").header_value_matcher
+    assert isinstance(default_matcher, HeaderValueMatcher)
+    assert default_matcher.matchers is legacy_matchers
+
+
+def test_legacy_header_matcher_preserves_case_sensitive_lookup():
+    matcher = HeaderValueMatcher({"X-Custom": HeaderValueMatcher.default_header_value_matcher})
+
+    assert matcher("X-Custom", "value", "value")
+    with pytest.raises(NoMethodFoundForMatchingHeaderValueError, match="x-custom"):
+        matcher("x-custom", "value", "value")
+
+
+def test_legacy_header_matcher_preserves_defaultdict_lookup():
+    matchers = defaultdict(lambda: HeaderValueMatcher.default_header_value_matcher)
+    matcher = HeaderValueMatcher(matchers)
+
+    assert matcher("x-custom", "value", "value")
+    assert matchers["x-custom"] is HeaderValueMatcher.default_header_value_matcher
+    matchers["X-Custom"] = lambda actual, expected: actual == expected.upper()
+    assert not matcher("x-custom", "VALUE", "value")
+
+
+def test_case_insensitive_header_matcher_bake_is_opt_in(httpserver: HTTPServer):
+    matcher = CaseInsensitiveHeaderValueMatcher(
+        {"x-CUSTOM": lambda actual, expected: actual is not None and actual.lower() == expected.lower()}
+    )
+    server = httpserver.bake(header_value_matcher=matcher)
+    server.expect_request("/opt-in", headers={"X-Custom": "value"}).respond_with_data("OK")
+    httpserver.expect_request("/default", headers={"X-Custom": "value"}).respond_with_data("OK")
+
+    assert requests.get(server.url_for("/opt-in"), headers={"X-Custom": "VALUE"}).status_code == 200
+    assert requests.get(httpserver.url_for("/default"), headers={"X-Custom": "VALUE"}).status_code == 500
+
+
+@pytest.mark.parametrize(
+    ("matcher_class", "comparator_name", "header_name"),
+    [
+        (HeaderValueMatcher, "Custom", "Custom"),
+        (CaseInsensitiveHeaderValueMatcher, "Custom", "Custom"),
+        (CaseInsensitiveHeaderValueMatcher, "CuStOm", "custom"),
+        (CaseInsensitiveHeaderValueMatcher, "custom", "CUSTOM"),
+    ],
+)
+def test_custom_headers(httpserver: HTTPServer, matcher_class, comparator_name: str, header_name: str):
+    headers_with_values_in_direct_order = {header_name: 'Scheme key0="value0", key1="value1"'}
     httpserver.expect_request(uri="/", headers=headers_with_values_in_direct_order).respond_with_data("OK")
     response = requests.get(httpserver.url_for("/"), headers=headers_with_values_in_direct_order)
     assert response.status_code == 200
@@ -21,16 +132,16 @@ def test_custom_headers(httpserver: HTTPServer):
     assert response.status_code == 500
 
     # Define header_value_matcher that ignores the order of items in header value dict
-    def custom_header_value_matcher(actual: str, expected: str) -> bool:
+    def custom_header_value_matcher(actual: str | None, expected: str) -> bool:
+        if actual is None:
+            return False
         actual_scheme, _, actual_dict_str = actual.partition(" ")
         expected_scheme, _, expected_dict_str = expected.partition(" ")
         actual_dict = parse_dict_header(actual_dict_str)
         expected_dict = parse_dict_header(expected_dict_str)
         return actual_scheme == expected_scheme and actual_dict == expected_dict
 
-    matchers = HeaderValueMatcher.DEFAULT_MATCHERS.copy()  # type: ignore
-    matchers["Custom"] = custom_header_value_matcher
-    header_value_matcher = HeaderValueMatcher(matchers)
+    header_value_matcher = matcher_class({comparator_name: custom_header_value_matcher})
 
     httpserver.handlers.clear()
     httpserver.expect_request(
@@ -42,9 +153,22 @@ def custom_header_value_matcher(actual: str, expected: str) -> bool:
 
 
 # See https://en.wikipedia.org/wiki/Digest_access_authentication
-def test_authorization_headers(httpserver: HTTPServer):
+@pytest.mark.parametrize(
+    ("header_name", "case_insensitive"),
+    [
+        ("Authorization", False),
+        ("Authorization", True),
+        ("authorization", True),
+        ("AUTHORIZATION", True),
+        ("aUtHoRiZaTiOn", True),
+    ],
+)
+def test_authorization_headers(httpserver: HTTPServer, header_name: str, *, case_insensitive: bool):
+    server = (
+        httpserver.bake(header_value_matcher=CaseInsensitiveHeaderValueMatcher()) if case_insensitive else httpserver
+    )
     headers_with_values_in_direct_order = {
-        "Authorization": (
+        header_name: (
             'Digest username="Mufasa",'
             'realm="testrealm@host.com",'
             'nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093",'
@@ -56,8 +180,8 @@ def test_authorization_headers(httpserver: HTTPServer):
             'opaque="5ccc069c403ebaf9f0171e9517f40e41"'
         )
     }
-    httpserver.expect_request(uri="/", headers=headers_with_values_in_direct_order).respond_with_data("OK")
-    response = requests.get(httpserver.url_for("/"), headers=headers_with_values_in_direct_order)
+    server.expect_request(uri="/", headers=headers_with_values_in_direct_order).respond_with_data("OK")
+    response = requests.get(server.url_for("/"), headers=headers_with_values_in_direct_order)
     assert response.status_code == 200
     assert response.text == "OK"
 
@@ -74,7 +198,7 @@ def test_authorization_headers(httpserver: HTTPServer):
             'opaque="5ccc069c403ebaf9f0171e9517f40e41"'
         )
     }
-    response = requests.get(httpserver.url_for("/"), headers=headers_with_values_in_modified_order)
+    response = requests.get(server.url_for("/"), headers=headers_with_values_in_modified_order)
     assert response.status_code == 200
     assert response.text == "OK"