Migrate conftest.py MockResponse fixture chains 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)

avi_vantage/tests/conftest.py

@@ -4,14 +4,14 @@
 import json
 import os
 from typing import Any, AnyStr
+from unittest.mock import MagicMock
 from urllib.parse import urlparse
 
-import mock
 import pytest
 
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.dev import docker_run, get_docker_hostname, get_here
 from datadog_checks.dev.conditions import CheckDockerLogs
-from datadog_checks.dev.http import MockResponse
 
 HERE = get_here()
 
@@ -57,7 +57,7 @@ def _get_metrics(metrics_folder=NO_TENANT_METRICS_FOLDER, endpoint=None):
 
 
 @pytest.fixture
-def mock_client():
+def mock_client(mock_http):
     def mock_get(url: AnyStr, *__: Any, **___: Any):
         parsed = urlparse(url)
         resource = [part for part in parsed.path.split('/') if len(part) > 0][-1]
@@ -69,20 +69,21 @@ def mock_get(url: AnyStr, *__: Any, **___: Any):
         path['tenant=admin%2Ctenant_a%2Ctenant_b'] = MULTIPLE_TENANTS_METRICS_FOLDER
 
         if query_params:
-            return MockResponse(
+            return MockHTTPResponse(
                 file_path=os.path.join(HERE, 'compose', 'fixtures', path[query_params], f'{resource}_metrics')
             )
 
-        return MockResponse(
+        return MockHTTPResponse(
             file_path=os.path.join(HERE, 'compose', 'fixtures', NO_TENANT_METRICS_FOLDER, f'{resource}_metrics')
         )
 
     def mock_post(url: AnyStr, *__: Any, **___: Any):
-        return mock.MagicMock(status_code=200, content=b'{"results": []}')
+        return MockHTTPResponse(json_data={"results": []})
 
-    with mock.patch('datadog_checks.base.utils.http.RequestsWrapper.get', side_effect=mock_get):
-        with mock.patch('datadog_checks.base.utils.http.RequestsWrapper.post', new=mock_post):
-            yield
+    mock_http.session = MagicMock(cookies={})
+    mock_http.get.side_effect = mock_get
+    mock_http.post.side_effect = mock_post
+    yield
 
 
 @pytest.fixture(scope='session')

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)

citrix_hypervisor/tests/conftest.py

@@ -3,11 +3,10 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 import os
 
-import mock
 import pytest
 
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.dev import docker_run
-from datadog_checks.dev.http import MockResponse
 
 from . import common
 
@@ -38,17 +37,17 @@ def mock_requests_get(url, *args, **kwargs):
     print(url_parts)
 
     if url_parts[0] == 'wrong':
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
     json_file = f"rrd_updates_{url_parts[0]}.json" if url_parts[1] == "rrd_updates" else f"{url_parts[1]}.json"
     path = os.path.join(common.HERE, 'fixtures', 'standalone', json_file)
     if not os.path.exists(path):
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
-    return MockResponse(file_path=path)
+    return MockHTTPResponse(file_path=path)
 
 
 @pytest.fixture
-def mock_responses():
-    with mock.patch('requests.Session.get', side_effect=mock_requests_get):
-        yield
+def mock_responses(mock_http):
+    mock_http.get.side_effect = mock_requests_get
+    yield

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/conftest.py

@@ -6,8 +6,8 @@
 
 import pytest
 
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 from datadog_checks.dev import get_here
-from datadog_checks.dev.http import MockResponse
 from datadog_checks.ecs_fargate import FargateCheck
 
 HERE = get_here()
@@ -83,38 +83,38 @@
 def mocked_requests_get_linux(*args, **kwargs):
     # v2
     if args[0].endswith("/metadata"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
     elif args[0].endswith("/stats"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', LINUX_STATS_FIXTURE))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', LINUX_STATS_FIXTURE))
     else:
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
 
 def mocked_requests_get_linux_v4(*args, **kwargs):
     if args[0].endswith("/task"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata_v4.json'))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata_v4.json'))
     elif args[0].endswith("/task/stats"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', LINUX_STATS_FIXTURE_V4))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', LINUX_STATS_FIXTURE_V4))
     else:
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
 
 def mocked_requests_get_windows(*args, **kwargs):
     if args[0].endswith("/metadata"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
     elif args[0].endswith("/stats"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', WINDOWS_STATS_FIXTURE))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', WINDOWS_STATS_FIXTURE))
     else:
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
 
 def mocked_requests_get_sys_delta(*args, **kwargs):
     if args[0].endswith("/metadata"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', 'metadata.json'))
     elif args[0].endswith("/stats"):
-        return MockResponse(file_path=os.path.join(HERE, 'fixtures', 'stats_wrong_system_delta.json'))
+        return MockHTTPResponse(file_path=os.path.join(HERE, 'fixtures', 'stats_wrong_system_delta.json'))
     else:
-        return MockResponse(status_code=404)
+        return MockHTTPResponse(status_code=404)
 
 
 def mocked_get_tags(entity, _):

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)