mcp-dbutils
by donghao1393
- tests
- integration
"""Test MySQL configuration parsing"""
import tempfile
import pytest
import yaml
from mcp_dbutils.mysql.config import MySQLConfig
def test_basic_config():
"""Test basic MySQL configuration parsing"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"port": 3306,
"database": "test_db",
"user": "test_user",
"password": "test_pass",
"charset": "utf8mb4"
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
config = MySQLConfig.from_yaml(tmp.name, "test_mysql")
assert config.type == "mysql"
assert config.host == "localhost"
assert config.port == "3306" # port is converted to string
assert config.database == "test_db"
assert config.user == "test_user"
assert config.password == "test_pass"
assert config.charset == "utf8mb4"
assert config.ssl is None
def test_url_config():
"""Test MySQL URL configuration parsing"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"url": "mysql://localhost:3306/test_db?charset=utf8mb4",
"user": "test_user",
"password": "test_pass"
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
config = MySQLConfig.from_yaml(tmp.name, "test_mysql")
assert config.type == "mysql"
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"
def test_ssl_config():
"""Test MySQL SSL configuration parsing"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"port": 3306,
"database": "test_db",
"user": "test_user",
"password": "test_pass",
"charset": "utf8mb4",
"ssl": {
"mode": "verify_identity",
"ca": "/path/to/ca.pem",
"cert": "/path/to/client-cert.pem",
"key": "/path/to/client-key.pem"
}
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
config = MySQLConfig.from_yaml(tmp.name, "test_mysql")
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_ssl_url_config():
"""Test MySQL SSL configuration via URL parameters"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"url": "mysql://localhost:3306/test_db?ssl-mode=verify_identity&ssl-ca=/path/to/ca.pem",
"user": "test_user",
"password": "test_pass"
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
config = MySQLConfig.from_yaml(tmp.name, "test_mysql")
assert config.ssl is not None
assert config.ssl.mode == "verify_identity"
assert config.ssl.ca == "/path/to/ca.pem"
def test_invalid_type():
"""Test configuration with invalid database type"""
config_data = {
"connections": {
"test_mysql": {
"type": "postgres",
"host": "localhost",
"database": "test_db",
"user": "test_user",
"password": "test_pass"
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
with pytest.raises(ValueError, match="Configuration is not MySQL type"):
MySQLConfig.from_yaml(tmp.name, "test_mysql")
def test_missing_required_fields():
"""Test configuration with missing required fields"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"host": "localhost"
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
with pytest.raises(ValueError, match="User must be specified"):
MySQLConfig.from_yaml(tmp.name, "test_mysql")
def test_invalid_ssl_mode():
"""Test configuration with invalid SSL mode"""
config_data = {
"connections": {
"test_mysql": {
"type": "mysql",
"host": "localhost",
"port": 3306,
"database": "test_db",
"user": "test_user",
"password": "test_pass",
"ssl": {
"mode": "invalid_mode"
}
}
}
}
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml') as tmp:
yaml.dump(config_data, tmp)
tmp.flush()
with pytest.raises(ValueError, match="Invalid ssl-mode"):
MySQLConfig.from_yaml(tmp.name, "test_mysql")
def test_masked_connection_info():
"""Test that sensitive connection information is properly masked"""
config = MySQLConfig(
database="test_db",
user="test_user",
password="secret",
host="localhost",
port="3306",
charset="utf8mb4"
)
masked_info = config.get_masked_connection_info()
assert "password" not in masked_info
assert masked_info["database"] == "test_db"
assert masked_info["host"] == "localhost"
assert masked_info["port"] == "3306"
assert masked_info["charset"] == "utf8mb4"