Reparse unnamed statements before cursor binding (#1364)

With `statement_cache_size=0`, type introspection can replace the unnamed
statement created by `prepare()`. The statement is marked unprepared, but
awaiting `PreparedStatement.cursor()` binds it without reparsing, so the
server receives arguments for the introspection query instead.

Make `bind()` honor the unprepared flag, as `bind_execute()` already does
to fix this.

Fixes #1335.
Closes #1345.

Author: elprans

asyncpg/protocol/coreproto.pyx

@@ -313,6 +313,10 @@ cdef class CoreProtocol:
             # ErrorResponse
             self._parse_msg_error_response(True)
 
+        elif mtype == b'1':
+            # ParseComplete, in case `_bind()` is reparsing
+            self.buffer.discard_message()
+
         elif mtype == b'2':
             # BindComplete
             self.buffer.discard_message()

asyncpg/protocol/protocol.pyx

@@ -280,6 +280,9 @@ cdef class BaseProtocol(CoreProtocol):
 
         waiter = self._new_waiter(timeout)
         try:
+            if not state.prepared:
+                self._send_parse_message(state.name, state.query)
+
             self._bind(
                 portal_name,
                 state.name,

tests/test_cursor.py

@@ -159,3 +159,22 @@ async def test_cursor_04(self):
             st = await self.con.cursor('SELECT generate_series(0, 100)')
             await st.forward(42)
             self.assertEqual(await st.fetchrow(), (42,))
+
+    @tb.with_connection_options(statement_cache_size=0)
+    async def test_cursor_05_unnamed_statement_reparsed(self):
+        await self.con.execute(
+            "CREATE TYPE cursor_05_t AS ENUM ('foo', 'bar')"
+        )
+        try:
+            async with self.con.transaction():
+                # Enum introspection replaces the unnamed statement on the
+                # server, so opening the cursor must re-parse it first.
+                st = await self.con.prepare('''
+                    SELECT $1::int, $2::int, 'foo'::cursor_05_t
+                ''')
+                self.assertEqual(st.get_name(), '')
+
+                cur = await st.cursor(1, 2)
+                self.assertEqual(await cur.fetch(1), [(1, 2, 'foo')])
+        finally:
+            await self.con.execute('DROP TYPE cursor_05_t')