diff --git a/doc/api.rst b/doc/api.rst index d3a0573..c52ab63 100644 --- a/doc/api.rst +++ b/doc/api.rst @@ -69,6 +69,12 @@ HeaderValueMatcher .. autoclass:: HeaderValueMatcher :members: +CaseInsensitiveHeaderValueMatcher +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + .. autoclass:: CaseInsensitiveHeaderValueMatcher + :members: + URIPattern ~~~~~~~~~~ diff --git a/doc/howto.rst b/doc/howto.rst index 04a5609..a000961 100644 --- a/doc/howto.rst +++ b/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, diff --git a/pytest_httpserver/__init__.py b/pytest_httpserver/__init__.py index e81e58a..c3d2ef4 100644 --- a/pytest_httpserver/__init__.py +++ b/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 diff --git a/pytest_httpserver/blocking_httpserver.py b/pytest_httpserver/blocking_httpserver.py index cb42795..abe1fc6 100644 --- a/pytest_httpserver/blocking_httpserver.py +++ b/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. diff --git a/pytest_httpserver/httpserver.py b/pytest_httpserver/httpserver.py index fdb3a3f..90cfe01 100644 --- a/pytest_httpserver/httpserver.py +++ b/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 diff --git a/tests/test_blocking_httpserver.py b/tests/test_blocking_httpserver.py index 1f5f277..360fa6a 100644 --- a/tests/test_blocking_httpserver.py +++ b/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", diff --git a/tests/test_headers.py b/tests/test_headers.py index 92439e5..e4947c6 100644 --- a/tests/test_headers.py +++ b/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"