2125_GCE/integration_tests/test_determinism.py

#!/usr/bin/env python3
import sys
from pathlib import Path
import json

sys.path.insert(0, str(Path.cwd()))

from gx_cli.main import main
from runtime_executor.gx_loader import load_gx


def test_deterministic_structure():
    """Test that compiled .gx files have the same structure"""
    outputs = [Path(f"/tmp/test_struct_{i}.gx") for i in range(3)]

    for output in outputs:
        if output.exists():
            output.unlink()
        main(["compile", "/home/dave/sample_code.py", "-o", str(output)])

    # Load all files and check structure
    results = []
    for output in outputs:
        manifest, payload = load_gx(str(output))
        results.append({
            "manifest_keys": sorted(manifest.keys()),
            "payload_size": len(payload),
            "file_size": output.stat().st_size,
            "segments": len(manifest.get("codex_lineage", {}).get("segments", []))
        })

    # Check structure consistency (ignore timestamps)
    if not all(r["manifest_keys"] == results[0]["manifest_keys"] for r in results):
        print("FAIL: manifest keys differ")
        return False

    if not all(r["payload_size"] == results[0]["payload_size"] for r in results):
        print("FAIL: payload sizes differ")
        return False

    if not all(r["segments"] == results[0]["segments"] for r in results):
        print("FAIL: segment counts differ")
        return False

    print(f"PASS: .gx structure is deterministic (payload={results[0]['payload_size']} bytes, {results[0]['segments']} segments)")
    return True


def test_deterministic_magic_and_size():
    """Test that magic bytes and file sizes are deterministic"""
    outputs = [Path(f"/tmp/test_magic_{i}.gx") for i in range(3)]

    sizes = []
    for output in outputs:
        if output.exists():
            output.unlink()
        main(["compile", "/home/dave/sample_code.py", "-o", str(output)])
        sizes.append(output.stat().st_size)

    # All sizes should match
    if not all(s == sizes[0] for s in sizes):
        print(f"FAIL: sizes vary: {sizes}")
        return False

    # Check magic bytes
    for output in outputs:
        data = output.read_bytes()
        if not data.startswith(b'XIC'):
            print("FAIL: invalid magic bytes")
            return False

    print(f"PASS: Magic bytes and file sizes are deterministic ({sizes[0]} bytes)")
    return True


def test_payload_determinism():
    """Test that compressed payloads are deterministic"""
    outputs = [Path(f"/tmp/test_payload_{i}.gx") for i in range(2)]

    for output in outputs:
        if output.exists():
            output.unlink()
        main(["compile", "/home/dave/sample_code.py", "-o", str(output)])

    # Extract payloads
    _, payload1 = load_gx(str(outputs[0]))
    _, payload2 = load_gx(str(outputs[1]))

    if payload1 != payload2:
        print("FAIL: compressed payloads differ")
        return False

    print(f"PASS: Compressed payload is deterministic ({len(payload1)} bytes)")
    return True


def main_test():
    print("[TEST SUITE] test_determinism.py")
    print()

    tests = [
        ("Deterministic structure", test_deterministic_structure),
        ("Deterministic magic/size", test_deterministic_magic_and_size),
        ("Payload determinism", test_payload_determinism),
    ]

    passed = 0
    failed = 0

    for name, test_func in tests:
        print(f"Running: {name}...", end=" ")
        try:
            if test_func():
                passed += 1
            else:
                failed += 1
        except Exception as e:
            print(f"FAIL: {e}")
            failed += 1

    print()
    print(f"Results: {passed} passed, {failed} failed")
    print()
    print("Note: Full byte-for-byte determinism is not expected due to")
    print("      timestamp fields in the manifest. This test verifies")
    print("      that structure, size, magic bytes, and payload are")
    print("      deterministic, which is sufficient for the pipeline.")

    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main_test())