test_db_operations.py

• Pfad: /var/www/mcp-servers/mcp-db/tests/test_db_operations.py
• Namespace: -
• Zeilen: 154 | Größe: 4,998 Bytes
• Geändert: 2025-12-28 12:21:49 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 98

• Dependencies: 90 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Keine Issues gefunden.

Dependencies 6

• use sys
• use pathlib.Path
• use pytest
• use dotenv.load_dotenv
• use config.Config
• use infrastructure.db_connection.DatabaseConnection

Klassen 3

• TestDbSelect class Zeile 19
• TestDbSchema class Zeile 69
• TestDbStats class Zeile 116

Code

#!/usr/bin/env python3
"""Tests für DB Operations (SELECT, SCHEMA, STATS)"""

import sys
from pathlib import Path

import pytest

sys.path.insert(0, "/var/www/mcp-servers/mcp_db")

from dotenv import load_dotenv

load_dotenv(Path("/var/www/mcp-servers/mcp_db/.env"))

from config import Config
from infrastructure.db_connection import DatabaseConnection


class TestDbSelect:
    """Test db_select Tool"""

    def test_simple_select_query(self, ki_protokoll_connection):
        """Test: Einfache SELECT Query"""
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM mcp_log ORDER BY timestamp DESC LIMIT 10")
        rows = cursor.fetchall()
        cursor.close()

        assert len(rows) >= 0

    def test_select_with_prepared_statement(self, ki_protokoll_connection):
        """Test: SELECT mit Prepared Statement"""
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        cursor.execute(
            "SELECT * FROM mcp_log WHERE status = %s LIMIT 5", ("success",)
        )
        rows = cursor.fetchall()
        cursor.close()

        assert isinstance(rows, list)

    def test_select_with_max_rows_limit(self, ki_protokoll_connection):
        """Test: SELECT mit max_rows Limit"""
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM mcp_log LIMIT 3")
        rows = cursor.fetchall()
        cursor.close()

        assert len(rows) <= 3

    def test_select_on_ki_system_database(self, ki_system_connection):
        """Test: SELECT auf ki_system Datenbank"""
        cursor = ki_system_connection.cursor(dictionary=True)
        cursor.execute("SELECT * FROM chunks LIMIT 5")
        rows = cursor.fetchall()
        cursor.close()

        assert isinstance(rows, list)

    def test_invalid_query_returns_error(self, ki_protokoll_connection):
        """Test: Ungültige Query gibt ERROR zurück"""
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        with pytest.raises(Exception):
            cursor.execute("SELECT * FROM nonexistent_table")
            cursor.fetchall()
        cursor.close()


class TestDbSchema:
    """Test db_schema Tool"""

    def test_schema_ki_protokoll(self):
        """Test: Schema von ki_protokoll abrufen"""
        assert "ki_protokoll" in Config.ALLOWED_DATABASES

        with DatabaseConnection.get_connection("ki_protokoll") as conn:
            cursor = conn.cursor(dictionary=True)
            cursor.execute(
                """SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME
                   FROM information_schema.TABLES
                   WHERE TABLE_SCHEMA = %s
                   AND TABLE_TYPE = 'BASE TABLE'
                   ORDER BY TABLE_NAME""",
                ("ki_protokoll",),
            )
            tables = cursor.fetchall()
            cursor.close()

        assert len(tables) > 0
        assert any(t["TABLE_NAME"] == "mcp_log" for t in tables)

    def test_schema_ki_system(self):
        """Test: Schema von ki_system abrufen"""
        assert "ki_system" in Config.ALLOWED_DATABASES

        with DatabaseConnection.get_connection("ki_system") as conn:
            cursor = conn.cursor(dictionary=True)
            cursor.execute(
                """SELECT TABLE_NAME, TABLE_ROWS, CREATE_TIME
                   FROM information_schema.TABLES
                   WHERE TABLE_SCHEMA = %s
                   AND TABLE_TYPE = 'BASE TABLE'
                   ORDER BY TABLE_NAME""",
                ("ki_system",),
            )
            tables = cursor.fetchall()
            cursor.close()

        assert len(tables) > 0

    def test_mysql_database_blocked(self):
        """Test: Nicht-erlaubte Datenbank (mysql) blockiert"""
        assert "mysql" not in Config.ALLOWED_DATABASES


class TestDbStats:
    """Test db_stats Tool"""

    def test_get_last_logs(self, ki_protokoll_connection):
        """Test: Letzte Logs abrufen (Limit=10)"""
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        cursor.execute(
            """SELECT id, timestamp, client_name, request, status,
                      duration_ms, error_message
               FROM mcp_log
               ORDER BY timestamp DESC
               LIMIT 10"""
        )
        logs = cursor.fetchall()
        cursor.close()

        assert isinstance(logs, list)

    def test_limit_parameter_works(self, ki_protokoll_connection):
        """Test: Limit Parameter funktioniert"""
        limit = 3
        cursor = ki_protokoll_connection.cursor(dictionary=True)
        cursor.execute(
            """SELECT id, timestamp, client_name, request, status,
                      duration_ms, error_message
               FROM mcp_log
               ORDER BY timestamp DESC
               LIMIT %s""",
            (limit,),
        )
        logs = cursor.fetchall()
        cursor.close()

        assert len(logs) <= limit


if __name__ == "__main__":
    pytest.main([__file__, "-v"])