"""Pure-asyncpg reproducer of the 0.31.0 statement_cache_size=0 cursor regression.

The failing server-side cursor sequence is:

    prepared_stmt = await conn.prepare(operation, name=None)
    cursor        = await prepared_stmt.cursor(*parameters)

Key conditions to trigger the bug:
  - statement_cache_size=0        -> prepare(name=None) produces the UNNAMED "" statement
  - result contains an enum       -> asyncpg runs its own introspection as "" and clobbers it
  - a FRESH connection            -> the enum is unknown, so introspection fires here
  - >=1 bind params (2 here)      -> gives the clean ProtocolViolationError (2 vs 1)

The trigger is asyncpg's own type introspection running *inside* conn.prepare()
on a fresh connection (the enum result type is unknown), which clobbers the ""
statement; the later cursor bind then targets it. This is a direct connection,
no connection pooler involved.
"""
import asyncio
import asyncpg

DSN = dict(user='asyncpg', password='asyncpg_pw',
           database='asyncpg_db', host='127.0.0.1', port=5433)

OPERATION = 'SELECT id, value FROM mytable WHERE id > $1 AND id < $2 ORDER BY id'
PARAMETERS = (0, 100)


async def setup():
    # Setup on a SEPARATE connection so the streaming connection below is fresh
    # and has never introspected my_enum.
    conn = await asyncpg.connect(**DSN)
    await conn.execute('DROP TABLE IF EXISTS mytable')
    await conn.execute('DROP TYPE IF EXISTS my_enum')
    await conn.execute("CREATE TYPE my_enum AS ENUM ('value1', 'value2')")
    await conn.execute(
        'CREATE TABLE mytable(id serial PRIMARY KEY, value my_enum NOT NULL)')
    await conn.execute("INSERT INTO mytable(value) VALUES('value1'), ('value2')")
    await conn.close()


async def stream_over_cursor():
    # Fresh connection with the cache disabled -> unnamed "" statements.
    conn = await asyncpg.connect(**DSN, statement_cache_size=0)

    # A server-side cursor must run inside a transaction.
    tr = conn.transaction()
    await tr.start()
    try:
        # Step 1: prepare(name=None). With cache disabled this is the "" statement.
        # Because the result has an unknown enum type, asyncpg's own type
        # introspection runs here as "" too, clobbering this user statement.
        prepared_stmt = await conn.prepare(OPERATION, name=None)
        print('prepared name =', repr(prepared_stmt.get_name()))

        # Step 2: drive the server-side cursor with the bound params. The bind now
        # lands on the wrong "" statement (the 1-param introspection query).
        cursor = await prepared_stmt.cursor(*PARAMETERS)
        rows = await cursor.fetch(50)
        print('rows =', rows)
    finally:
        await tr.rollback()
        await conn.close()


async def main():
    await setup()
    await stream_over_cursor()


asyncio.run(main())