conftest.py

Code Hygiene Score: 93

Keine Issues gefunden.

Dependencies 8

Funktionen 7

Code

#!/usr/bin/env python3
"""Gemeinsame Test-Fixtures für MCP-DB Tests"""

import sys
from pathlib import Path

import pytest

# Pfad zum mcp-db Server hinzufügen
sys.path.insert(0, "/var/www/mcp-servers/mcp_db")

# .env laden BEVOR Config importiert wird
from dotenv import load_dotenv

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

from config import Config
from domain.log_contract import LogEntry
from infrastructure.protokoll_logger import ProtokollLogger


@pytest.fixture
def db_connection():
    """Fixture für Datenbank-Verbindung"""
    import mysql.connector

    conn = mysql.connector.connect(
        host=Config.DB_HOST,
        user=Config.DB_USER,
        password=Config.DB_PASSWORD,
        charset="utf8mb4",
    )
    yield conn
    conn.close()


@pytest.fixture
def ki_protokoll_connection():
    """Fixture für ki_protokoll Datenbank-Verbindung"""
    import mysql.connector

    conn = mysql.connector.connect(
        host=Config.DB_HOST,
        database="ki_protokoll",
        user=Config.DB_USER,
        password=Config.DB_PASSWORD,
        charset="utf8mb4",
    )
    yield conn
    conn.close()


@pytest.fixture
def ki_system_connection():
    """Fixture für ki_system Datenbank-Verbindung"""
    import mysql.connector

    conn = mysql.connector.connect(
        host=Config.DB_HOST,
        database="ki_system",
        user=Config.DB_USER,
        password=Config.DB_PASSWORD,
        charset="utf8mb4",
    )
    yield conn
    conn.close()


@pytest.fixture
def logger():
    """Fixture für ProtokollLogger"""
    return ProtokollLogger()


@pytest.fixture
def test_log_entry():
    """Fixture für Test LogEntry"""
    return LogEntry(
        request="TEST: SELECT * FROM mcp_log",
        status="success",
        duration_ms=42,
    )


@pytest.fixture
def test_denied_log_entry():
    """Fixture für Test Denied LogEntry"""
    return LogEntry(
        request="TEST: DROP TABLE mcp_log",
        status="denied",
        duration_ms=0,
        error_message="Blocked keyword: DROP",
    )


@pytest.fixture
def test_error_log_entry():
    """Fixture für Test Error LogEntry"""
    return LogEntry(
        request="TEST: SELECT * FROM nonexistent",
        status="error",
        duration_ms=15,
        error_message="Table doesn't exist",
    )
← Übersicht