1jehuang--jcode/tests/test_selfdev_reload.py

#!/usr/bin/env python3
"""
Test script to verify selfdev reload works correctly.

This tests that:
1. The selfdev reload tool returns appropriate output
2. The reload context is saved
3. After restart, the continuation message is sent to the model

Run with: python tests/test_selfdev_reload.py
"""

import socket
import json
import time
import os
import sys
import glob

RUNTIME_DIR = os.environ.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}"
SOCKET_PATH = os.path.join(RUNTIME_DIR, "jcode-debug.sock")
JCODE_DIR = os.path.expanduser("~/.jcode")

def send_cmd(sock, cmd, session_id=None, timeout=60):
    """Send a debug command and get the response."""
    req = {"type": "debug_command", "id": 1, "command": cmd}
    if session_id:
        req["session_id"] = session_id
    sock.send((json.dumps(req) + '\n').encode())
    sock.settimeout(timeout)
    data = b""
    while True:
        try:
            chunk = sock.recv(65536)
            if not chunk:
                break
            data += chunk
            try:
                return json.loads(data.decode())
            except json.JSONDecodeError:
                continue
        except socket.timeout:
            break
    return json.loads(data.decode()) if data else None

def test_selfdev_status():
    """Test that selfdev status works."""
    print("=" * 60)
    print("Test: selfdev status")
    print("=" * 60)

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        sock.connect(SOCKET_PATH)
    except FileNotFoundError:
        print(f"ERROR: Debug socket not found at {SOCKET_PATH}")
        return False

    try:
        # Create a test session
        result = send_cmd(sock, "create_session:selfdev:/home/jeremy/jcode")
        if not result or not result.get('ok'):
            print(f"Failed to create session: {result}")
            return False
        session_id = json.loads(result['output'])['session_id']
        print(f"   Session ID: {session_id}")

        # Check state to verify selfdev is available
        result = send_cmd(sock, "state", session_id)
        if result and result.get('ok'):
            state = json.loads(result['output'])
            print(f"   Is canary: {state.get('is_canary', False)}")

        # Call selfdev status
        print("\n1. Calling selfdev status...")
        result = send_cmd(sock, 'tool:selfdev {"action":"status"}', session_id, timeout=30)

        if not result:
            print("   No response")
            return False

        if result.get('ok'):
            output = result.get('output', '')
            print(f"   Status output (preview):")
            for line in output.split('\n')[:10]:
                print(f"     {line}")
            if len(output.split('\n')) > 10:
                print(f"     ... ({len(output.split(chr(10)))} lines total)")
        else:
            error = result.get('error', 'Unknown error')
            if 'selfdev' in error.lower() and 'not available' in error.lower():
                print(f"   SKIP: selfdev not available (not in self-dev mode)")
                send_cmd(sock, f"destroy_session:{session_id}")
                return True  # Skip is not a failure
            print(f"   Error: {error}")
            send_cmd(sock, f"destroy_session:{session_id}")
            return False

        # Cleanup
        send_cmd(sock, f"destroy_session:{session_id}")

        print("\n" + "=" * 60)
        print("TEST PASSED: selfdev status works")
        print("=" * 60)
        return True

    finally:
        sock.close()

def test_selfdev_socket_info():
    """Test that selfdev socket-info works."""
    print("\n" + "=" * 60)
    print("Test: selfdev socket-info")
    print("=" * 60)

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        sock.connect(SOCKET_PATH)
    except FileNotFoundError:
        print(f"ERROR: Debug socket not found")
        return False

    try:
        result = send_cmd(sock, "create_session:selfdev:/home/jeremy/jcode")
        if not result or not result.get('ok'):
            return False
        session_id = json.loads(result['output'])['session_id']

        # Call selfdev socket-info
        print("1. Calling selfdev socket-info...")
        result = send_cmd(sock, 'tool:selfdev {"action":"socket-info"}', session_id, timeout=30)

        if not result:
            print("   No response")
            return False

        if result.get('ok'):
            output = result.get('output', '')
            print(f"   Output (preview):")
            for line in output.split('\n')[:5]:
                print(f"     {line}")

            # Verify it contains expected info
            if 'debug_socket' in output.lower() or 'socket' in output.lower():
                print("   ✓ Contains socket info")
            else:
                print("   Warning: May not contain expected socket info")
        else:
            error = result.get('error', '')
            if 'not available' in error.lower():
                print("   SKIP: selfdev not available")
                send_cmd(sock, f"destroy_session:{session_id}")
                return True
            print(f"   Error: {error}")
            return False

        send_cmd(sock, f"destroy_session:{session_id}")

        print("\n" + "=" * 60)
        print("TEST PASSED: selfdev socket-info works")
        print("=" * 60)
        return True

    finally:
        sock.close()

def test_reload_context():
    """Test that reload context file exists and is valid JSON."""
    print("\n" + "=" * 60)
    print("Test: Reload context file format")
    print("=" * 60)

    context_candidates = sorted(
        glob.glob(os.path.join(JCODE_DIR, "reload-context-*.json"))
    )
    legacy_context_path = os.path.join(JCODE_DIR, "reload-context.json")
    if os.path.exists(legacy_context_path):
        context_candidates.append(legacy_context_path)

    # Check if there's an existing context file
    if context_candidates:
        context_path = context_candidates[0]
        print(f"1. Found existing reload context at {context_path}")
        try:
            with open(context_path) as f:
                ctx = json.load(f)
            print(f"   Fields: {list(ctx.keys())}")

            # Verify expected fields
            expected = ['task_context', 'version_before', 'version_after', 'session_id', 'timestamp']
            missing = [f for f in expected if f not in ctx]
            if missing:
                print(f"   Warning: Missing fields: {missing}")
            else:
                print("   ✓ All expected fields present")

            print("\n" + "=" * 60)
            print("TEST PASSED: Reload context format is valid")
            print("=" * 60)
            return True

        except json.JSONDecodeError as e:
            print(f"   ERROR: Invalid JSON: {e}")
            return False
    else:
        print(f"1. No existing reload context (will be created on selfdev reload)")
        print("   This is expected if no reload has been performed recently.")

        print("\n" + "=" * 60)
        print("TEST SKIPPED: No reload context to check")
        print("=" * 60)
        return True

def main():
    print("Selfdev Reload Tests")
    print("=" * 60)
    print()

    all_passed = True

    # Test 1: selfdev status
    if not test_selfdev_status():
        all_passed = False

    # Test 2: selfdev socket-info
    if not test_selfdev_socket_info():
        all_passed = False

    # Test 3: Reload context format
    if not test_reload_context():
        all_passed = False

    print()
    if all_passed:
        print("✓ All tests passed!")
        return 0
    else:
        print("✗ Some tests failed")
        return 1

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