Add MockHTTPResponse for library-agnostic HTTP response mocking

MockHTTPResponse implements HTTPResponseProtocol without depending on
requests or httpx. It supports the full response API (.json(), .text,
.content, .status_code, .headers, .cookies, .elapsed), streaming via
iter_content() and iter_lines(), raise_for_status() using HTTPStatusError,
and context manager usage.

Demonstrates usage by migrating test_authtoken.py from MockResponse to
MockHTTPResponse — a drop-in replacement with protocol compliance.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: mwdd146980

datadog_checks_base/changelog.d/22677.added

@@ -0,0 +1 @@
+Add MockHTTPResponse for library-agnostic HTTP response mocking in tests.

datadog_checks_base/datadog_checks/base/utils/http_testing.py

@@ -0,0 +1,127 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Testing utilities for HTTP client mocking."""
+
+import json
+from io import BytesIO
+from typing import Any, Iterator
+
+__all__ = ['MockHTTPResponse']
+
+
+class MockHTTPResponse:
+    """Library-agnostic mock HTTP response implementing HTTPResponseProtocol."""
+
+    def __init__(
+        self,
+        content: str | bytes = '',
+        status_code: int = 200,
+        headers: dict[str, str] | None = None,
+        json_data: dict[str, Any] | None = None,
+        file_path: str | None = None,
+        cookies: dict[str, str] | None = None,
+        elapsed_seconds: float = 0.1,
+        normalize_content: bool = True,
+    ):
+        if json_data is not None:
+            content = json.dumps(json_data)
+            if headers is None:
+                headers = {}
+            headers.setdefault('Content-Type', 'application/json')
+        elif file_path is not None:
+            # Open in binary mode to handle both text and binary files correctly
+            # This prevents encoding errors and platform-specific newline translation
+            with open(file_path, 'rb') as f:
+                content = f.read()
+
+        if normalize_content and (
+            (isinstance(content, str) and content.startswith('\n'))
+            or (isinstance(content, bytes) and content.startswith(b'\n'))
+        ):
+            content = content[1:]
+
+        self._content = content.encode('utf-8') if isinstance(content, str) else content
+        self.status_code = status_code
+        self.headers = headers or {}
+        self.cookies = cookies or {}
+
+        from datetime import timedelta
+
+        self.elapsed = timedelta(seconds=elapsed_seconds)
+        self._stream = BytesIO(self._content)
+        self._stream_consumed = False
+
+        def mock_getpeercert(self, binary_form=False):
+            return b'mock-cert' if binary_form else {}
+
+        self.raw = type(
+            'MockRaw',
+            (),
+            {
+                'connection': type(
+                    'MockConnection', (), {'sock': type('MockSocket', (), {'getpeercert': mock_getpeercert})()}
+                )()
+            },
+        )()
+
+    @property
+    def content(self) -> bytes:
+        return self._content
+
+    @property
+    def text(self) -> str:
+        return self._content.decode('utf-8')
+
+    def json(self, **kwargs: Any) -> Any:
+        return json.loads(self.text, **kwargs)
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            from datadog_checks.base.utils.http_exceptions import HTTPStatusError
+
+            message = (
+                f'{self.status_code} Client Error' if self.status_code < 500 else f'{self.status_code} Server Error'
+            )
+            raise HTTPStatusError(message, response=self)
+
+    def iter_content(self, chunk_size: int | None = None, decode_unicode: bool = False) -> Iterator[bytes | str]:
+        # chunk_size=None means return the entire content as a single chunk (matches requests behavior)
+        chunk_size = chunk_size if chunk_size is not None else len(self._content) or 1
+        if not self._stream_consumed:
+            self._stream.seek(0)
+        while chunk := self._stream.read(chunk_size):
+            # Decode to string when decode_unicode=True (matches requests behavior)
+            yield chunk.decode('utf-8') if decode_unicode else chunk
+        self._stream_consumed = True
+
+    def iter_lines(
+        self, chunk_size: int | None = None, decode_unicode: bool = False, delimiter: bytes | str | None = None
+    ) -> Iterator[bytes | str]:
+        # Handle string delimiter by converting to bytes
+        if isinstance(delimiter, str):
+            delimiter = delimiter.encode('utf-8')
+        delimiter = delimiter or b'\n'
+
+        if not self._stream_consumed:
+            self._stream.seek(0)
+
+        lines = self._stream.read().split(delimiter)
+        # bytes.split() produces a trailing empty element when content ends with the
+        # delimiter (e.g. b'a\nb\n'.split(b'\n') == [b'a', b'b', b'']). requests uses
+        # splitlines() for the default case which does not have this behavior, so we
+        # strip the trailing empty element to match.
+        if lines and not lines[-1]:
+            lines.pop()
+        for line in lines:
+            # Decode to string when decode_unicode=True (matches requests behavior)
+            yield line.decode('utf-8') if decode_unicode else line
+
+        self._stream_consumed = True
+
+    def __enter__(self) -> 'MockHTTPResponse':
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool | None:
+        self._stream.close()
+        return None

datadog_checks_base/tests/base/utils/http/test_authtoken.py

@@ -10,10 +10,13 @@
 
 from datadog_checks.base import ConfigurationError
 from datadog_checks.base.utils.http import DEFAULT_EXPIRATION, RequestsWrapper
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.base.utils.time import get_timestamp
+
+# Note: MockHTTPResponse is a drop-in replacement for dev.http.MockResponse
+# with protocol compliance and httpx compatibility
 from datadog_checks.dev import TempDir
 from datadog_checks.dev.fs import read_file, write_file
-from datadog_checks.dev.http import MockResponse
 
 from .common import DEFAULT_OPTIONS, FIXTURE_PATH
 
@@ -607,14 +610,14 @@ def login(*args, **kwargs):
                 assert isinstance(decoded['exp'], int)
                 assert abs(decoded['exp'] - (get_timestamp() + exp)) < 10
 
-                return MockResponse(json_data={'token': 'auth-token'})
-            return MockResponse(status_code=404)
+                return MockHTTPResponse(json_data={'token': 'auth-token'})
+            return MockHTTPResponse(status_code=404)
 
         def auth(*args, **kwargs):
             if args[0] == 'https://leader.mesos/service/some-service':
                 assert kwargs['headers']['Authorization'] == 'token=auth-token'
-                return MockResponse(json_data={})
-            return MockResponse(status_code=404)
+                return MockHTTPResponse(json_data={})
+            return MockHTTPResponse(status_code=404)
 
         with mock.patch('requests.post', side_effect=login), mock.patch('requests.Session.get', side_effect=auth):
             http = RequestsWrapper(instance, init_config)
@@ -726,7 +729,7 @@ def raise_error_once(*args, **kwargs):
                 if counter['errors'] <= 1:
                     raise Exception
 
-                return MockResponse()
+                return MockHTTPResponse()
 
             expected_headers = {'Authorization': 'Bearer secret2'}
             expected_headers.update(DEFAULT_OPTIONS['headers'])

datadog_checks_base/tests/base/utils/http/test_http_testing.py

@@ -0,0 +1,87 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Tests for HTTP testing utilities.
+
+Verifies MockHTTPResponse implementation details:
+- Default status_code=200
+- json_data auto-sets Content-Type header
+- raise_for_status() logic for 4xx/5xx codes
+- iter_lines() preserves empty lines (matches requests behavior)
+- Raw response mock structure for certificate access
+- Leading newline normalization
+"""
+
+import pytest
+
+from datadog_checks.base.utils.http_exceptions import HTTPStatusError
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
+
+
+class TestMockHTTPResponseBasics:
+    """Test basic MockHTTPResponse functionality."""
+
+    def test_default_status_code(self):
+        """Default status code is 200."""
+        response = MockHTTPResponse(content='test')
+
+        assert response.status_code == 200
+
+
+class TestMockHTTPResponseJSON:
+    """Test JSON response functionality."""
+
+    def test_json_with_custom_headers(self):
+        """json_data sets Content-Type but preserves other headers."""
+        headers = {'X-Custom': 'value'}
+        response = MockHTTPResponse(json_data={'key': 'value'}, headers=headers)
+
+        assert response.headers['Content-Type'] == 'application/json'
+        assert response.headers['X-Custom'] == 'value'
+
+
+class TestMockHTTPResponseStatus:
+    """Test raise_for_status functionality."""
+
+    def test_client_error_raises(self):
+        """4xx status codes raise HTTPStatusError."""
+        response = MockHTTPResponse(content='Not Found', status_code=404)
+
+        with pytest.raises(HTTPStatusError) as exc_info:
+            response.raise_for_status()
+
+        assert '404 Client Error' in str(exc_info.value)
+        assert exc_info.value.response is response
+
+    def test_server_error_raises(self):
+        """5xx status codes raise HTTPStatusError."""
+        response = MockHTTPResponse(content='Server Error', status_code=500)
+
+        with pytest.raises(HTTPStatusError) as exc_info:
+            response.raise_for_status()
+
+        assert '500 Server Error' in str(exc_info.value)
+        assert exc_info.value.response is response
+
+
+class TestMockHTTPResponseStreaming:
+    """Test streaming functionality (iter_content, iter_lines)."""
+
+    def test_iter_lines_preserves_empty_lines(self):
+        """Empty lines are preserved but trailing delimiter does not produce an extra element."""
+        content = 'line1\n\nline3\n'
+        response = MockHTTPResponse(content=content)
+
+        lines = list(response.iter_lines())
+        assert lines == [b'line1', b'', b'line3']
+
+
+class TestMockHTTPResponseNormalization:
+    """Test content normalization."""
+
+    def test_normalize_leading_newline(self):
+        """Leading newline is removed by default."""
+        content = '\nActual content'
+        response = MockHTTPResponse(content=content)
+
+        assert response.text == 'Actual content'