mcp-dbutils
by donghao1393
- tests
- integration
"""Test MySQL configuration helper methods"""
import tempfile
import pytest
import yaml
from mcp_dbutils.mysql.config import MySQLConfig, SSLConfig
def test_validate_connection_config():
"""Test _validate_connection_config with valid configuration"""
config_data = {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"port": 3306,
"database": "test_db",
"user": "test_user",
"password": "test_pass"
}
}
result = MySQLConfig._validate_connection_config(config_data, "test_mysql")
assert result == config_data["test_mysql"]
def test_validate_connection_config_missing_db_name():
"""Test _validate_connection_config with missing db_name"""
config_data = {
"test_mysql": {
"type": "mysql",
"host": "localhost"
}
}
with pytest.raises(ValueError, match="Connection name must be specified"):
MySQLConfig._validate_connection_config(config_data, "")
def test_validate_connection_config_missing_connection():
"""Test _validate_connection_config with non-existent connection"""
config_data = {
"test_mysql": {
"type": "mysql",
"host": "localhost"
}
}
with pytest.raises(ValueError, match="Connection configuration not found"):
MySQLConfig._validate_connection_config(config_data, "nonexistent")
def test_validate_connection_config_missing_type():
"""Test _validate_connection_config with missing type"""
config_data = {
"test_mysql": {
"host": "localhost",
"user": "test_user",
"password": "test_pass"
}
}
with pytest.raises(ValueError, match="Connection configuration must include 'type' field"):
MySQLConfig._validate_connection_config(config_data, "test_mysql")
def test_validate_connection_config_wrong_type():
"""Test _validate_connection_config with wrong type"""
config_data = {
"test_mysql": {
"type": "postgres",
"host": "localhost",
"user": "test_user",
"password": "test_pass"
}
}
with pytest.raises(ValueError, match="Configuration is not MySQL type"):
MySQLConfig._validate_connection_config(config_data, "test_mysql")
def test_validate_connection_config_missing_user():
"""Test _validate_connection_config with missing user"""
config_data = {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"password": "test_pass"
}
}
with pytest.raises(ValueError, match="User must be specified"):
MySQLConfig._validate_connection_config(config_data, "test_mysql")
def test_validate_connection_config_missing_password():
"""Test _validate_connection_config with missing password"""
config_data = {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"user": "test_user"
}
}
with pytest.raises(ValueError, match="Password must be specified"):
MySQLConfig._validate_connection_config(config_data, "test_mysql")
def test_create_config_from_url():
"""Test _create_config_from_url with valid URL"""
db_config = {
"url": "mysql://localhost:3306/test_db?charset=utf8mb4",
"user": "test_user",
"password": "test_pass"
}
config = MySQLConfig._create_config_from_url(db_config)
assert config.host == "localhost"
assert config.port == "3306"
assert config.database == "test_db"
assert config.user == "test_user"
assert config.password == "test_pass"
assert config.charset == "utf8mb4"
assert config.url == "mysql://localhost:3306/test_db?charset=utf8mb4"
def test_create_config_from_url_with_ssl():
"""Test _create_config_from_url with SSL parameters"""
db_config = {
"url": "mysql://localhost:3306/test_db?ssl-mode=verify_identity&ssl-ca=/path/to/ca.pem",
"user": "test_user",
"password": "test_pass"
}
config = MySQLConfig._create_config_from_url(db_config)
assert config.ssl is not None
assert config.ssl.mode == "verify_identity"
assert config.ssl.ca == "/path/to/ca.pem"
def test_create_config_from_url_with_local_host():
"""Test _create_config_from_url with local_host parameter"""
db_config = {
"url": "mysql://db.example.com:3306/test_db",
"user": "test_user",
"password": "test_pass"
}
config = MySQLConfig._create_config_from_url(db_config, local_host="127.0.0.1")
assert config.host == "db.example.com"
assert config.local_host == "127.0.0.1"
def test_create_config_from_params():
"""Test _create_config_from_params with valid parameters"""
db_config = {
"database": "test_db",
"host": "localhost",
"port": 3306,
"user": "test_user",
"password": "test_pass",
"charset": "utf8mb4"
}
config = MySQLConfig._create_config_from_params(db_config)
assert config.database == "test_db"
assert config.host == "localhost"
assert config.port == "3306"
assert config.user == "test_user"
assert config.password == "test_pass"
assert config.charset == "utf8mb4"
def test_create_config_from_params_missing_database():
"""Test _create_config_from_params with missing database"""
db_config = {
"host": "localhost",
"port": 3306,
"user": "test_user",
"password": "test_pass"
}
with pytest.raises(ValueError, match="MySQL database name must be specified"):
MySQLConfig._create_config_from_params(db_config)
def test_create_config_from_params_missing_host():
"""Test _create_config_from_params with missing host"""
db_config = {
"database": "test_db",
"port": 3306,
"user": "test_user",
"password": "test_pass"
}
with pytest.raises(ValueError, match="Host must be specified"):
MySQLConfig._create_config_from_params(db_config)
def test_create_config_from_params_missing_port():
"""Test _create_config_from_params with missing port"""
db_config = {
"database": "test_db",
"host": "localhost",
"user": "test_user",
"password": "test_pass"
}
with pytest.raises(ValueError, match="Port must be specified"):
MySQLConfig._create_config_from_params(db_config)
def test_create_config_from_params_with_ssl():
"""Test _create_config_from_params with SSL configuration"""
db_config = {
"database": "test_db",
"host": "localhost",
"port": 3306,
"user": "test_user",
"password": "test_pass",
"ssl": {
"mode": "verify_identity",
"ca": "/path/to/ca.pem",
"cert": "/path/to/client-cert.pem",
"key": "/path/to/client-key.pem"
}
}
config = MySQLConfig._create_config_from_params(db_config)
assert config.ssl is not None
assert config.ssl.mode == "verify_identity"
assert config.ssl.ca == "/path/to/ca.pem"
assert config.ssl.cert == "/path/to/client-cert.pem"
assert config.ssl.key == "/path/to/client-key.pem"
def test_parse_ssl_config():
"""Test _parse_ssl_config with valid SSL config"""
db_config = {
"ssl": {
"mode": "verify_identity",
"ca": "/path/to/ca.pem",
"cert": "/path/to/client-cert.pem",
"key": "/path/to/client-key.pem"
}
}
ssl_config = MySQLConfig._parse_ssl_config(db_config)
assert ssl_config is not None
assert ssl_config.mode == "verify_identity"
assert ssl_config.ca == "/path/to/ca.pem"
assert ssl_config.cert == "/path/to/client-cert.pem"
assert ssl_config.key == "/path/to/client-key.pem"
def test_parse_ssl_config_missing_ssl():
"""Test _parse_ssl_config with missing SSL config"""
db_config = {
"host": "localhost",
"port": 3306
}
ssl_config = MySQLConfig._parse_ssl_config(db_config)
assert ssl_config is None
def test_parse_ssl_config_invalid_type():
"""Test _parse_ssl_config with invalid SSL config type"""
db_config = {
"ssl": "enable" # Should be a dict
}
with pytest.raises(ValueError, match="SSL configuration must be a dictionary"):
MySQLConfig._parse_ssl_config(db_config)
def test_parse_ssl_config_invalid_mode():
"""Test _parse_ssl_config with invalid SSL mode"""
db_config = {
"ssl": {
"mode": "invalid_mode",
"ca": "/path/to/ca.pem"
}
}
with pytest.raises(ValueError, match="Invalid ssl-mode"):
MySQLConfig._parse_ssl_config(db_config)