test_step_type_mapping.py

• Pfad: /var/www/scripts/pipeline/tests/test_step_type_mapping.py
• Namespace: pipeline
• Zeilen: 122 | Größe: 3,201 Bytes
• Geändert: 2025-12-31 03:01:09 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 100

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

Keine Issues gefunden.

Dependencies 5

• use os
• use sys
• use db.db
• use pipeline_config.PipelineConfigError
• use pipeline_config.get_step_model

Funktionen 3

• test_step_types_exist_in_db() Zeile 31
• test_step_types_have_valid_models() Zeile 54
• test_no_duplicate_step_types() Zeile 77

Code

#!/usr/bin/env python3
"""
Test: Step-Type Mapping in DB vorhanden.

Supervision-Anforderung: Prüft ob alle verwendeten step_types in DB existieren.
Tippfehler in step_type = sofortiger Ausfall.
Task #516
"""

import os
import sys

# Add parent directory to path for imports
sys.path.insert(0, str(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))


# Step-Types die in der Pipeline verwendet werden
REQUIRED_STEP_TYPES = [
    "text_semantic_analyze",
    "knowledge_semantic_analyze",
    "entity_extract",
    "relation_extract",
    "ontology_classify",
    "taxonomy_build",
    "enrich",
    "embed",
    "vision",
]


def test_step_types_exist_in_db():
    """Prüft ob alle verwendeten step_types in DB existieren."""
    from db import db
    from pipeline_config import PipelineConfigError, get_step_model

    db.connect()

    missing = []
    found = []

    for step_type in REQUIRED_STEP_TYPES:
        try:
            model = get_step_model(step_type)
            found.append(f"{step_type}: {model}")
        except PipelineConfigError:
            missing.append(step_type)

    db.disconnect()

    if missing:
        raise AssertionError(f"Step-Types nicht in DB konfiguriert: {missing}\nGefunden: {found}")


def test_step_types_have_valid_models():
    """Prüft ob alle step_types gültige (nicht-leere) Models haben."""
    from db import db
    from pipeline_config import get_step_model

    db.connect()

    invalid = []

    for step_type in REQUIRED_STEP_TYPES:
        try:
            model = get_step_model(step_type)
            if not model or len(model.strip()) == 0:
                invalid.append(f"{step_type}: empty model")
        except Exception as e:
            invalid.append(f"{step_type}: {e}")

    db.disconnect()

    if invalid:
        raise AssertionError(f"Ungültige Models: {invalid}")


def test_no_duplicate_step_types():
    """Prüft ob keine Duplikate in pipeline_steps existieren."""
    from db import db

    db.connect()

    cursor = db.execute("""
        SELECT pipeline_id, step_type, COUNT(*) as cnt
        FROM pipeline_steps
        GROUP BY pipeline_id, step_type
        HAVING COUNT(*) > 1
    """)
    duplicates = cursor.fetchall()
    cursor.close()

    db.disconnect()

    if duplicates:
        raise AssertionError(f"Duplikate in pipeline_steps gefunden: {duplicates}")


if __name__ == "__main__":
    print("Running step_type mapping tests...")

    # Set DB password if not set
    if "DB_PASSWORD" not in os.environ:
        print("Warning: DB_PASSWORD not set, tests may fail")

    try:
        test_step_types_exist_in_db()
        print("✓ test_step_types_exist_in_db passed")
    except AssertionError as e:
        print(f"✗ test_step_types_exist_in_db FAILED:\n{e}")

    try:
        test_step_types_have_valid_models()
        print("✓ test_step_types_have_valid_models passed")
    except AssertionError as e:
        print(f"✗ test_step_types_have_valid_models FAILED:\n{e}")

    try:
        test_no_duplicate_step_types()
        print("✓ test_no_duplicate_step_types passed")
    except AssertionError as e:
        print(f"✗ test_no_duplicate_step_types FAILED:\n{e}")