"""
Pytest fixtures for Prometheus MCP tests.
"""
import os
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
# Set environment variables before importing modules
os.environ["PROMETHEUS_WORKSPACE_ID"] = "ws-test-workspace-id"
os.environ["AWS_REGION"] = "us-east-1"
@pytest.fixture
def mock_aws_credentials():
"""Mock AWS credentials for testing."""
with patch.dict(os.environ, {
"AWS_ACCESS_KEY_ID": "testing",
"AWS_SECRET_ACCESS_KEY": "testing",
"AWS_SECURITY_TOKEN": "testing",
"AWS_SESSION_TOKEN": "testing",
"AWS_DEFAULT_REGION": "us-east-1",
}):
yield
@pytest.fixture
def mock_boto3_session():
"""Mock boto3 session for credential retrieval."""
mock_credentials = MagicMock()
mock_credentials.access_key = "test-access-key"
mock_credentials.secret_key = "test-secret-key"
mock_credentials.token = "test-token"
mock_session = MagicMock()
mock_session.get_credentials.return_value = mock_credentials
with patch("boto3.Session", return_value=mock_session):
yield mock_session
@pytest.fixture
def sample_instant_query_response():
"""Sample response from an instant query."""
return {
"status": "success",
"data": {
"resultType": "vector",
"result": [
{
"metric": {
"__name__": "up",
"instance": "localhost:9090",
"job": "prometheus"
},
"value": [1705315200, "1"]
},
{
"metric": {
"__name__": "up",
"instance": "localhost:9100",
"job": "node"
},
"value": [1705315200, "1"]
}
]
}
}
@pytest.fixture
def sample_range_query_response():
"""Sample response from a range query."""
return {
"status": "success",
"data": {
"resultType": "matrix",
"result": [
{
"metric": {
"__name__": "up",
"instance": "localhost:9090",
"job": "prometheus"
},
"values": [
[1705315200, "1"],
[1705315260, "1"],
[1705315320, "1"]
]
}
]
}
}
@pytest.fixture
def sample_labels_response():
"""Sample response from labels endpoint."""
return {
"status": "success",
"data": ["__name__", "instance", "job", "namespace", "pod"]
}
@pytest.fixture
def sample_label_values_response():
"""Sample response from label values endpoint."""
return {
"status": "success",
"data": ["prometheus", "node", "alertmanager", "grafana"]
}
@pytest.fixture
def sample_metadata_response():
"""Sample response from metadata endpoint."""
return {
"status": "success",
"data": {
"up": [
{
"type": "gauge",
"help": "The current health status of the target",
"unit": ""
}
],
"http_requests_total": [
{
"type": "counter",
"help": "Total number of HTTP requests",
"unit": ""
}
]
}
}
@pytest.fixture
def sample_error_response():
"""Sample error response."""
return {
"status": "error",
"errorType": "bad_data",
"error": "invalid expression: parse error at char 5"
}