{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/mcp-servers\/mcp-db\/tests\/test_db_operations.py",
"content": "#!\/usr\/bin\/env python3\n\"\"\"Tests für DB Operations (SELECT, SCHEMA, STATS)\"\"\"\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 config import Config\nfrom infrastructure.db_connection import DatabaseConnection\n\n\nclass TestDbSelect:\n \"\"\"Test db_select Tool\"\"\"\n\n def test_simple_select_query(self, ki_protokoll_connection):\n \"\"\"Test: Einfache SELECT Query\"\"\"\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM mcp_log ORDER BY timestamp DESC LIMIT 10\")\n rows = cursor.fetchall()\n cursor.close()\n\n assert len(rows) >= 0\n\n def test_select_with_prepared_statement(self, ki_protokoll_connection):\n \"\"\"Test: SELECT mit Prepared Statement\"\"\"\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n cursor.execute(\n \"SELECT * FROM mcp_log WHERE status = %s LIMIT 5\", (\"success\",)\n )\n rows = cursor.fetchall()\n cursor.close()\n\n assert isinstance(rows, list)\n\n def test_select_with_max_rows_limit(self, ki_protokoll_connection):\n \"\"\"Test: SELECT mit max_rows Limit\"\"\"\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM mcp_log LIMIT 3\")\n rows = cursor.fetchall()\n cursor.close()\n\n assert len(rows) <= 3\n\n def test_select_on_ki_system_database(self, ki_system_connection):\n \"\"\"Test: SELECT auf ki_system Datenbank\"\"\"\n cursor = ki_system_connection.cursor(dictionary=True)\n cursor.execute(\"SELECT * FROM chunks LIMIT 5\")\n rows = cursor.fetchall()\n cursor.close()\n\n assert isinstance(rows, list)\n\n def test_invalid_query_returns_error(self, ki_protokoll_connection):\n \"\"\"Test: Ungültige Query gibt ERROR zurück\"\"\"\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n with pytest.raises(Exception):\n cursor.execute(\"SELECT * FROM nonexistent_table\")\n cursor.fetchall()\n cursor.close()\n\n\nclass TestDbSchema:\n \"\"\"Test db_schema Tool\"\"\"\n\n def test_schema_ki_protokoll(self):\n \"\"\"Test: Schema von ki_protokoll abrufen\"\"\"\n assert \"ki_protokoll\" in Config.ALLOWED_DATABASES\n\n with DatabaseConnection.get_connection(\"ki_protokoll\") as conn:\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\n \"\"\"SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME\n FROM information_schema.TABLES\n WHERE TABLE_SCHEMA = %s\n AND TABLE_TYPE = 'BASE TABLE'\n ORDER BY TABLE_NAME\"\"\",\n (\"ki_protokoll\",),\n )\n tables = cursor.fetchall()\n cursor.close()\n\n assert len(tables) > 0\n assert any(t[\"TABLE_NAME\"] == \"mcp_log\" for t in tables)\n\n def test_schema_ki_system(self):\n \"\"\"Test: Schema von ki_system abrufen\"\"\"\n assert \"ki_system\" in Config.ALLOWED_DATABASES\n\n with DatabaseConnection.get_connection(\"ki_system\") as conn:\n cursor = conn.cursor(dictionary=True)\n cursor.execute(\n \"\"\"SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME\n FROM information_schema.TABLES\n WHERE TABLE_SCHEMA = %s\n AND TABLE_TYPE = 'BASE TABLE'\n ORDER BY TABLE_NAME\"\"\",\n (\"ki_system\",),\n )\n tables = cursor.fetchall()\n cursor.close()\n\n assert len(tables) > 0\n\n def test_mysql_database_blocked(self):\n \"\"\"Test: Nicht-erlaubte Datenbank (mysql) blockiert\"\"\"\n assert \"mysql\" not in Config.ALLOWED_DATABASES\n\n\nclass TestDbStats:\n \"\"\"Test db_stats Tool\"\"\"\n\n def test_get_last_logs(self, ki_protokoll_connection):\n \"\"\"Test: Letzte Logs abrufen (Limit=10)\"\"\"\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n cursor.execute(\n \"\"\"SELECT id, timestamp, client_name, request, status,\n duration_ms, error_message\n FROM mcp_log\n ORDER BY timestamp DESC\n LIMIT 10\"\"\"\n )\n logs = cursor.fetchall()\n cursor.close()\n\n assert isinstance(logs, list)\n\n def test_limit_parameter_works(self, ki_protokoll_connection):\n \"\"\"Test: Limit Parameter funktioniert\"\"\"\n limit = 3\n cursor = ki_protokoll_connection.cursor(dictionary=True)\n cursor.execute(\n \"\"\"SELECT id, timestamp, client_name, request, status,\n duration_ms, error_message\n FROM mcp_log\n ORDER BY timestamp DESC\n LIMIT %s\"\"\",\n (limit,),\n )\n logs = cursor.fetchall()\n cursor.close()\n\n assert len(logs) <= limit\n\n\nif __name__ == \"__main__\":\n pytest.main([__file__, \"-v\"])\n"
}
}