63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""Tests for dashboard.prometheus_utils.query."""
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from django.test import TestCase
|
|
|
|
from dashboard.prometheus_utils.query import query_prometheus
|
|
|
|
|
|
class QueryPrometheusTest(TestCase):
|
|
"""Tests for query_prometheus."""
|
|
|
|
@patch("dashboard.prometheus_utils.query.requests.get")
|
|
def test_single_result_returns_value_string(self, mock_get):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"data": {
|
|
"result": [
|
|
{"value": ["1234567890", "42"]}
|
|
]
|
|
}
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_response
|
|
|
|
result = query_prometheus("some_query")
|
|
|
|
self.assertEqual(result, "42")
|
|
mock_response.raise_for_status.assert_called_once()
|
|
mock_get.assert_called_once()
|
|
call_kw = mock_get.call_args
|
|
self.assertIn("params", call_kw.kwargs)
|
|
self.assertEqual(call_kw.kwargs["params"]["query"], "some_query")
|
|
|
|
@patch("dashboard.prometheus_utils.query.requests.get")
|
|
def test_multiple_results_returns_full_result_list(self, mock_get):
|
|
mock_response = MagicMock()
|
|
result_list = [
|
|
{"metric": {"host": "h1"}, "value": ["1", "10"]},
|
|
{"metric": {"host": "h2"}, "value": ["1", "20"]},
|
|
]
|
|
mock_response.json.return_value = {"data": {"result": result_list}}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_response
|
|
|
|
result = query_prometheus("vector_query")
|
|
|
|
self.assertEqual(result, result_list)
|
|
self.assertEqual(len(result), 2)
|
|
|
|
@patch("dashboard.prometheus_utils.query.requests.get")
|
|
def test_uses_prometheus_url_from_settings(self, mock_get):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": {"result": [{"value": ["0", "1"]}]}}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_response
|
|
|
|
query_prometheus("test")
|
|
|
|
mock_get.assert_called_once()
|
|
args, kwargs = mock_get.call_args
|
|
url = args[0] if args else kwargs.get("url", "")
|
|
self.assertIn("/api/v1/query", url)
|