Software Orchestrator
5769c1cd22
- Auto-Registration-Bug behoben (register_project/get_project_id/resolve_project Trennung) - 25 Tests gruen (Pytest) - block_compactor-Tool refactored (Option B: Soft-Check statt Hard-Block) - 4 Restbaustellen gefixt - DB-Schema: plugin_settings-Tabelle hinzugefuegt - 3 Schattenprojekte aus DB geloescht - Plan v3 + Refactor-Plan + Worklog dokumentiert
115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""AST-based smoke test for all plugin tools.
|
|
|
|
Why AST and not import? The /a0/ framework root is a namespace package
|
|
(no __init__.py) and cannot be reliably imported from a standalone venv.
|
|
AST parsing avoids the venv/sys.path issue while still verifying:
|
|
- file parses cleanly (no syntax errors)
|
|
- module defines exactly one Tool subclass
|
|
- class name matches expected convention (CamelCase of module name)
|
|
- class inherits from framework Tool base
|
|
- class defines `async def execute(self, **kwargs) -> Response`
|
|
- the execute() signature is `async def execute(self, ...)`
|
|
|
|
Run from the plugin root:
|
|
python3 scripts/test_all_tools.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import ast
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PLUGIN_ROOT = Path(__file__).resolve().parent.parent
|
|
TOOLS_DIR = PLUGIN_ROOT / "tools"
|
|
|
|
EXPECTED_TOOLS = [
|
|
"artifact_guard", "block_compactor", "block_resume",
|
|
"capability_check", "context_compactor", "handover_to_specialist",
|
|
"next_step", "orchestrator_state", "plan_mode_guard",
|
|
"project_registry", "quality_gate", "read_briefing", "repo_manifest",
|
|
"resume_checker", "scorecard_update", "tool_registry", "orchestrator_self_check",
|
|
]
|
|
|
|
def to_camel(name: str) -> str:
|
|
return "".join(part.capitalize() for part in name.split("_"))
|
|
|
|
def check_tool_file(py_path: Path) -> dict:
|
|
result = {
|
|
"file": py_path.name,
|
|
"module": py_path.stem,
|
|
"parses": False,
|
|
"class_found": None,
|
|
"inherits_tool": False,
|
|
"has_execute": False,
|
|
"execute_async": False,
|
|
"errors": [],
|
|
}
|
|
src = py_path.read_text(encoding="utf-8")
|
|
try:
|
|
tree = ast.parse(src, filename=str(py_path))
|
|
result["parses"] = True
|
|
except SyntaxError as e:
|
|
result["errors"].append(f"SyntaxError: {e}")
|
|
return result
|
|
|
|
expected_class = to_camel(py_path.stem)
|
|
found_class = None
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ClassDef) and node.name == expected_class:
|
|
found_class = node
|
|
break
|
|
if found_class is None:
|
|
result["errors"].append(f"class {expected_class!r} not found")
|
|
return result
|
|
result["class_found"] = found_class.name
|
|
|
|
for base in found_class.bases:
|
|
is_tool = (isinstance(base, ast.Name) and base.id == "Tool") or \
|
|
(isinstance(base, ast.Attribute) and base.attr == "Tool")
|
|
if is_tool:
|
|
result["inherits_tool"] = True
|
|
break
|
|
if not result["inherits_tool"]:
|
|
result["errors"].append(f"{found_class.name} does not inherit from Tool")
|
|
|
|
for item in found_class.body:
|
|
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)) and item.name == "execute":
|
|
result["has_execute"] = True
|
|
if isinstance(item, ast.AsyncFunctionDef):
|
|
result["execute_async"] = True
|
|
else:
|
|
result["errors"].append("execute() is not async")
|
|
break
|
|
if not result["has_execute"]:
|
|
result["errors"].append("no execute() method found")
|
|
|
|
return result
|
|
|
|
def main():
|
|
print(f"AST smoke test for {len(EXPECTED_TOOLS)} tools in {TOOLS_DIR}")
|
|
print("=" * 78)
|
|
passed = 0
|
|
failed = 0
|
|
for tool_name in EXPECTED_TOOLS:
|
|
py_path = TOOLS_DIR / f"{tool_name}.py"
|
|
if not py_path.exists():
|
|
print(f" [FAIL] {tool_name:30s} -> file not found: {py_path}")
|
|
failed += 1
|
|
continue
|
|
r = check_tool_file(py_path)
|
|
if not r["errors"]:
|
|
print(f" [PASS] {tool_name:30s} -> class {r['class_found']!r} "
|
|
f"(inherits Tool, async execute())")
|
|
passed += 1
|
|
else:
|
|
print(f" [FAIL] {tool_name:30s} -> {'; '.join(r['errors'])}")
|
|
failed += 1
|
|
print("=" * 78)
|
|
print(f"Result: {passed}/{len(EXPECTED_TOOLS)} passed, {failed} failed")
|
|
sys.exit(0 if failed == 0 else 1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|