{
"event": "PreToolUse",
"tool_name": "Edit",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_validator.py",
"old_string": "import sys\nfrom pathlib import Path\n\nimport pytest\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")",
"new_string": "import 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"
}
}
{
"tool_response": {
"filePath": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_validator.py",
"oldString": "import sys\nfrom pathlib import Path\n\nimport pytest\n\nsys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")",
"newString": "import 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",
"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\")\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": 7,
"oldLines": 6,
"newStart": 7,
"newLines": 8,
"lines": [
" import pytest",
" ",
" sys.path.insert(0, \"\/opt\/mcp-servers\/mcp-db\")",
"+sys.path.insert(0, \"\/var\/www\/mcp-servers\/shared\")",
"+from constants import PERCENT_HALF",
" ",
" from dotenv import load_dotenv",
" "
]
}
],
"userModified": false,
"replaceAll": false
}
}