validate.py

import subprocess, sys, os, re

dst = r'C:\Projects\RTOS-Graph-RAG'
results = {}

py314 = r'C:\Users\satya\AppData\Local\Programs\Python\Python314\python.exe'

# --- Validation 1: Python imports ---
print("=" * 60)
print("VALIDATION 1: Python imports")
print("=" * 60)
imports = [
    "graph.rtos_graph.cli",
    "graph.rtos_graph.config",
    "graph.rtos_graph.models",
    "graph.rtos_graph.inventory",
    "graph.rtos_graph.extractor",
    "graph.rtos_graph.behavior",
    "graph.rtos_graph.integration",
    "graph.rtos_graph.query_engine",
    "graph.rtos_graph.risk_analysis",
    "graph.rtos_graph.metrics",
    "graph.rtos_graph.validation",
    "core.base_node",
    "core.base_edge",
    "core.domain_plugin",
    "core.registry",
    "plugins.firmware.plugin",
    "plugins.firmware.nodes",
    "plugins.firmware.edges",
    "plugins.firmware.extractors.buffer_extractor",
]
ok = 0
fail = 0
for mod in imports:
    r = subprocess.run([py314, "-c", "import " + mod], capture_output=True, text=True, cwd=dst)
    if r.returncode == 0:
        print("  PASS: " + mod)
        ok += 1
    else:
        err = r.stderr.strip()[-100:]
        print("  FAIL: " + mod + " -> " + err)
        fail += 1
results["imports"] = (ok, fail)
print("Result: %d passed, %d failed\n" % (ok, fail))

# --- Validation 2: PluginRegistry loads ---
print("=" * 60)
print("VALIDATION 2: PluginRegistry loads")
print("=" * 60)
code = (
    "from plugins.firmware import plugin; from core.registry import registry; "
    "plugins = registry.all_plugins(); "
    "print('Loaded %d plugins: %s' % (len(plugins), [p.name for p in plugins]))"
)
r = subprocess.run([py314, "-c", code], capture_output=True, text=True, cwd=dst)
if r.returncode == 0 and "firmware" in r.stdout:
    print("  PASS: " + r.stdout.strip())
    results["plugin_registry"] = True
else:
    print("  FAIL: " + r.stderr.strip())
    results["plugin_registry"] = False
print()

# --- Validation 3: RTOSExtractor import ---
print("=" * 60)
print("VALIDATION 3: RTOSExtractor (run_extraction)")
print("=" * 60)
r = subprocess.run([py314, "-c", "from graph.rtos_graph.extractor import run_extraction; print('OK')"],
    capture_output=True, text=True, cwd=dst)
if r.returncode == 0:
    print("  PASS: " + r.stdout.strip())
    results["extractor"] = True
else:
    print("  FAIL: " + r.stderr.strip())
    results["extractor"] = False
print()

# --- Validation 4: RTOSQueryEngine import ---
print("=" * 60)
print("VALIDATION 4: RTOSQueryEngine")
print("=" * 60)
r = subprocess.run([py314, "-c", "from graph.rtos_graph.query_engine import RTOSQueryEngine; print('OK')"],
    capture_output=True, text=True, cwd=dst)
if r.returncode == 0:
    print("  PASS: " + r.stdout.strip())
    results["query_engine"] = True
else:
    print("  FAIL: " + r.stderr.strip())
    results["query_engine"] = False
print()

# --- Validation 5: MCP server import ---
print("=" * 60)
print("VALIDATION 5: MCP server")
print("=" * 60)
code = (
    "from mcp.mcp_server import mcp; print('MCP server: ' + mcp.name)"
)
r = subprocess.run([py314, "-c", code], capture_output=True, text=True, cwd=dst)
if r.returncode == 0:
    print("  PASS: " + r.stdout.strip())
    results["mcp"] = True
else:
    print("  FAIL: " + r.stderr.strip()[-200:])
    results["mcp"] = False
print()

