"""Basic tests for the DP-MCP Server."""
import os
import pytest
from unittest.mock import patch, MagicMock
# Set up test environment
os.environ.update({
"POSTGRES_HOST": "localhost",
"POSTGRES_USER": "test_user",
"POSTGRES_PASSWORD": "test_password",
"POSTGRES_DATABASE": "test_db",
"MINIO_ACCESS_KEY": "test_access",
"MINIO_SECRET_KEY": "test_secret"
})
import sys
sys.path.insert(0, 'src')
from dp_mcp.utils.config import get_config, validate_config
def test_get_config():
"""Test configuration loading."""
config = get_config()
assert config.postgres.host == "localhost"
assert config.postgres.user == "test_user"
assert config.minio.access_key == "test_access"
def test_validate_config():
"""Test configuration validation."""
config = get_config()
validation = validate_config(config)
assert "valid" in validation
assert "errors" in validation
@patch('dp_mcp.tools.postgres_tools.get_db_connection')
async def test_execute_query(mock_conn):
"""Test query execution."""
from dp_mcp.tools.postgres_tools import execute_query
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = [{'id': 1, 'name': 'Test'}]
mock_connection = MagicMock()
mock_connection.cursor.return_value = mock_cursor
mock_conn.return_value = mock_connection
result = await execute_query("SELECT * FROM test")
assert "Query executed successfully" in result
@patch('dp_mcp.tools.minio_tools.get_minio_client')
async def test_list_buckets(mock_client_func):
"""Test bucket listing."""
from dp_mcp.tools.minio_tools import list_buckets
mock_bucket = MagicMock()
mock_bucket.name = "test-bucket"
mock_bucket.creation_date = None
mock_client = MagicMock()
mock_client.list_buckets.return_value = [mock_bucket]
mock_client_func.return_value = mock_client
result = await list_buckets()
assert "test-bucket" in result
