Function test_basic_message

scripts/test_soft_interrupt.py:144–188

Test that basic messaging works.

()

        print(f"  [{i}] {role}{suffix}: {text}...")

def test_basic_message():
    """Test that basic messaging works."""
    print("\n" + "="*60)
    print("TEST: Basic message (no interrupt)")
    print("="*60)
    
    session_id = create_test_session()
    if not session_id:
        print("❌ Failed to create session")
        return False
    print(f"Created session: {session_id}")
    
    try:
        result_q = queue_mod.Queue()
        thread = threading.Thread(
            target=send_message_async,
            args=("What is 2+2? Just answer with the number.", session_id, result_q)
        )
        thread.start()
        thread.join(timeout=120)
        
        if thread.is_alive():
            print("❌ Message timed out")
            return False
        
        ok, output = result_q.get()
        if not ok:
            print(f"❌ Message failed: {output[:200]}")
            return False
        
        print(f"Response: {output[:100]}...")
        
        history = get_history(session_id)
        roles = [m.get('role') for m in history]
        print(f"Roles: {roles}")
        
        if roles == ['user', 'assistant']:
            print("✅ Basic message works")
            return True
        else:
            print(f"❌ Unexpected roles: {roles}")
            return False
            
    finally:
        destroy_session(session_id)

def test_soft_interrupt_during_streaming():
    """