{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_validator.py",
"oldString": ", 50",
"newString": ", PERCENT_HALF",
"originalFile": "#!\/usr\/bin\/env python3\n\"\"\"Tests für QueryValidator\"\"\"\n\nimport sys\nfrom pathlib import Path\n\nimport pytest\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")\nsys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")\nfrom constants import PERCENT_HALF\n\nfrom dotenv import load_dotenv\n\nload_dotenv(Path(\"\/opt\/mcp-servers\/mcp-db\/.env\"))\n\nfrom validators.query_validator import QueryValidator\n\n\nclass TestQueryValidator:\n \"\"\"Test QueryValidator Funktionalität\"\"\"\n\n def test_simple_valid_query(self):\n \"\"\"Test: Einfache gültige Query\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log\", \"ki_protokoll\", 50\n )\n assert valid is True\n assert error == \"\"\n\n def test_drop_keyword_blocked(self):\n \"\"\"Test: DROP Keyword blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log; DROP TABLE mcp_log\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"DROP\" in error\n\n def test_sleep_keyword_blocked(self):\n \"\"\"Test: SLEEP Keyword blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT SLEEP(10) FROM mcp_log\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"SLEEP\" in error\n\n def test_insert_blocked(self):\n \"\"\"Test: INSERT blockiert (kein SELECT)\"\"\"\n valid, error = QueryValidator.validate_query(\n \"INSERT INTO mcp_log VALUES (1,2,3)\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"Only SELECT\" in error\n\n def test_update_keyword_blocked(self):\n \"\"\"Test: UPDATE Keyword 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 assert valid is False\n assert \"UPDATE\" in error\n\n def test_delete_keyword_blocked(self):\n \"\"\"Test: DELETE Keyword blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log; DELETE FROM mcp_log\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"DELETE\" in error\n\n def test_load_file_blocked(self):\n \"\"\"Test: LOAD_FILE blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT LOAD_FILE('\/etc\/passwd')\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"LOAD_FILE\" in error\n\n def test_mysql_database_blocked(self):\n \"\"\"Test: mysql Datenbank blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\"SELECT * FROM users\", \"mysql\", 50)\n assert valid is False\n assert \"not allowed\" in error\n\n def test_users_table_blocked(self):\n \"\"\"Test: users Tabelle blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM users\", \"ki_protokoll\", 50\n )\n assert valid is False\n assert \"not allowed\" in error\n\n def test_query_too_long(self):\n \"\"\"Test: Query > 2000 Zeichen blockiert\"\"\"\n long_query = \"SELECT * FROM mcp_log WHERE \" + \"id=1 OR \" * 500\n valid, error = QueryValidator.validate_query(long_query, \"ki_protokoll\", 50)\n assert valid is False\n assert \"max\" in error\n assert \"chars\" in error\n\n def test_max_rows_too_high(self):\n \"\"\"Test: max_rows > 100 blockiert\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM mcp_log\", \"ki_protokoll\", 999\n )\n assert valid is False\n assert \"max_rows\" in error\n\n def test_ki_system_database_allowed(self):\n \"\"\"Test: ki_system Datenbank erlaubt\"\"\"\n valid, error = QueryValidator.validate_query(\n \"SELECT * FROM chunks\", \"ki_system\", 50\n )\n assert valid is True\n assert error == \"\"\n\n\nif __name__ == \"__main__\":\n pytest.main([__file__, \"-v\"])\n",
"structuredPatch": [
{
"oldStart": 23,
"oldLines": 7,
"newStart": 23,
"newLines": 7,
"lines": [
" def test_simple_valid_query(self):",
" \"\"\"Test: Einfache gültige Query\"\"\"",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM mcp_log\", \"ki_protokoll\", 50",
"+ \"SELECT * FROM mcp_log\", \"ki_protokoll\", PERCENT_HALF",
" )",
" assert valid is True",
" assert error == \"\""
]
},
{
"oldStart": 31,
"oldLines": 7,
"newStart": 31,
"newLines": 7,
"lines": [
" def test_drop_keyword_blocked(self):",
" \"\"\"Test: DROP Keyword blockiert\"\"\"",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM mcp_log; DROP TABLE mcp_log\", \"ki_protokoll\", 50",
"+ \"SELECT * FROM mcp_log; DROP TABLE mcp_log\", \"ki_protokoll\", PERCENT_HALF",
" )",
" assert valid is False",
" assert \"DROP\" in error"
]
},
{
"oldStart": 39,
"oldLines": 7,
"newStart": 39,
"newLines": 7,
"lines": [
" def test_sleep_keyword_blocked(self):",
" \"\"\"Test: SLEEP Keyword 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",
" )",
" assert valid is False",
" assert \"SLEEP\" in error"
]
},
{
"oldStart": 47,
"oldLines": 7,
"newStart": 47,
"newLines": 7,
"lines": [
" def test_insert_blocked(self):",
" \"\"\"Test: INSERT blockiert (kein SELECT)\"\"\"",
" 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",
" )",
" assert valid is False",
" assert \"Only SELECT\" in error"
]
},
{
"oldStart": 65,
"oldLines": 7,
"newStart": 65,
"newLines": 7,
"lines": [
" def test_delete_keyword_blocked(self):",
" \"\"\"Test: DELETE Keyword 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",
" )",
" assert valid is False",
" assert \"DELETE\" in error"
]
},
{
"oldStart": 73,
"oldLines": 21,
"newStart": 73,
"newLines": 21,
"lines": [
" def test_load_file_blocked(self):",
" \"\"\"Test: 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",
" )",
" assert valid is False",
" assert \"LOAD_FILE\" in error",
" ",
" def test_mysql_database_blocked(self):",
" \"\"\"Test: mysql Datenbank blockiert\"\"\"",
"- valid, error = QueryValidator.validate_query(\"SELECT * FROM users\", \"mysql\", 50)",
"+ valid, error = QueryValidator.validate_query(\"SELECT * FROM users\", \"mysql\", PERCENT_HALF)",
" assert valid is False",
" assert \"not allowed\" in error",
" ",
" def test_users_table_blocked(self):",
" \"\"\"Test: users Tabelle blockiert\"\"\"",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM users\", \"ki_protokoll\", 50",
"+ \"SELECT * FROM users\", \"ki_protokoll\", PERCENT_HALF",
" )",
" assert valid is False",
" assert \"not allowed\" in error"
]
},
{
"oldStart": 95,
"oldLines": 7,
"newStart": 95,
"newLines": 7,
"lines": [
" def test_query_too_long(self):",
" \"\"\"Test: Query > 2000 Zeichen blockiert\"\"\"",
" long_query = \"SELECT * FROM mcp_log WHERE \" + \"id=1 OR \" * 500",
"- valid, error = QueryValidator.validate_query(long_query, \"ki_protokoll\", 50)",
"+ valid, error = QueryValidator.validate_query(long_query, \"ki_protokoll\", PERCENT_HALF)",
" assert valid is False",
" assert \"max\" in error",
" assert \"chars\" in error"
]
},
{
"oldStart": 111,
"oldLines": 7,
"newStart": 111,
"newLines": 7,
"lines": [
" def test_ki_system_database_allowed(self):",
" \"\"\"Test: ki_system Datenbank erlaubt\"\"\"",
" valid, error = QueryValidator.validate_query(",
"- \"SELECT * FROM chunks\", \"ki_system\", 50",
"+ \"SELECT * FROM chunks\", \"ki_system\", PERCENT_HALF",
" )",
" assert valid is True",
" assert error == \"\""
]
}
],
"userModified": false,
"replaceAll": true
}
}