Migrate kube_apiserver_metrics, kube_dns, kube_metrics_server, kube_proxy tests to mock_openmetrics_http

kube_apiserver_metrics/tests/common.py

@@ -3,6 +3,15 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
 import os
+from unittest.mock import MagicMock
+
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
 
 APISERVER_INSTANCE_BEARER_TOKEN = "XXX"
 HERE = os.path.dirname(os.path.abspath(__file__))
+
+
+def make_mock_metrics(mock_openmetrics_http: MagicMock, fixture_filename: str) -> MagicMock:
+    f_name = os.path.join(HERE, 'fixtures', fixture_filename)
+    mock_openmetrics_http.get.return_value = MockHTTPResponse(file_path=f_name, headers={'Content-Type': 'text/plain'})
+    return mock_openmetrics_http

kube_apiserver_metrics/tests/test_kube_apiserver_metrics_1_25.py

@@ -2,14 +2,12 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-# stdlib
-import os
-
-import mock
 import pytest
 
 from datadog_checks.kube_apiserver_metrics import KubeAPIServerMetricsCheck
 
+from .common import make_mock_metrics
+
 customtag = "custom:tag"
 
 instance = {
@@ -20,19 +18,8 @@
 
 
 @pytest.fixture()
-def mock_get():
-    f_name = os.path.join(os.path.dirname(__file__), 'fixtures', 'metrics_1.25.3.txt')
-    with open(f_name, 'r') as f:
-        text_data = f.read()
-    with mock.patch(
-        'requests.Session.get',
-        return_value=mock.MagicMock(
-            status_code=200,
-            iter_lines=lambda **kwargs: text_data.split("\n"),
-            headers={'Content-Type': "text/plain", 'Authorization': "Bearer XXX"},
-        ),
-    ):
-        yield
+def mock_get(mock_openmetrics_http):
+    return make_mock_metrics(mock_openmetrics_http, 'metrics_1.25.3.txt')
 
 
 class TestKubeAPIServerMetrics:

kube_apiserver_metrics/tests/test_sli_metrics.py

@@ -2,34 +2,21 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-import os
 from collections import namedtuple
 
-import mock
 import pytest
 
 from datadog_checks.kube_apiserver_metrics import KubeAPIServerMetricsCheck
 
-from .common import HERE
+from .common import make_mock_metrics
 
 # Constants
 CHECK_NAME = "kube_apiserver"
 
 
 @pytest.fixture()
-def mock_metrics():
-    f_name = os.path.join(HERE, "fixtures", "metrics_slis_1.27.3.txt")
-    with open(f_name, "r") as f:
-        text_data = f.read()
-    with mock.patch(
-        "requests.Session.get",
-        return_value=mock.MagicMock(
-            status_code=200,
-            iter_lines=lambda **kwargs: text_data.split("\n"),
-            headers={"Content-Type": "text/plain"},
-        ),
-    ):
-        yield
+def mock_metrics(mock_openmetrics_http):
+    return make_mock_metrics(mock_openmetrics_http, 'metrics_slis_1.27.3.txt')
 
 
 def test_check_metrics_slis(aggregator, mock_metrics, instance):

kube_dns/tests/common.py

@@ -0,0 +1,16 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import os
+from unittest.mock import MagicMock
+
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
+from datadog_checks.dev import get_here
+
+HERE = get_here()
+
+
+def make_mock_metrics(mock_openmetrics_http: MagicMock, fixture_filename: str) -> MagicMock:
+    f_name = os.path.join(HERE, 'fixtures', fixture_filename)
+    mock_openmetrics_http.get.return_value = MockHTTPResponse(file_path=f_name, headers={'Content-Type': 'text/plain'})
+    return mock_openmetrics_http

kube_dns/tests/conftest.py

@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from copy import deepcopy
+from unittest import mock
 
 import pytest
 
@@ -20,3 +21,10 @@ def dd_environment():
 @pytest.fixture
 def instance():
     return deepcopy(INSTANCE)
+
+
+@pytest.fixture
+def mock_healthcheck_wrapper():
+    # _healthcheck_http_handler builds its own RequestsWrapper outside self.http
+    with mock.patch('datadog_checks.kube_dns.kube_dns.RequestsWrapper'):
+        yield

kube_dns/tests/test_kube_dns.py

@@ -2,34 +2,23 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-# stdlib
-import os
-
 import mock
 import pytest
 import requests
 
-# project
 from datadog_checks.base import AgentCheck
 from datadog_checks.kube_dns import KubeDNSCheck
 
+from .common import make_mock_metrics
+
 customtag = "custom:tag"
 
 instance = {'prometheus_endpoint': 'http://localhost:10055/metrics', 'tags': [customtag]}
 
 
 @pytest.fixture()
-def mock_get():
-    f_name = os.path.join(os.path.dirname(__file__), 'fixtures', 'metrics.txt')
-    with open(f_name, 'r') as f:
-        text_data = f.read()
-    with mock.patch(
-        'requests.Session.get',
-        return_value=mock.MagicMock(
-            status_code=200, iter_lines=lambda **kwargs: text_data.split("\n"), headers={'Content-Type': "text/plain"}
-        ),
-    ):
-        yield
+def mock_get(mock_openmetrics_http):
+    return make_mock_metrics(mock_openmetrics_http, 'metrics.txt')
 
 
 @pytest.fixture
@@ -60,7 +49,7 @@ class TestKubeDNS:
         NAMESPACE + '.cachemiss_count.count',
     ]
 
