Protokoll #15633

{
    "event": "PreToolUse",
    "tool_name": "Write",
    "tool_input": {
        "file_path": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_validator.py",
        "content": "#!\/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"
    }
}

{
    "tool_response": {
        "type": "create",
        "filePath": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_validator.py",
        "content": "#!\/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": [],
        "originalFile": null
    }
}