Function main

tests/python/test_row_description_debug.py:40–205

()

    sock.send(msg)

def main():
    # Start pgsqlite
    db_file = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
    db_file.close()
    db_path = db_file.name
    
    port = 15445
    print(f"Starting pgsqlite on port {port}")
    env = os.environ.copy()
    env['RUST_LOG'] = 'info'
    pgsqlite_proc = subprocess.Popen([
        '../../target/release/pgsqlite',
        '--database', db_path,
        '--port', str(port)
    ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env)
    
    time.sleep(1)
    
    try:
        # Connect using raw socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect(('localhost', port))
        
        # Send startup message
        startup = struct.pack('!HH', 3, 0)  # Protocol version 3.0
        startup += b'user\x00postgres\x00'
        startup += b'database\x00main\x00\x00'
        send_message(sock, None, startup)
        
        # Read authentication response
        while True:
            msg_type, body = read_message(sock)
            print(f"Received: {msg_type}")
            if msg_type == 'Z':  # ReadyForQuery
                break
        
        # Create tables
        query = b"CREATE TABLE users (id SERIAL PRIMARY KEY, username VARCHAR(50));\x00"
        send_message(sock, 'Q', query)
        while True:
            msg_type, body = read_message(sock)
            if msg_type == 'Z':
                break
        
        query = b"CREATE TABLE orders (id SERIAL PRIMARY KEY, customer_id INTEGER, total_amount NUMERIC(12,2));\x00"
        send_message(sock, 'Q', query)
        while True:
            msg_type, body = read_message(sock)
            if msg_type == 'Z':
                break
        
        print("\n=== Testing Extended Protocol ===")
        
        # Parse the exact query that SQLAlchemy uses
        stmt_name = b"stmt1\x00"
        query = b"SELECT orders.id AS orders_id, orders.customer_id AS orders_customer_id, orders.total_amount AS orders_total_amount FROM orders WHERE $1::INTEGER = orders.customer_id\x00"
        param_types = struct.pack('!H', 1) + struct.pack('!I', 23)  # 1 param of type INT4