-    def test_check(self, aggregator, mock_get):
+    def test_check(self, aggregator, mock_get, mock_healthcheck_wrapper):
         """
         Testing kube_dns check.
         """
@@ -83,26 +72,26 @@ def test_check(self, aggregator, mock_get):
         aggregator.assert_metric(name)
         aggregator.assert_metric(name, tags=['custom:tag', 'system:reverse'])
 
-    def test_service_check_ok(self, monkeypatch):
+    @pytest.mark.parametrize(
+        'side_effect, expected_status, extra_kwargs',
+        [
+            (None, AgentCheck.OK, {}),
+            (requests.HTTPError('health check failed'), AgentCheck.CRITICAL, {'message': 'health check failed'}),
+        ],
+        ids=['ok', 'http_error'],
+    )
+    def test_service_check(self, monkeypatch, side_effect, expected_status, extra_kwargs):
         instance_tags = [customtag]
-
         check = KubeDNSCheck(self.CHECK_NAME, {}, [instance])
-
         monkeypatch.setattr(check, 'service_check', mock.Mock())
 
-        calls = [
-            mock.call(self.NAMESPACE + '.up', AgentCheck.OK, tags=instance_tags),
-            mock.call(self.NAMESPACE + '.up', AgentCheck.CRITICAL, tags=instance_tags, message='health check failed'),
-        ]
-
-        # successful health check
-        with mock.patch('requests.Session.get', return_value=mock.MagicMock(status_code=200)):
-            check._perform_service_check(instance)
+        healthcheck_url = check.instance['health_url']
+        handler = mock.MagicMock()
+        handler.get.return_value.raise_for_status = mock.Mock(side_effect=side_effect)
+        check._http_handlers[healthcheck_url] = handler
 
-        # failed health check
-        raise_error = mock.Mock()
-        raise_error.side_effect = requests.HTTPError('health check failed')
-        with mock.patch('requests.Session.get', return_value=mock.MagicMock(raise_for_status=raise_error)):
-            check._perform_service_check(instance)
+        check._perform_service_check(instance)
 
-        check.service_check.assert_has_calls(calls)
+        check.service_check.assert_called_with(
+            self.NAMESPACE + '.up', expected_status, tags=instance_tags, **extra_kwargs
+        )

kube_metrics_server/tests/common.py

@@ -0,0 +1,16 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import os
+from unittest.mock import MagicMock
+
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
+from datadog_checks.dev import get_here
+
+HERE = get_here()
+
+
+def make_mock_metrics(mock_openmetrics_http: MagicMock, fixture_filename: str) -> MagicMock:
+    f_name = os.path.join(HERE, 'fixtures', fixture_filename)
+    mock_openmetrics_http.get.return_value = MockHTTPResponse(file_path=f_name, headers={'Content-Type': 'text/plain'})
+    return mock_openmetrics_http

kube_metrics_server/tests/conftest.py

@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from copy import deepcopy
+from unittest import mock
 
 import pytest
 
@@ -21,3 +22,10 @@ def dd_environment():
 @pytest.fixture
 def instance():
     return deepcopy(INSTANCE)
+
+
+@pytest.fixture
+def mock_healthcheck_wrapper():
+    # _healthcheck_http_handler builds its own RequestsWrapper outside self.http
+    with mock.patch('datadog_checks.kube_metrics_server.kube_metrics_server.RequestsWrapper'):
+        yield

kube_metrics_server/tests/test_kube_metrics_server.py

