Add HTTP protocol types, exception hierarchy, and MockHTTPResponse

datadog_checks_base/changelog.d/22676.added

@@ -0,0 +1 @@
+Add HTTP protocol types, exception hierarchy, and MockHTTPResponse test utility.

datadog_checks_base/datadog_checks/base/utils/http_exceptions.py

@@ -0,0 +1,40 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from typing import Any
+
+__all__ = [
+    'HTTPError',
+    'HTTPRequestError',
+    'HTTPStatusError',
+    'HTTPTimeoutError',
+    'HTTPConnectionError',
+    'HTTPSSLError',
+]
+
+
+class HTTPError(Exception):
+    def __init__(self, message: str, response: Any = None, request: Any = None):
+        super().__init__(message)
+        self.response = response
+        self.request = request
+
+
+class HTTPRequestError(HTTPError):
+    pass
+
+
+class HTTPStatusError(HTTPError):
+    pass
+
+
+class HTTPTimeoutError(HTTPRequestError):
+    pass
+
+
+class HTTPConnectionError(HTTPRequestError):
+    pass
+
+
+class HTTPSSLError(HTTPConnectionError):
+    pass

datadog_checks_base/datadog_checks/base/utils/http_protocol.py

@@ -0,0 +1,46 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any, Iterator, Protocol
+
+
+class HTTPResponseProtocol(Protocol):
+    status_code: int
+    content: bytes
+    text: str
+    headers: Mapping[str, str]
+
+    @property
+    def ok(self) -> bool: ...
+    @property
+    def reason(self) -> str: ...
+
+    def json(self, **kwargs: Any) -> Any: ...
+    def raise_for_status(self) -> None: ...
+    def close(self) -> None: ...
+    def iter_content(self, chunk_size: int | None = None, decode_unicode: bool = False) -> Iterator[bytes | str]: ...
+    def iter_lines(
+        self,
+        chunk_size: int | None = None,
+        decode_unicode: bool = False,
+        delimiter: bytes | str | None = None,
+    ) -> Iterator[bytes | str]: ...
+    def __enter__(self) -> HTTPResponseProtocol: ...
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool | None: ...
+
+
+class HTTPClientProtocol(Protocol):
+    options: dict[str, Any]
+
+    def get(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def post(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def head(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def put(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def patch(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def delete(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def options_method(self, url: str, **options: Any) -> HTTPResponseProtocol: ...
+    def get_header(self, name: str, default: str | None = None) -> str | None: ...
+    def set_header(self, name: str, value: str) -> None: ...

datadog_checks_base/datadog_checks/base/utils/http_testing.py

@@ -0,0 +1,169 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import json
+from datetime import timedelta
+from http.client import responses as http_responses
+from io import BytesIO
+from textwrap import dedent
+from typing import Any, Iterator
+from unittest.mock import MagicMock
+
+from datadog_checks.base.utils.http_exceptions import HTTPStatusError
+
+__all__ = ['MockHTTPResponse']
+
+
+class _CaseInsensitiveDict(dict):
+    """Case-insensitive dict for HTTP headers. Keys are stored lowercased."""
+
+    def __init__(self, data=None):
+        super().__init__()
+        if data:
+            for k, v in data.items():
+                self[k] = v
+
+    def __setitem__(self, key, value):
+        super().__setitem__(key.lower() if isinstance(key, str) else key, value)
+
+    def __getitem__(self, key):
+        return super().__getitem__(key.lower() if isinstance(key, str) else key)
+
+    def __contains__(self, key):
+        return super().__contains__(key.lower() if isinstance(key, str) else key)
+
+    def __delitem__(self, key):
+        super().__delitem__(key.lower() if isinstance(key, str) else key)
+
+    def get(self, key, default=None):
+        return super().get(key.lower() if isinstance(key, str) else key, default)
+
+    def pop(self, key, *args):
+        return super().pop(key.lower() if isinstance(key, str) else key, *args)
+
+    def update(self, other=(), **kwargs):
+        if isinstance(other, dict):
+            other = {(k.lower() if isinstance(k, str) else k): v for k, v in other.items()}
+        elif other:
+            other = [(k.lower() if isinstance(k, str) else k, v) for k, v in other]
+        kwargs = {k.lower(): v for k, v in kwargs.items()}
+        super().update(other, **kwargs)
+
+    def setdefault(self, key, default=None):
+        return super().setdefault(key.lower() if isinstance(key, str) else key, default)
+
+
+class MockHTTPResponse:
+    """Library-agnostic mock HTTP response implementing HTTPResponseProtocol."""
+
+    # Parameter order differs from MockResponse; not a compatibility concern since all callers use keyword args.
+    def __init__(
+        self,
+        content: str | bytes = '',
+        status_code: int = 200,
+        headers: dict[str, str] | None = None,
+        json_data: Any = None,
+        file_path: str | None = None,
+        cookies: dict[str, str] | None = None,
+        elapsed_seconds: float = 0.1,
+        normalize_content: bool = True,
+        url: str = '',
+    ):
+        self.url = url
+
+        if json_data is not None:
+            content = json.dumps(json_data)
+            # Copy to avoid mutating the caller's dict
+            headers = dict(headers) if headers is not None else {}
+            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 = dedent(content[1:]) if isinstance(content, str) else content[1:]
+
+        self._content = content.encode('utf-8') if isinstance(content, str) else content
+        self.status_code = status_code
+        self.headers = _CaseInsensitiveDict(headers or {})
+        self.cookies = cookies or {}
+        self.encoding: str | None = None
+        self.elapsed = timedelta(seconds=elapsed_seconds)
+        self._stream = BytesIO(self._content)
+
+        self.raw = MagicMock()
+        self.raw.read = self._stream.read
+        self.raw.connection.sock.getpeercert.side_effect = lambda binary_form=False: b'mock-cert' if binary_form else {}
+
+    @property
+    def content(self) -> bytes:
+        return self._content
+
+    @property
+    def text(self) -> str:
+        return self._content.decode('utf-8')
+
+    @property
+    def ok(self) -> bool:
+        # Transitional: mirrors requests.Response.ok for current production code.
+        # httpx uses is_success/is_client_error/is_server_error instead.
+        return self.status_code < 400
+
+    @property
+    def reason(self) -> str:
+        return http_responses.get(self.status_code, '')
+
+    def json(self, **kwargs: Any) -> Any:
+        return json.loads(self.text, **kwargs)
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            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
+        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
+
+    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'
+
+        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
+
+    def close(self) -> None:
+        # No-op: requests.Response.close() releases the network connection, but
+        # content is already buffered in memory. Matching that behaviour here
+        # so the same instance can be returned by a mock multiple times.
+        pass
+
+    def __enter__(self) -> 'MockHTTPResponse':
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> bool | None:
+        return None

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

@@ -0,0 +1,135 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import json
+
+import pytest
+
+from datadog_checks.base.utils.http_exceptions import HTTPStatusError
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
+
+
+def test_mock_response_json_with_custom_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'
+
+
+def test_mock_response_json_does_not_mutate_caller_headers():
+    headers = {'X-Custom': 'value'}
+    MockHTTPResponse(json_data={'key': 'value'}, headers=headers)
+
+    assert list(headers.keys()) == ['X-Custom']
+
+
+def test_mock_response_file_path(tmp_path):
+    f = tmp_path / 'fixture.txt'
+    f.write_bytes(b'file content')
+
+    response = MockHTTPResponse(file_path=str(f))
+    assert response.content == b'file content'
+
+
+def test_mock_response_raise_for_status():
+    response_404 = MockHTTPResponse(content='Not Found', status_code=404)
+    with pytest.raises(HTTPStatusError) as exc_info:
+        response_404.raise_for_status()
+    assert '404 Client Error' in str(exc_info.value)
+    assert exc_info.value.response is response_404
+
+    response_500 = MockHTTPResponse(content='Server Error', status_code=500)
+    with pytest.raises(HTTPStatusError) as exc_info:
+        response_500.raise_for_status()
+    assert '500 Server Error' in str(exc_info.value)
+    assert exc_info.value.response is response_500
+
+
+def test_mock_response_iter_content_chunks():
+    response = MockHTTPResponse(content='hello world')
+
+    chunks = list(response.iter_content(chunk_size=5))
+    assert chunks == [b'hello', b' worl', b'd']
+
+
+def test_mock_response_iter_lines_preserves_empty_lines():
+    content = 'line1\n\nline3\n'
+    response = MockHTTPResponse(content=content)
+
+    lines = list(response.iter_lines())
+    assert lines == [b'line1', b'', b'line3']
+
+
+def test_mock_response_normalize_leading_newline():
+    content = '\nActual content'
+    response = MockHTTPResponse(content=content)
+
+    assert response.text == 'Actual content'
+
+
+def test_mock_response_normalize_leading_newline_with_indent():
+    content = """
+        line one
+        line two
+    """
+    response = MockHTTPResponse(content=content)
+    assert response.text == "line one\nline two\n"
+
+
+def test_mock_response_ok_property():
+    assert MockHTTPResponse(status_code=200).ok is True
+    assert MockHTTPResponse(status_code=399).ok is True
+    assert MockHTTPResponse(status_code=400).ok is False
+    assert MockHTTPResponse(status_code=500).ok is False
+
+
+def test_mock_response_reason_property():
+    assert MockHTTPResponse(status_code=200).reason == 'OK'
+    assert MockHTTPResponse(status_code=404).reason == 'Not Found'
+    assert MockHTTPResponse(status_code=999).reason == ''
+
+
+def test_mock_response_headers_case_insensitive():
+    response = MockHTTPResponse(headers={'Content-Type': 'text/plain', 'X-Custom': 'val'})
+
+    assert response.headers['Content-Type'] == 'text/plain'
+    assert response.headers['content-type'] == 'text/plain'
+    assert response.headers.get('Content-Type') == 'text/plain'
+    assert response.headers.get('cOnTeNt-tYpE') == 'text/plain'
+
+
+def test_mock_response_headers_delete_and_pop():
+    response = MockHTTPResponse(headers={'Content-Type': 'text/plain', 'X-Custom': 'val'})
+
+    del response.headers['Content-Type']
+    assert 'content-type' not in response.headers
+
+    assert response.headers.pop('X-Custom') == 'val'
+    assert response.headers.pop('X-Custom', 'gone') == 'gone'
+
+
+def test_mock_response_headers_update_and_setdefault():
+    response = MockHTTPResponse(headers={'Content-Type': 'text/plain'})
+
+    response.headers.update({'X-New': 'new_val'})
+    assert response.headers['x-new'] == 'new_val'
+
+    response.headers.setdefault('X-Default', 'default_val')
+    assert response.headers['x-default'] == 'default_val'
+
+    response.headers.setdefault('Content-Type', 'should-not-change')
+    assert response.headers['content-type'] == 'text/plain'
+
+    response.headers.update([('X-Iter', 'iter_val')])
+    assert response.headers['x-iter'] == 'iter_val'
+
+
+def test_mock_response_url():
+    assert MockHTTPResponse(url='http://example.com').url == 'http://example.com'
+    assert MockHTTPResponse().url == ''
+
+
+def test_mock_response_raw_readable():
+    response = MockHTTPResponse(json_data={'key': 'value'})
+    assert json.load(response.raw) == {'key': 'value'}