{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/scripts\/pipeline\/tests\/test_step_type_mapping.py",
"content": "#!\/usr\/bin\/env python3\n\"\"\"\nTest: Step-Type Mapping in DB vorhanden.\n\nSupervision-Anforderung: Prüft ob alle verwendeten step_types in DB existieren.\nTippfehler in step_type = sofortiger Ausfall.\nTask #516\n\"\"\"\n\nimport os\nimport sys\n\n# Add parent directory to path for imports\nsys.path.insert(0, str(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))\n\n\n# Step-Types die in der Pipeline verwendet werden\nREQUIRED_STEP_TYPES = [\n \"text_semantic_analyze\",\n \"knowledge_semantic_analyze\",\n \"entity_extract\",\n \"relation_extract\",\n \"ontology_classify\",\n \"taxonomy_build\",\n \"enrich\",\n \"embed\",\n \"vision\",\n]\n\n\ndef test_step_types_exist_in_db():\n \"\"\"Prüft ob alle verwendeten step_types in DB existieren.\"\"\"\n from pipeline_config import get_step_model, PipelineConfigError\n from db import db\n\n db.connect()\n\n missing = []\n found = []\n\n for step_type in REQUIRED_STEP_TYPES:\n try:\n model = get_step_model(step_type)\n found.append(f\"{step_type}: {model}\")\n except PipelineConfigError as e:\n missing.append(step_type)\n\n db.disconnect()\n\n if missing:\n raise AssertionError(\n f\"Step-Types nicht in DB konfiguriert: {missing}\\n\"\n f\"Gefunden: {found}\"\n )\n\n\ndef test_step_types_have_valid_models():\n \"\"\"Prüft ob alle step_types gültige (nicht-leere) Models haben.\"\"\"\n from pipeline_config import get_step_model\n from db import db\n\n db.connect()\n\n invalid = []\n\n for step_type in REQUIRED_STEP_TYPES:\n try:\n model = get_step_model(step_type)\n if not model or len(model.strip()) == 0:\n invalid.append(f\"{step_type}: empty model\")\n except Exception as e:\n invalid.append(f\"{step_type}: {e}\")\n\n db.disconnect()\n\n if invalid:\n raise AssertionError(f\"Ungültige Models: {invalid}\")\n\n\ndef test_no_duplicate_step_types():\n \"\"\"Prüft ob keine Duplikate in pipeline_steps existieren.\"\"\"\n from db import db\n\n db.connect()\n\n cursor = db.execute(\"\"\"\n SELECT pipeline_id, step_type, COUNT(*) as cnt\n FROM pipeline_steps\n GROUP BY pipeline_id, step_type\n HAVING COUNT(*) > 1\n \"\"\")\n duplicates = cursor.fetchall()\n cursor.close()\n\n db.disconnect()\n\n if duplicates:\n raise AssertionError(\n f\"Duplikate in pipeline_steps gefunden: {duplicates}\"\n )\n\n\nif __name__ == \"__main__\":\n print(\"Running step_type mapping tests...\")\n\n # Set DB password if not set\n if \"DB_PASSWORD\" not in os.environ:\n print(\"Warning: DB_PASSWORD not set, tests may fail\")\n\n try:\n test_step_types_exist_in_db()\n print(\"✓ test_step_types_exist_in_db passed\")\n except AssertionError as e:\n print(f\"✗ test_step_types_exist_in_db FAILED:\\n{e}\")\n\n try:\n test_step_types_have_valid_models()\n print(\"✓ test_step_types_have_valid_models passed\")\n except AssertionError as e:\n print(f\"✗ test_step_types_have_valid_models FAILED:\\n{e}\")\n\n try:\n test_no_duplicate_step_types()\n print(\"✓ test_no_duplicate_step_types passed\")\n except AssertionError as e:\n print(f\"✗ test_no_duplicate_step_types FAILED:\\n{e}\")\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/scripts\/pipeline\/tests\/test_step_type_mapping.py",
"content": "#!\/usr\/bin\/env python3\n\"\"\"\nTest: Step-Type Mapping in DB vorhanden.\n\nSupervision-Anforderung: Prüft ob alle verwendeten step_types in DB existieren.\nTippfehler in step_type = sofortiger Ausfall.\nTask #516\n\"\"\"\n\nimport os\nimport sys\n\n# Add parent directory to path for imports\nsys.path.insert(0, str(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))\n\n\n# Step-Types die in der Pipeline verwendet werden\nREQUIRED_STEP_TYPES = [\n \"text_semantic_analyze\",\n \"knowledge_semantic_analyze\",\n \"entity_extract\",\n \"relation_extract\",\n \"ontology_classify\",\n \"taxonomy_build\",\n \"enrich\",\n \"embed\",\n \"vision\",\n]\n\n\ndef test_step_types_exist_in_db():\n \"\"\"Prüft ob alle verwendeten step_types in DB existieren.\"\"\"\n from pipeline_config import get_step_model, PipelineConfigError\n from db import db\n\n db.connect()\n\n missing = []\n found = []\n\n for step_type in REQUIRED_STEP_TYPES:\n try:\n model = get_step_model(step_type)\n found.append(f\"{step_type}: {model}\")\n except PipelineConfigError as e:\n missing.append(step_type)\n\n db.disconnect()\n\n if missing:\n raise AssertionError(\n f\"Step-Types nicht in DB konfiguriert: {missing}\\n\"\n f\"Gefunden: {found}\"\n )\n\n\ndef test_step_types_have_valid_models():\n \"\"\"Prüft ob alle step_types gültige (nicht-leere) Models haben.\"\"\"\n from pipeline_config import get_step_model\n from db import db\n\n db.connect()\n\n invalid = []\n\n for step_type in REQUIRED_STEP_TYPES:\n try:\n model = get_step_model(step_type)\n if not model or len(model.strip()) == 0:\n invalid.append(f\"{step_type}: empty model\")\n except Exception as e:\n invalid.append(f\"{step_type}: {e}\")\n\n db.disconnect()\n\n if invalid:\n raise AssertionError(f\"Ungültige Models: {invalid}\")\n\n\ndef test_no_duplicate_step_types():\n \"\"\"Prüft ob keine Duplikate in pipeline_steps existieren.\"\"\"\n from db import db\n\n db.connect()\n\n cursor = db.execute(\"\"\"\n SELECT pipeline_id, step_type, COUNT(*) as cnt\n FROM pipeline_steps\n GROUP BY pipeline_id, step_type\n HAVING COUNT(*) > 1\n \"\"\")\n duplicates = cursor.fetchall()\n cursor.close()\n\n db.disconnect()\n\n if duplicates:\n raise AssertionError(\n f\"Duplikate in pipeline_steps gefunden: {duplicates}\"\n )\n\n\nif __name__ == \"__main__\":\n print(\"Running step_type mapping tests...\")\n\n # Set DB password if not set\n if \"DB_PASSWORD\" not in os.environ:\n print(\"Warning: DB_PASSWORD not set, tests may fail\")\n\n try:\n test_step_types_exist_in_db()\n print(\"✓ test_step_types_exist_in_db passed\")\n except AssertionError as e:\n print(f\"✗ test_step_types_exist_in_db FAILED:\\n{e}\")\n\n try:\n test_step_types_have_valid_models()\n print(\"✓ test_step_types_have_valid_models passed\")\n except AssertionError as e:\n print(f\"✗ test_step_types_have_valid_models FAILED:\\n{e}\")\n\n try:\n test_no_duplicate_step_types()\n print(\"✓ test_no_duplicate_step_types passed\")\n except AssertionError as e:\n print(f\"✗ test_no_duplicate_step_types FAILED:\\n{e}\")\n",
"structuredPatch": [],
"originalFile": null
}
}