"""
Unit tests for PromQL tools.
"""
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from prometheus_mcp.promql.models import (
QueryResult,
RangeQueryResult,
LabelList,
LabelValues,
MetricList,
)
from prometheus_mcp.promql import tools
class TestPromQLModels:
"""Tests for Pydantic models."""
def test_query_result_from_response(self, sample_instant_query_response):
"""Test QueryResult parsing."""
result = QueryResult.from_prometheus_response(sample_instant_query_response)
assert result.status == "success"
assert result.result_type == "vector"
assert len(result.result) == 2
assert result.result[0].metric["__name__"] == "up"
assert result.result[0].value.value == "1"
def test_query_result_error(self, sample_error_response):
"""Test QueryResult error parsing."""
result = QueryResult.from_prometheus_response(sample_error_response)
assert result.status == "error"
assert result.error_type == "bad_data"
assert "parse error" in result.error
def test_range_query_result_from_response(self, sample_range_query_response):
"""Test RangeQueryResult parsing."""
result = RangeQueryResult.from_prometheus_response(sample_range_query_response)
assert result.status == "success"
assert result.result_type == "matrix"
assert len(result.result) == 1
assert len(result.result[0].values) == 3
def test_label_list_from_response(self, sample_labels_response):
"""Test LabelList parsing."""
result = LabelList.from_prometheus_response(sample_labels_response)
assert result.status == "success"
assert "__name__" in result.labels
assert "job" in result.labels
def test_label_values_from_response(self, sample_label_values_response):
"""Test LabelValues parsing."""
result = LabelValues.from_prometheus_response(
sample_label_values_response,
label_name="job"
)
assert result.status == "success"
assert result.label_name == "job"
assert "prometheus" in result.values
def test_metric_list_from_metadata(self, sample_metadata_response):
"""Test MetricList from metadata response."""
result = MetricList.from_metadata_response(sample_metadata_response)
assert result.status == "success"
assert len(result.metrics) == 2
metric_names = [m.name for m in result.metrics]
assert "up" in metric_names
assert "http_requests_total" in metric_names
up_metric = next(m for m in result.metrics if m.name == "up")
assert up_metric.type == "gauge"
class TestPromQLTools:
"""Tests for PromQL tool functions."""
@pytest.fixture(autouse=True)
def setup_client(self, mock_boto3_session):
"""Initialize the global client for tests."""
tools.init_client(workspace_id="ws-test", region="us-east-1")
@pytest.mark.asyncio
async def test_query_instant(self, sample_instant_query_response):
"""Test query_instant tool."""
with patch.object(
tools.get_client(),
"query",
new_callable=AsyncMock
) as mock_query:
mock_query.return_value = sample_instant_query_response
result = await tools.query_instant("up")
mock_query.assert_called_once_with(promql="up", time=None)
assert result.status == "success"
assert len(result.result) == 2
@pytest.mark.asyncio
async def test_query_instant_with_time(self, sample_instant_query_response):
"""Test query_instant tool with specific time."""
with patch.object(
tools.get_client(),
"query",
new_callable=AsyncMock
) as mock_query:
mock_query.return_value = sample_instant_query_response
await tools.query_instant("up", time="2024-01-15T10:00:00Z")
mock_query.assert_called_once_with(
promql="up",
time="2024-01-15T10:00:00Z"
)
@pytest.mark.asyncio
async def test_query_range(self, sample_range_query_response):
"""Test query_range tool."""
with patch.object(
tools.get_client(),
"query_range",
new_callable=AsyncMock
) as mock_query:
mock_query.return_value = sample_range_query_response
result = await tools.query_range(
query="up",
start="2024-01-15T00:00:00Z",
end="2024-01-15T01:00:00Z",
step="1m"
)
mock_query.assert_called_once_with(
promql="up",
start="2024-01-15T00:00:00Z",
end="2024-01-15T01:00:00Z",
step="1m"
)
assert result.status == "success"
@pytest.mark.asyncio
async def test_list_labels(self, sample_labels_response):
"""Test list_labels tool."""
with patch.object(
tools.get_client(),
"labels",
new_callable=AsyncMock
) as mock_labels:
mock_labels.return_value = sample_labels_response
result = await tools.list_labels()
mock_labels.assert_called_once_with(match=None)
assert result.status == "success"
assert "__name__" in result.labels
@pytest.mark.asyncio
async def test_list_labels_with_match(self, sample_labels_response):
"""Test list_labels tool with match filter."""
with patch.object(
tools.get_client(),
"labels",
new_callable=AsyncMock
) as mock_labels:
mock_labels.return_value = sample_labels_response
await tools.list_labels(match=["up"])
mock_labels.assert_called_once_with(match=["up"])
@pytest.mark.asyncio
async def test_get_label_values(self, sample_label_values_response):
"""Test get_label_values tool."""
with patch.object(
tools.get_client(),
"label_values",
new_callable=AsyncMock
) as mock_label_values:
mock_label_values.return_value = sample_label_values_response
result = await tools.get_label_values("job")
mock_label_values.assert_called_once_with(label_name="job", match=None)
assert result.status == "success"
assert "prometheus" in result.values
@pytest.mark.asyncio
async def test_list_metrics_without_metadata(self, sample_label_values_response):
"""Test list_metrics tool without metadata."""
with patch.object(
tools.get_client(),
"label_values",
new_callable=AsyncMock
) as mock_label_values:
mock_label_values.return_value = sample_label_values_response
result = await tools.list_metrics(with_metadata=False)
mock_label_values.assert_called_once_with("__name__")
assert result.status == "success"
@pytest.mark.asyncio
async def test_list_metrics_with_metadata(self, sample_metadata_response):
"""Test list_metrics tool with metadata."""
with patch.object(
tools.get_client(),
"metadata",
new_callable=AsyncMock
) as mock_metadata:
mock_metadata.return_value = sample_metadata_response
result = await tools.list_metrics(with_metadata=True)
mock_metadata.assert_called_once()
assert result.status == "success"
assert len(result.metrics) == 2
def test_get_client_not_initialized(self):
"""Test that get_client raises when not initialized."""
# Reset the global client
tools._client = None
with pytest.raises(RuntimeError, match="not initialized"):
tools.get_client()
# Re-initialize for other tests
tools.init_client(workspace_id="ws-test", region="us-east-1")
