107 lines
3.3 KiB
Python
Executable File
107 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Validate that the callback retry fix is properly implemented in generated files.
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def check_file(path: Path) -> tuple[bool, list[str]]:
|
|
"""Check if a PLC logic file has the safe callback fix."""
|
|
if not path.exists():
|
|
return False, [f"File not found: {path}"]
|
|
|
|
content = path.read_text()
|
|
errors = []
|
|
|
|
# Check 1: Has import time
|
|
if "import time" not in content:
|
|
errors.append(f"{path.name}: Missing 'import time'")
|
|
|
|
# Check 2: Has _safe_callback function
|
|
if "def _safe_callback(" not in content:
|
|
errors.append(f"{path.name}: Missing '_safe_callback()' function")
|
|
|
|
# Check 3: Has retry logic in _safe_callback
|
|
if "for attempt in range(retries):" not in content:
|
|
errors.append(f"{path.name}: Missing retry loop in _safe_callback")
|
|
|
|
# Check 4: Has exception handling in _safe_callback
|
|
if "except Exception as e:" not in content:
|
|
errors.append(f"{path.name}: Missing exception handling in _safe_callback")
|
|
|
|
# Check 5: _write calls _safe_callback, not cb() directly
|
|
if "_safe_callback(cbs[key])" not in content:
|
|
errors.append(f"{path.name}: _write() not calling _safe_callback()")
|
|
|
|
# Check 6: _write does NOT call cbs[key]() directly (would crash)
|
|
lines = content.split("\n")
|
|
in_write = False
|
|
for i, line in enumerate(lines):
|
|
if "def _write(" in line:
|
|
in_write = True
|
|
elif in_write and line.strip().startswith("def "):
|
|
in_write = False
|
|
elif in_write and "cbs[key]()" in line and "_safe_callback" not in line:
|
|
errors.append(
|
|
f"{path.name}:{i+1}: _write() calls cbs[key]() directly (UNSAFE!)"
|
|
)
|
|
|
|
return len(errors) == 0, errors
|
|
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("Validating Callback Retry Fix")
|
|
print("=" * 60)
|
|
|
|
scenario_dir = Path("outputs/scenario_run")
|
|
logic_dir = scenario_dir / "logic"
|
|
|
|
if not logic_dir.exists():
|
|
print(f"\n❌ ERROR: Logic directory not found: {logic_dir}")
|
|
print(f"\nRun: .venv/bin/python3 build_scenario.py --overwrite")
|
|
return 1
|
|
|
|
plc_files = sorted(logic_dir.glob("plc*.py"))
|
|
|
|
if not plc_files:
|
|
print(f"\n❌ ERROR: No PLC logic files found in {logic_dir}")
|
|
return 1
|
|
|
|
print(f"\nChecking {len(plc_files)} PLC files...\n")
|
|
|
|
all_ok = True
|
|
for plc_file in plc_files:
|
|
ok, errors = check_file(plc_file)
|
|
|
|
if ok:
|
|
print(f"✅ {plc_file.name}: OK (retry fix present)")
|
|
else:
|
|
print(f"❌ {plc_file.name}: FAILED")
|
|
for error in errors:
|
|
print(f" - {error}")
|
|
all_ok = False
|
|
|
|
print("\n" + "=" * 60)
|
|
|
|
if all_ok:
|
|
print("✅ SUCCESS: All PLC files have the callback retry fix")
|
|
print("=" * 60)
|
|
print("\nYou can now:")
|
|
print(" 1. Run: ./test_simlab.sh")
|
|
print(" 2. Monitor PLC2 logs for crashes (should see none)")
|
|
return 0
|
|
else:
|
|
print("❌ FAILURE: Some files are missing the fix")
|
|
print("=" * 60)
|
|
print("\nTo fix:")
|
|
print(" 1. Run: .venv/bin/python3 build_scenario.py --overwrite")
|
|
print(" 2. Run: .venv/bin/python3 validate_fix.py")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|