# --- Validation 6: Providers load ---
print("=" * 60)
print("VALIDATION 6: Providers")
print("=" * 60)
code = (
    "from providers import get_provider, OpenRouterProvider, GeminiProvider; "
    "print('Providers imported')"
)
r = subprocess.run([py314, "-c", code], capture_output=True, text=True, cwd=dst)
if r.returncode == 0:
    print("  PASS: " + r.stdout.strip())
    results["providers"] = True
else:
    print("  FAIL: " + r.stderr.strip()[-200:])
    results["providers"] = False
print()

# --- Validation 7: Absolute path scan ---
print("=" * 60)
print("VALIDATION 7: Absolute path scan")
print("=" * 60)
pattern = re.compile("C:\\\\" + "Projects\\\\" + "1st_Attempt" + "|" + "C:/Projects/" + "1st_Attempt")
hits = []
for root, dirs, files in os.walk(dst):
    dirs[:] = [d for d in dirs if d not in ('__pycache__', '.git', 'graph_snapshots', 'legacy')]
    for f in files:
        if f.endswith(('.py', '.json', '.md', '.txt', '.ps1')):
            path = os.path.join(root, f)
            try:
                with open(path, 'r', encoding='utf-8', errors='ignore') as fh:
                    for i, line in enumerate(fh, 1):
                        if pattern.search(line):
                            hits.append("%s:%d" % (path, i))
            except:
                pass
if not hits:
    print("  PASS: No hardcoded source paths found")
    results["path_scan"] = True
else:
    print("  FAIL: %d hardcoded paths found:" % len(hits))
    for h in hits[:10]:
        print("    " + h)
    results["path_scan"] = False
print()

# --- Validation 8: Golden fixtures ---
print("=" * 60)
print("VALIDATION 8: Golden fixtures")
print("=" * 60)
golden_dir = os.path.join(dst, 'tests', 'golden')
golden_files = [f for f in os.listdir(golden_dir) if not f.endswith('.bak')]
print("  Golden files: %d" % len(golden_files))
for f in sorted(golden_files):
    size = os.path.getsize(os.path.join(golden_dir, f))
    print("    %s (%d bytes)" % (f, size))
results["golden"] = len(golden_files) > 0
print()

# --- Validation 9: requirements-lock.txt ---
print("=" * 60)
print("VALIDATION 9: requirements-lock.txt")
print("=" * 60)
lock_path = os.path.join(dst, 'requirements-lock.txt')
if os.path.exists(lock_path):
    with open(lock_path) as f:
        lines = [l for l in f.readlines() if l.strip() and not l.startswith('#')]
    print("  PASS: %d packages locked" % len(lines))
    results["lockfile"] = True
else:
    print("  FAIL: requirements-lock.txt not found")
    results["lockfile"] = False
print()

# --- Summary ---
print("=" * 60)
print("VALIDATION SUMMARY")
print("=" * 60)
all_pass = True
for k, v in results.items():
    if isinstance(v, tuple):
        status = "PASS" if v[1] == 0 else "FAIL"
        if v[1] > 0:
            all_pass = False
        print("  [%s] %s (%d/%d)" % (status, k, v[0], v[0]+v[1]))
    else:
        status = "PASS" if v else "FAIL"
        if not v:
            all_pass = False
        print("  [%s] %s" % (status, k))

print("\nOverall: %s" % ("ALL PASSED" if all_pass else "SOME FAILED"))

# --- Now run pytest with Python314 ---
print("\n" + "=" * 60)
print("PYTEST SUITE (Python314)")
print("=" * 60)
r = subprocess.run([py314, "-m", "pytest", "tests/", "-v", "--tb=short"],
    capture_output=True, text=True, cwd=dst, timeout=180)
# Print last 4000 chars of stdout
out = r.stdout
if len(out) > 4000:
    print("... (truncated) ...")
    print(out[-4000:])
else:
    print(out)
if r.stderr:
    err = r.stderr
    if len(err) > 1000:
        print("STDERR:", err[-1000:])
    else:
        print("STDERR:", err)