Batch migrate integration tests from MockResponse to MockHTTPResponse

arangodb/tests/test_arangodb.py

@@ -9,7 +9,7 @@
 from requests import HTTPError
 
 from datadog_checks.arangodb import ArangodbCheck
-from datadog_checks.dev.http import MockResponse
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.dev.utils import get_metadata_metrics
 
 from .common import METRICS
@@ -54,7 +54,7 @@ def test_check(instance, dd_run_check, aggregator, tag_condition, base_tags):
 
     def mock_requests_get(session, url, *args, **kwargs):
         fixture = url.rsplit('/', 1)[-1]
-        return MockResponse(file_path=os.path.join(os.path.dirname(__file__), 'fixtures', tag_condition, fixture))
+        return MockHTTPResponse(file_path=os.path.join(os.path.dirname(__file__), 'fixtures', tag_condition, fixture))
 
     with mock.patch('requests.Session.get', side_effect=mock_requests_get, autospec=True):
         dd_run_check(check)

cert_manager/tests/test_cert_manager.py

@@ -6,8 +6,8 @@
 import mock
 import pytest
 
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.cert_manager import CertManagerCheck
-from datadog_checks.dev.http import MockResponse
 
 from .common import ACME_METRICS, CERT_METRICS, CONTROLLER_METRICS, MOCK_INSTANCE
 
@@ -32,7 +32,7 @@ def test_check(aggregator, dd_run_check):
     check = CertManagerCheck('cert_manager', {}, [MOCK_INSTANCE])
 
     def mock_requests_get(url, *args, **kwargs):
-        return MockResponse(file_path=os.path.join(os.path.dirname(__file__), 'fixtures', 'cert_manager.txt'))
+        return MockHTTPResponse(file_path=os.path.join(os.path.dirname(__file__), 'fixtures', 'cert_manager.txt'))
 
     with mock.patch('requests.Session.get', side_effect=mock_requests_get, autospec=True):
         dd_run_check(check)

datadog_checks_base/datadog_checks/base/utils/http_testing.py

@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 import json
+import re
 from collections.abc import Mapping
 from datetime import timedelta
 from http.client import responses as http_responses
@@ -119,6 +120,31 @@ def ok(self) -> bool:
     def reason(self) -> str:
         return http_responses.get(self.status_code, '')
 
+    @property
+    def links(self) -> dict[str, dict[str, str]]:
+        """Parse Link header into a dict keyed by rel, matching requests.Response.links."""
+        header = self.headers.get('link', '').strip().strip("'\"")
+        result: dict[str, dict[str, str]] = {}
+        if not header:
+            return result
+        # Split on ", <" to avoid breaking URLs that contain commas (matches requests behavior)
+        for val in re.split(', *<', header):
+            try:
+                url, params_str = val.split(';', 1)
+            except ValueError:
+                url, params_str = val, ''
+            link: dict[str, str] = {'url': url.strip("<> '\"")}
+            for param in params_str.split(';'):
+                try:
+                    key, value = param.split('=')
+                except ValueError:
+                    break
+                link[key.strip(" '\"")] = value.strip(" '\"")
+            key = link.get('rel') or link.get('url')
+            if key:
+                result[key] = link
+        return result
+
     def json(self, **kwargs: Any) -> Any:
         return json.loads(self.text, **kwargs)
 

datadog_checks_base/tests/base/checks/prometheus/test_prometheus.py

@@ -1963,11 +1963,14 @@ def test_text_filter_input():
 
 
 def test_ssl_verify_not_raise_warning(caplog, mocked_prometheus_check, text_data):
-    from datadog_checks.dev.http import MockResponse
+    from datadog_checks.base.utils.http_testing import MockHTTPResponse
 
     check = mocked_prometheus_check
 
-    with caplog.at_level(logging.DEBUG), mock.patch('requests.Session.get', return_value=MockResponse('httpbin.org')):
+    with (
+        caplog.at_level(logging.DEBUG),
+        mock.patch('requests.Session.get', return_value=MockHTTPResponse(content='httpbin.org')),
+    ):
         resp = check.poll('https://httpbin.org/get')
 
     assert 'httpbin.org' in resp.content.decode('utf-8')
