{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp-db\/comprehensive_test.py",
"oldString": "result = mcp.tools['db_stats'](limit=200)\nresults.append((\"db_stats\", \"limit=200 clamped to 100\", \"success\",\n f\"count={result.get('count', 0)}\", 'logs' in result and result.get('count', 0) <= 100))",
"newString": "result = mcp.tools['db_stats'](limit=200)\nresults.append((\"db_stats\", \"limit=200 clamped to 100\", \"success\",\n f\"count={result.get('count', 0)}\", 'logs' in result and result.get('count', 0) <= DEFAULT_LIMIT))",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"\nMCP-DB Comprehensive Test Suite\nVollständiger Funktionstest aller MCP-DB Features\n\"\"\"\n\nimport json\nimport sys\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import List, Tuple\n\nsys.path.insert(0, '\/var\/www\/mcp-servers\/shared')\nfrom constants import DEFAULT_LIMIT, MAX_ROWS, PERCENT_FULL, PERCENT_HALF\n\n# Load environment\nfrom dotenv import load_dotenv\nload_dotenv(Path('\/opt\/mcp-servers\/mcp-db\/.env'))\n\n# Reset connection pool\nfrom infrastructure.db_connection import DatabaseConnection\nDatabaseConnection._pool = None\n\n# Import tools\nfrom tools.select_tool import register_select_tool\nfrom tools.schema_tool import register_schema_tool\nfrom tools.stats_tool import register_stats_tool\nfrom config import Config\n\n# Mock MCP\nclass MockMCP:\n def __init__(self):\n self.tools = {}\n def tool(self):\n def decorator(func):\n self.tools[func.__name__] = func\n return func\n return decorator\n\nmcp = MockMCP()\nregister_select_tool(mcp)\nregister_schema_tool(mcp)\nregister_stats_tool(mcp)\n\n# Test Results\nresults: List[Tuple[str, str, str, str, bool]] = []\n\ndef test(category: str, name: str, expected: str, func, *args, **kwargs):\n \"\"\"Run a test and record result\"\"\"\n try:\n result = func(*args, **kwargs)\n status = result.get('status', 'no_status')\n error = result.get('error', '')\n \n # Handle different response formats\n if expected == 'success':\n if status == 'success':\n passed = True\n elif status == 'no_status' and 'error' not in result:\n passed = True # db_schema\/db_stats without error\n elif status == 'no_status' and result.get('error') is None:\n passed = True\n else:\n passed = False\n elif expected == 'denied':\n if status == 'denied':\n passed = True\n elif status == 'no_status' and error:\n passed = True # db_schema returns {error: ..., tables: []}\n else:\n passed = False\n elif expected == 'error':\n passed = status == 'error'\n elif expected.startswith('denied:'):\n expected_msg = expected.split(':', 1)[1]\n if status == 'denied':\n passed = expected_msg.lower() in (error or '').lower()\n elif status == 'no_status' and error:\n passed = expected_msg.lower() in error.lower()\n else:\n passed = False\n elif expected.startswith('has:'):\n # Check if result has specific key\n key = expected.split(':', 1)[1]\n passed = key in result and result[key]\n else:\n passed = False\n \n actual_str = f\"{status}: {error[:50] if error else 'OK'}\"\n results.append((category, name, expected, actual_str, passed))\n except Exception as e:\n results.append((category, name, expected, f\"EXCEPTION: {str(e)[:50]}\", False))\n\n# ============================================================\n# TEST CATEGORY: db_select - Basic Functionality\n# ============================================================\n\ntest(\"db_select\/basic\", \"Simple SELECT\", \"success\",\n mcp.tools['db_select'], \n query=\"SELECT 1 as test\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/basic\", \"SELECT with columns\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT id, status FROM mcp_log LIMIT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/basic\", \"SELECT with WHERE\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log WHERE status = 'success' LIMIT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/basic\", \"SELECT with ORDER BY\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log ORDER BY id DESC LIMIT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/basic\", \"SELECT with COUNT\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT COUNT(*) as total FROM mcp_log\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/basic\", \"SELECT with GROUP BY\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT status, COUNT(*) FROM mcp_log GROUP BY status\",\n database=\"ki_protokoll\")\n\n# ============================================================\n# TEST CATEGORY: db_select - Prepared Statements\n# ============================================================\n\ntest(\"db_select\/params\", \"Single param\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log WHERE status = %s LIMIT 1\",\n database=\"ki_protokoll\",\n params=[\"success\"])\n\ntest(\"db_select\/params\", \"Multiple params\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log WHERE status = %s AND id > %s LIMIT 1\",\n database=\"ki_protokoll\",\n params=[\"success\", 0])\n\ntest(\"db_select\/params\", \"LIKE param\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log WHERE request LIKE %s LIMIT 1\",\n database=\"ki_protokoll\",\n params=[\"%SELECT%\"])\n\n# ============================================================\n# TEST CATEGORY: db_select - Database Allowlist\n# ============================================================\n\ntest(\"db_select\/db_allow\", \"ki_protokoll allowed\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/db_allow\", \"ki_system allowed\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1\",\n database=\"ki_system\")\n\ntest(\"db_select\/db_allow\", \"mysql denied\", \"denied:not allowed\",\n mcp.tools['db_select'],\n query=\"SELECT 1\",\n database=\"mysql\")\n\ntest(\"db_select\/db_allow\", \"information_schema denied\", \"denied:not allowed\",\n mcp.tools['db_select'],\n query=\"SELECT 1\",\n database=\"information_schema\")\n\n# ============================================================\n# TEST CATEGORY: db_select - Table Allowlist (existierende Tabellen)\n# ============================================================\n\n# ki_protokoll tables\ntest(\"db_select\/table_allow\", \"Table mcp_log (ki_protokoll)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1 FROM mcp_log LIMIT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/table_allow\", \"Table protokoll (ki_protokoll)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1 FROM protokoll LIMIT 1\",\n database=\"ki_protokoll\")\n\n# ki_system tables - sample\ntest(\"db_select\/table_allow\", \"Table documents (ki_system)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1 FROM documents LIMIT 1\",\n database=\"ki_system\")\n\ntest(\"db_select\/table_allow\", \"Table chunks (ki_system)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1 FROM chunks LIMIT 1\",\n database=\"ki_system\")\n\ntest(\"db_select\/table_allow\", \"Table content_orders (ki_system)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 1 FROM content_orders LIMIT 1\",\n database=\"ki_system\")\n\n# Denied tables\ntest(\"db_select\/table_allow\", \"Table users denied\", \"denied:not allowed\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM users\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/table_allow\", \"Table secrets denied\", \"denied:not allowed\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM secrets\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/table_allow\", \"Table mysql.user denied\", \"denied:not allowed\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mysql.user\",\n database=\"ki_protokoll\")\n\n# ============================================================\n# TEST CATEGORY: db_select - Keyword Blocklist (alle 15)\n# ============================================================\n\nblocked_tests = [\n (\"DROP\", \"SELECT * FROM mcp_log; DROP TABLE x\"),\n (\"DELETE\", \"SELECT * FROM mcp_log; DELETE FROM x\"),\n (\"INSERT\", \"SELECT * FROM mcp_log; INSERT INTO x VALUES(1)\"),\n (\"UPDATE\", \"SELECT * FROM mcp_log; UPDATE x SET y=1\"),\n (\"TRUNCATE\", \"SELECT * FROM mcp_log; TRUNCATE TABLE x\"),\n (\"ALTER\", \"SELECT * FROM mcp_log; ALTER TABLE x ADD y INT\"),\n (\"CREATE\", \"SELECT * FROM mcp_log; CREATE TABLE x(id INT)\"),\n (\"RENAME\", \"SELECT * FROM mcp_log; RENAME TABLE x TO y\"),\n (\"GRANT\", \"SELECT * FROM mcp_log; GRANT ALL ON x TO y\"),\n (\"REVOKE\", \"SELECT * FROM mcp_log; REVOKE ALL ON x FROM y\"),\n (\"LOAD_FILE\", \"SELECT LOAD_FILE('\/etc\/passwd')\"),\n (\"INTO OUTFILE\", \"SELECT * FROM mcp_log INTO OUTFILE '\/tmp\/x'\"),\n (\"INTO DUMPFILE\", \"SELECT * FROM mcp_log INTO DUMPFILE '\/tmp\/x'\"),\n (\"BENCHMARK\", \"SELECT BENCHMARK(1000000, MD5('test'))\"),\n (\"SLEEP\", \"SELECT SLEEP(10)\"),\n]\n\nfor keyword, query in blocked_tests:\n test(\"db_select\/blocklist\", f\"Keyword {keyword} blocked\", f\"denied:{keyword}\",\n mcp.tools['db_select'],\n query=query,\n database=\"ki_protokoll\")\n\n# Word boundary tests\ntest(\"db_select\/blocklist\", \"Word boundary: dropdown != DROP\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 'dropdown' as word FROM mcp_log LIMIT 1\",\n database=\"ki_protokoll\")\n\ntest(\"db_select\/blocklist\", \"Word boundary: updated_at != UPDATE\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT 'updated_at' as col FROM mcp_log LIMIT 1\",\n database=\"ki_protokoll\")\n\n# ============================================================\n# TEST CATEGORY: db_select - Limits\n# ============================================================\n\ntest(\"db_select\/limits\", \"max_rows=1\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log\",\n database=\"ki_protokoll\",\n max_rows=1)\n\ntest(\"db_select\/limits\", \"max_rows=50\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log\",\n database=\"ki_protokoll\",\n max_rows=PERCENT_HALF)\n\ntest(\"db_select\/limits\", \"max_rows=100 (max)\", \"success\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM mcp_log\",\n database=\"ki_protokoll\",\n max_rows=MAX_ROWS)\n\ntest(\"db_select\/limits\", \"max_rows=0 denied\", \"denied:max_rows\",\n mcp.tools['db_select'],\n query=\"SELECT * FROM ... [TRUNCATED-dbb8bd5027c4c63e]",
"structuredPatch": [
{
"oldStart": 346,
"oldLines": 7,
"newStart": 346,
"newLines": 7,
"lines": [
" ",
" result = mcp.tools['db_stats'](limit=200)",
" results.append((\"db_stats\", \"limit=200 clamped to 100\", \"success\",",
"- f\"count={result.get('count', 0)}\", 'logs' in result and result.get('count', 0) <= 100))",
"+ f\"count={result.get('count', 0)}\", 'logs' in result and result.get('count', 0) <= DEFAULT_LIMIT))",
" ",
" # ============================================================",
" # GENERATE REPORT"
]
}
],
"userModified": false,
"replaceAll": false
}
}