@@ -2,15 +2,15 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-import os
-
 import mock
 import pytest
 import requests
 
 from datadog_checks.base import AgentCheck
 from datadog_checks.kube_metrics_server import KubeMetricsServerCheck
 
+from .common import make_mock_metrics
+
 instance = {
     'prometheus_url': 'https://localhost:443/metrics',
     'send_histograms_buckets': True,
@@ -23,20 +23,11 @@
 
 
 @pytest.fixture()
-def mock_metrics():
-    f_name = os.path.join(os.path.dirname(__file__), 'fixtures', 'metrics.txt')
-    with open(f_name, 'r') as f:
-        text_data = f.read()
-    with mock.patch(
-        'requests.Session.get',
-        return_value=mock.MagicMock(
-            status_code=200, iter_lines=lambda **kwargs: text_data.split("\n"), headers={'Content-Type': "text/plain"}
-        ),
-    ):
-        yield
-
-
-def test_check_metrics(aggregator, mock_metrics):
+def mock_metrics(mock_openmetrics_http):
+    return make_mock_metrics(mock_openmetrics_http, 'metrics.txt')
+
+
+def test_check_metrics(aggregator, mock_metrics, mock_healthcheck_wrapper):
     c = KubeMetricsServerCheck(CHECK_NAME, {}, [instance])
     c.check(instance)
 
@@ -63,26 +54,24 @@ def assert_metric(name, **kwargs):
     aggregator.assert_all_metrics_covered()
 
 
-def test_service_check_ok(monkeypatch):
+@pytest.mark.parametrize(
+    'side_effect, expected_status, extra_kwargs',
+    [
+        (None, AgentCheck.OK, {}),
+        (requests.HTTPError('health check failed'), AgentCheck.CRITICAL, {'message': 'health check failed'}),
+    ],
+    ids=['ok', 'http_error'],
+)
+def test_service_check(monkeypatch, side_effect, expected_status, extra_kwargs):
     instance_tags = []
-
     check = KubeMetricsServerCheck(CHECK_NAME, {}, [instance])
-
     monkeypatch.setattr(check, 'service_check', mock.Mock())
 
-    calls = [
-        mock.call(NAMESPACE + '.up', AgentCheck.OK, tags=instance_tags),
-        mock.call(NAMESPACE + '.up', AgentCheck.CRITICAL, tags=instance_tags, message='health check failed'),
-    ]
-
-    # successful health check
-    with mock.patch('requests.Session.get', return_value=mock.MagicMock(status_code=200)):
-        check._perform_service_check(instance)
+    healthcheck_url = check.instance['health_url']
+    handler = mock.MagicMock()
+    handler.get.return_value.raise_for_status = mock.Mock(side_effect=side_effect)
+    check._http_handlers[healthcheck_url] = handler
 
-    # failed health check
-    raise_error = mock.Mock()
-    raise_error.side_effect = requests.HTTPError('health check failed')
-    with mock.patch('requests.Session.get', return_value=mock.MagicMock(raise_for_status=raise_error)):
-        check._perform_service_check(instance)
+    check._perform_service_check(instance)
 
-    check.service_check.assert_has_calls(calls)
+    check.service_check.assert_called_with(NAMESPACE + '.up', expected_status, tags=instance_tags, **extra_kwargs)

kube_proxy/tests/__init__.py

@@ -0,0 +1,3 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)

kube_proxy/tests/common.py

@@ -0,0 +1,16 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import os
+from unittest.mock import MagicMock
+
+from datadog_checks.base.utils.http_testing import MockHTTPResponse
+from datadog_checks.dev import get_here
+
+HERE = get_here()
+
+
+def make_mock_metrics(mock_openmetrics_http: MagicMock, fixture_filename: str) -> MagicMock:
+    f_name = os.path.join(HERE, 'fixtures', fixture_filename)
+    mock_openmetrics_http.get.return_value = MockHTTPResponse(file_path=f_name, headers={'Content-Type': 'text/plain'})
+    return mock_openmetrics_http

kube_proxy/tests/conftest.py

@@ -2,6 +2,7 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from copy import deepcopy
+from unittest import mock
 
 import pytest
 
@@ -19,3 +20,10 @@ def dd_environment():
 @pytest.fixture
 def instance():
     return deepcopy(INSTANCE)
+
+
+@pytest.fixture
+def mock_healthcheck_wrapper():
+    # _healthcheck_http_handler builds its own RequestsWrapper outside self.http
+    with mock.patch('datadog_checks.kube_proxy.kube_proxy.RequestsWrapper'):
+        yield