@@ -1978,12 +1981,15 @@ def test_ssl_verify_not_raise_warning(caplog, mocked_prometheus_check, text_data
 
 
 def test_ssl_verify_not_raise_warning_cert_false(caplog, mocked_prometheus_check, text_data):
-    from datadog_checks.dev.http import MockResponse
+    from datadog_checks.base.utils.http_testing import MockHTTPResponse
 
     check = mocked_prometheus_check
     check.ssl_ca_cert = False
 
-    with caplog.at_level(logging.DEBUG), mock.patch('requests.Session.get', return_value=MockResponse('httpbin.org')):
+    with (
+        caplog.at_level(logging.DEBUG),
+        mock.patch('requests.Session.get', return_value=MockHTTPResponse(content='httpbin.org')),
+    ):
         resp = check.poll('https://httpbin.org/get')
 
     assert 'httpbin.org' in resp.content.decode('utf-8')

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

@@ -83,19 +83,6 @@ def test_mock_response_normalize_leading_newline_with_indent():
     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'})
 
@@ -105,16 +92,6 @@ def test_mock_response_headers_case_insensitive():
     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'})
 
@@ -131,31 +108,42 @@ def test_mock_response_headers_update_and_setdefault():
     assert response.headers['x-iter'] == 'iter_val'
 
 
-def test_mock_response_headers_update_with_non_dict_mapping():
-    from collections.abc import Mapping
+def test_mock_response_links_standard():
+    response = MockHTTPResponse(headers={'link': '<http://example.com/page2>; rel=next; type="text/plain"'})
+    assert 'next' in response.links
+    assert response.links['next']['url'] == 'http://example.com/page2'
+    assert response.links['next']['type'] == 'text/plain'
+
+
+def test_mock_response_links_multiple():
+    response = MockHTTPResponse(
+        headers={'link': '<http://example.com/page2>; rel=next, <http://example.com/page1>; rel=prev'}
+    )
+    assert len(response.links) == 2
+    assert response.links['next']['url'] == 'http://example.com/page2'
+    assert response.links['prev']['url'] == 'http://example.com/page1'
+
 
-    class CustomHeaders(Mapping):
-        def __init__(self, d):
-            self._d = d
+def test_mock_response_links_empty():
+    assert MockHTTPResponse().links == {}
+    assert MockHTTPResponse(headers={'link': ''}).links == {}
 
-        def __getitem__(self, key):
-            return self._d[key]
 
-        def __iter__(self):
-            return iter(self._d)
+def test_mock_response_links_no_rel_keys_by_url():
+    response = MockHTTPResponse(headers={'link': '<http://example.com/page2>; type="text/plain"'})
+    assert 'http://example.com/page2' in response.links
 
-        def __len__(self):
-            return len(self._d)
 
-    response = MockHTTPResponse(headers={'A': '1'})
-    response.headers.update(CustomHeaders({'B': '2'}))
-    assert response.headers['b'] == '2'
-    assert response.headers['a'] == '1'
+def test_mock_response_links_url_with_comma():
+    response = MockHTTPResponse(headers={'link': '<http://example.com/path?a=1,2>; rel=next'})
+    assert response.links['next']['url'] == 'http://example.com/path?a=1,2'
 
 
-def test_mock_response_url():
-    assert MockHTTPResponse(url='http://example.com').url == 'http://example.com'
-    assert MockHTTPResponse().url == ''
+def test_mock_response_links_cleared_after_header_pop():
+    response = MockHTTPResponse(headers={'link': '<http://example.com>; rel=next'})
+    assert 'next' in response.links
+    response.headers.pop('link')
+    assert response.links == {}
 
 
 def test_mock_response_raw_readable():

ecs_fargate/tests/test_unit.py

@@ -8,7 +8,7 @@
 import mock
 import pytest
 
-from datadog_checks.dev.http import MockResponse
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.ecs_fargate import FargateCheck
 
 from .conftest import (
@@ -42,7 +42,8 @@ def test_failing_check(check, aggregator, dd_run_check):
     Testing fargate metadata endpoint error.
     """
     with mock.patch(
-        'datadog_checks.ecs_fargate.ecs_fargate.requests.Session.get', return_value=MockResponse('{}', status_code=500)
+        'datadog_checks.ecs_fargate.ecs_fargate.requests.Session.get',
+        return_value=MockHTTPResponse('{}', status_code=500),
     ):
         dd_run_check(check)
 
@@ -55,7 +56,8 @@ def test_invalid_response_check(check, aggregator, dd_run_check):
     Testing invalid fargate metadata payload.
     """
     with mock.patch(
-        'datadog_checks.ecs_fargate.ecs_fargate.requests.Session.get', return_value=MockResponse('{}', status_code=200)
+        'datadog_checks.ecs_fargate.ecs_fargate.requests.Session.get',
+        return_value=MockHTTPResponse('{}', status_code=200),
     ):
         dd_run_check(check)
 

elastic/tests/test_unit.py

@@ -8,7 +8,7 @@
 import pytest
 
 from datadog_checks.base import ConfigurationError, is_affirmative
-from datadog_checks.dev.http import MockResponse
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.elastic import ESCheck
 from datadog_checks.elastic.elastic import AuthenticationError, get_value_from_path
 from datadog_checks.elastic.metrics import INDEX_STATS_METRICS, stats_for_version
@@ -135,7 +135,7 @@ def test_get_template_metrics(aggregator, instance, mock_http_response):
 def test_get_template_metrics_raise_exception(aggregator, instance):
     with mock.patch(
         'requests.Session.get',
-        return_value=MockResponse(status_code=403),
+        return_value=MockHTTPResponse(status_code=403),
     ):
         check = ESCheck('elastic', {}, instances=[instance])
         # Make sure we do not throw an exception and move on
@@ -152,7 +152,7 @@ def test_get_value_from_path():
 def test__get_data_throws_authentication_error(instance):
     with mock.patch(
         'requests.Session.get',
-        return_value=MockResponse(status_code=400),
+        return_value=MockHTTPResponse(status_code=400),
     ):
         check = ESCheck('elastic', {}, instances=[instance])
 
@@ -163,7 +163,7 @@ def test__get_data_throws_authentication_error(instance):
 def test__get_data_creates_critical_service_alert(aggregator, instance):
     with mock.patch(
         'requests.Session.get',
-        return_value=MockResponse(status_code=500),
+        return_value=MockHTTPResponse(status_code=500),
     ):
         check = ESCheck('elastic', {}, instances=[instance])
 
@@ -174,7 +174,7 @@ def test__get_data_creates_critical_service_alert(aggregator, instance):
             check.SERVICE_CHECK_CONNECT_NAME,
             status=check.CRITICAL,
             tags=check._config.service_check_tags,
-            message="Error 500 Server Error: None for url: None when hitting test.com",
+            message="Error 500 Server Error when hitting test.com",
         )
 
 
@@ -194,7 +194,7 @@ def test__get_data_creates_critical_service_alert(aggregator, instance):
 def test_disable_legacy_sc_tags(aggregator, es_instance):
     with mock.patch(
         'requests.Session.get',
-        return_value=MockResponse(status_code=500),
+        return_value=MockHTTPResponse(status_code=500),
     ):
         check = ESCheck('elastic', {}, instances=[es_instance])
 
@@ -210,7 +210,7 @@ def test_disable_legacy_sc_tags(aggregator, es_instance):
             check.SERVICE_CHECK_CONNECT_NAME,
             status=check.CRITICAL,
             tags=expected_tags,
-            message="Error 500 Server Error: None for url: None when hitting test.com",
+            message="Error 500 Server Error when hitting test.com",
         )