{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp-db\/test_all_functions.py",
"oldString": ", 50",
"newString": ", PERCENT_HALF",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"Umfassende Tests für MCP-DB Server - Alle Funktionen\"\"\"\n\nimport sys\nfrom pathlib import Path\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\nfrom constants import DEFAULT_LIMIT, MAX_ROWS, PERCENT_HALF\n\n# .env laden BEVOR Config importiert wird\nfrom dotenv import load_dotenv\nload_dotenv(Path(\"\/opt\/mcp-servers\/mcp-db\/.env\"))\n\nfrom config import Config\nfrom validators.query_validator import QueryValidator\nfrom domain.query_contract import QueryRequest, QueryResponse, QueryStatus\nfrom tools.select_tool import SelectExecutor\nfrom infrastructure.db_connection import DatabaseConnection\nfrom infrastructure.protokoll_logger import ProtokollLogger\n\n\nclass TestReport:\n \"\"\"Test-Bericht Klasse\"\"\"\n\n def __init__(self):\n self.tests = []\n self.passed = 0\n self.failed = 0\n\n def add_test(self, category: str, name: str, success: bool, details: str = \"\"):\n \"\"\"Test hinzufügen\"\"\"\n status = \"PASS\" if success else \"FAIL\"\n self.tests.append(\n {\n \"category\": category,\n \"name\": name,\n \"status\": status,\n \"success\": success,\n \"details\": details,\n }\n )\n if success:\n self.passed += 1\n else:\n self.failed += 1\n\n def print_report(self):\n \"\"\"Bericht ausgeben\"\"\"\n print(\"\\n\" + \"=\" * 80)\n print(\"MCP-DB SERVER TEST REPORT\")\n print(\"=\" * 80 + \"\\n\")\n\n current_category = None\n for test in self.tests:\n if test[\"category\"] != current_category:\n current_category = test[\"category\"]\n print(f\"\\n{current_category}\")\n print(\"-\" * 80)\n\n status_symbol = \"✓\" if test[\"success\"] else \"✗\"\n print(f\"{status_symbol} [{test['status']}] {test['name']}\")\n if test[\"details\"]:\n print(f\" → {test['details']}\")\n\n print(\"\\n\" + \"=\" * 80)\n print(f\"SUMMARY: {self.passed} passed, {self.failed} failed\")\n print(\"=\" * 80 + \"\\n\")\n\n\ndef test_validator(report: TestReport):\n \"\"\"Test QueryValidator\"\"\"\n print(\"\\n[1\/6] Testing QueryValidator...\")\n\n # Test 1.1: Einfache gültige Query\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log\", \"ki_protokoll\", PERCENT_HALF\n )\n report.add_test(\n \"VALIDATOR\",\n \"Einfache gültige Query\",\n valid and error == \"\",\n f\"valid={valid}, error={error}\",\n )\n\n # Test 1.2: DROP blockiert\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log; DROP TABLE mcp_log\", \"ki_protokoll\", PERCENT_HALF\n )\n report.add_test(\n \"VALIDATOR\",\n \"DROP Keyword blockiert\",\n not valid and \"DROP\" in error,\n f\"error={error}\",\n )\n\n # Test 1.3: SLEEP blockiert\n valid, error = QueryValidator.validate_query(\n \"SELECT SLEEP(10) FROM mcp_log\", \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"SLEEP Keyword blockiert\",\n not valid and \"SLEEP\" in error,\n f\"error={error}\",\n )\n\n # Test 1.4: INSERT blockiert\n valid, error = QueryValidator.validate_query(\n \"INSERT INTO mcp_log VALUES (1,2,3)\", \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"INSERT blockiert (kein SELECT)\",\n not valid and \"Only SELECT\" in error,\n f\"error={error}\",\n )\n\n # Test 1.5: UPDATE blockiert\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log WHERE status='denied' OR 1=1; UPDATE mcp_log SET status='success'\",\n \"ki_protokoll\",\n 50,\n )\n report.add_test(\n \"VALIDATOR\",\n \"UPDATE Keyword blockiert\",\n not valid and \"UPDATE\" in error,\n f\"error={error}\",\n )\n\n # Test 1.6: DELETE blockiert\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log; DELETE FROM mcp_log\", \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"DELETE Keyword blockiert\",\n not valid and \"DELETE\" in error,\n f\"error={error}\",\n )\n\n # Test 1.7: LOAD_FILE blockiert\n valid, error = QueryValidator.validate_query(\n \"SELECT LOAD_FILE('\/etc\/passwd')\", \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"LOAD_FILE blockiert\",\n not valid and \"LOAD_FILE\" in error,\n f\"error={error}\",\n )\n\n # Test 1.8: Nicht-erlaubte Datenbank\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM users\", \"mysql\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"mysql Datenbank blockiert\",\n not valid and \"not allowed\" in error,\n f\"error={error}\",\n )\n\n # Test 1.9: Nicht-erlaubte Tabelle\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM users\", \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"users Tabelle blockiert\",\n not valid and \"not allowed\" in error,\n f\"error={error}\",\n )\n\n # Test 1.10: Query zu lang\n long_query = \"SELECT * FROM mcp_log WHERE \" + \"id=1 OR \" * 500\n valid, error = QueryValidator.validate_query(\n long_query, \"ki_protokoll\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"Query > 2000 Zeichen blockiert\",\n not valid and \"max\" in error and \"chars\" in error,\n f\"query_length={len(long_query)}, error={error}\",\n )\n\n # Test 1.11: max_rows zu hoch\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log\", \"ki_protokoll\", 999\n )\n report.add_test(\n \"VALIDATOR\",\n \"max_rows > 100 blockiert\",\n not valid and \"max_rows\" in error,\n f\"error={error}\",\n )\n\n # Test 1.12: Erlaubte Datenbank ki_system\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM chunks\", \"ki_system\", 50\n )\n report.add_test(\n \"VALIDATOR\",\n \"ki_system Datenbank erlaubt\",\n valid and error == \"\",\n f\"valid={valid}\",\n )\n\n\ndef test_db_select(report: TestReport):\n \"\"\"Test db_select Tool\"\"\"\n print(\"\\n[2\/6] Testing db_select Tool...\")\n\n # Test 2.1: Einfache SELECT Query\n try:\n import mysql.connector\n\n conn = mysql.connector.connect(\n host=Config.DB_HOST,\n database=\"ki_protokoll\",\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\"\n )\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM mcp_log ORDER BY timestamp DESC LIMIT 10\")\n rows = cursor.fetchall()\n cursor.close()\n conn.close()\n\n success = len(rows) >= 0\n report.add_test(\n \"DB_SELECT\",\n \"Einfache SELECT Query\",\n success,\n f\"rows={len(rows)}\",\n )\n except Exception as e:\n report.add_test(\"DB_SELECT\", \"Einfache SELECT Query\", False, f\"Exception: {e}\")\n\n # Test 2.2: SELECT mit Prepared Statement\n try:\n import mysql.connector\n\n conn = mysql.connector.connect(\n host=Config.DB_HOST,\n database=\"ki_protokoll\",\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\"\n )\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM mcp_log WHERE status = %s LIMIT 5\", (\"success\",))\n rows = cursor.fetchall()\n cursor.close()\n conn.close()\n\n success = isinstance(rows, list)\n report.add_test(\n \"DB_SELECT\",\n \"SELECT mit Prepared Statement\",\n success,\n f\"rows={len(rows)}\",\n )\n except Exception as e:\n report.add_test(\n \"DB_SELECT\", \"SELECT mit Prepared Statement\", False, f\"Exception: {e}\"\n )\n\n # Test 2.3: SELECT mit max_rows Limit\n try:\n import mysql.connector\n\n conn = mysql.connector.connect(\n host=Config.DB_HOST,\n database=\"ki_protokoll\",\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\"\n )\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM mcp_log LIMIT 3\")\n rows = cursor.fetchall()\n cursor.close()\n conn.close()\n\n success = len(rows) <= 3\n report.add_test(\n \"DB_SELECT\",\n \"SELECT mit max_rows Limit\",\n success,\n f\"max_rows=3, actual_rows={len(rows)}\",\n )\n except Exception as e:\n report.add_test(\n \"DB_SELECT\", \"SELECT mit max_rows Limit\", False, f\"Exception: {e}\"\n )\n\n # Test 2.4: SELECT auf ki_system Datenbank\n try:\n import mysql.connector\n\n conn = mysql.connector.connect(\n host=Config.DB_HOST,\n database=\"ki_system\",\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\"\n )\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM chunks LIMIT 5\")\n rows = cursor.fetchall()\n cursor.close()\n conn.close()\n\n success = isinstance(rows, list)\n report.add_test(\n \"DB_SELECT\",\n \"SELECT auf ki_system Datenbank\",\n success,\n f\"rows={len(rows)}\",\n )\n except Exception as e:\n report.add_test(\n \"DB_SELECT\", \"SELECT auf ki_system Datenbank\", False, f\"Exception: {e}\"\n )\n\n # Test 2.5: Ungültige Query führt zu Error\n try:\n import mysql.connector\n\n conn = mysql.connector.connect(\n host=Config.DB_HOST,\n database=\"ki_protokoll\",\n user=Config.DB_USER,\n password=Config.DB_PASSWORD,\n charset=\"utf8mb4\"\n )\n cursor = conn.cursor(dictionary=True)\n try:\n cursor.execute(\"SELECT * FROM nonexistent_table\")\n rows = cursor.fe... [TRUNCATED-e7f348c2d9847f18]",
"structuredPatch": [
{
"oldStart": 96,
"oldLines": 7,
"newStart": 96,
"newLines": 7,
"lines": [
" ",
" # Test 1.3: SLEEP blockiert",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT SLEEP(10) FROM mcp_log\", \"ki_protokoll\", 50",
"+ \"SELECT SLEEP(10) FROM mcp_log\", \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 107,
"oldLines": 7,
"newStart": 107,
"newLines": 7,
"lines": [
" ",
" # Test 1.4: INSERT blockiert",
" valid, error = QueryValidator.validate_query(",
"- \"INSERT INTO mcp_log VALUES (1,2,3)\", \"ki_protokoll\", 50",
"+ \"INSERT INTO mcp_log VALUES (1,2,3)\", \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 131,
"oldLines": 7,
"newStart": 131,
"newLines": 7,
"lines": [
" ",
" # Test 1.6: DELETE blockiert",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM mcp_log; DELETE FROM mcp_log\", \"ki_protokoll\", 50",
"+ \"SELECT * FROM mcp_log; DELETE FROM mcp_log\", \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 142,
"oldLines": 7,
"newStart": 142,
"newLines": 7,
"lines": [
" ",
" # Test 1.7: LOAD_FILE blockiert",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT LOAD_FILE('\/etc\/passwd')\", \"ki_protokoll\", 50",
"+ \"SELECT LOAD_FILE('\/etc\/passwd')\", \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 153,
"oldLines": 7,
"newStart": 153,
"newLines": 7,
"lines": [
" ",
" # Test 1.8: Nicht-erlaubte Datenbank",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM users\", \"mysql\", 50",
"+ \"SELECT * FROM users\", \"mysql\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 164,
"oldLines": 7,
"newStart": 164,
"newLines": 7,
"lines": [
" ",
" # Test 1.9: Nicht-erlaubte Tabelle",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM users\", \"ki_protokoll\", 50",
"+ \"SELECT * FROM users\", \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 176,
"oldLines": 7,
"newStart": 176,
"newLines": 7,
"lines": [
" # Test 1.10: Query zu lang",
" long_query = \"SELECT * FROM mcp_log WHERE \" + \"id=1 OR \" * 500",
" valid, error = QueryValidator.validate_query(",
"- long_query, \"ki_protokoll\", 50",
"+ long_query, \"ki_protokoll\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
},
{
"oldStart": 198,
"oldLines": 7,
"newStart": 198,
"newLines": 7,
"lines": [
" ",
" # Test 1.12: Erlaubte Datenbank ki_system",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM chunks\", \"ki_system\", 50",
"+ \"SELECT * FROM chunks\", \"ki_system\", PERCENT_HALF",
" )",
" report.add_test(",
" \"VALIDATOR\","
]
}
],
"userModified": false,
"replaceAll": true
}
}