diff --git a/datadog_checks_base/changelog.d/22676.added b/datadog_checks_base/changelog.d/22676.added new file mode 100644 index 0000000000000..3ec01ccf689f8 --- /dev/null +++ b/datadog_checks_base/changelog.d/22676.added @@ -0,0 +1 @@ +Add HTTP protocol types, exception hierarchy, and MockHTTPResponse test utility. diff --git a/datadog_checks_base/datadog_checks/base/utils/http_exceptions.py b/datadog_checks_base/datadog_checks/base/utils/http_exceptions.py new file mode 100644 index 0000000000000..a807c063780c8 --- /dev/null +++ b/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 diff --git a/datadog_checks_base/datadog_checks/base/utils/http_protocol.py b/datadog_checks_base/datadog_checks/base/utils/http_protocol.py new file mode 100644 index 0000000000000..4d103de5930c3 --- /dev/null +++ b/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: ... diff --git a/datadog_checks_base/datadog_checks/base/utils/http_testing.py b/datadog_checks_base/datadog_checks/base/utils/http_testing.py new file mode 100644 index 0000000000000..4e887c987eefa --- /dev/null +++ b/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 diff --git a/datadog_checks_base/tests/base/utils/http/test_http_testing.py b/datadog_checks_base/tests/base/utils/http/test_http_testing.py new file mode 100644 index 0000000000000..98806cb9575e1 --- /dev/null +++ b/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'}