Add async engine regression coverage for duplicate columns

tests/integration/test_restapi_integration.py

@@ -2520,6 +2520,30 @@ def test_join_via_sqlalchemy(self, d1_engine, joined_tables):
 
         assert rows == [(1, "x", 7)]
 
+    @pytest.mark.asyncio
+    async def test_join_via_async_sqlalchemy(self, joined_tables):
+        """Keep duplicate columns through the real async engine and adapter."""
+        from sqlalchemy import text
+        from sqlalchemy.ext.asyncio import create_async_engine
+
+        table_a, table_b = joined_tables
+        url = f"cloudflare_d1+async://{ACCOUNT_ID}:{API_TOKEN}@{DATABASE_ID}"
+        engine = create_async_engine(url)
+
+        try:
+            async with engine.connect() as conn:
+                result = await conn.execute(
+                    text(
+                        f"/* note */ SELECT {table_a}.id, {table_a}.name, "
+                        f"{table_b}.id FROM {table_a} JOIN {table_b} "
+                        f"ON {table_b}.a_id = {table_a}.id"
+                    )
+                )
+                assert list(result.keys()) == ["id", "name", "id"]
+                assert result.fetchall() == [(1, "x", 7)]
+        finally:
+            await engine.dispose()
+
 
 # MARK: - Autoincrement Insert Tests (Issue #12)
 

tests/unit/test_connection.py

@@ -10,6 +10,7 @@
 import httpx
 import pytest
 from sqlalchemy import create_engine, event, text
+from sqlalchemy.ext.asyncio import create_async_engine
 
 from sqlalchemy_cloudflare_d1.connection import AsyncConnection, Connection
 
@@ -92,5 +93,27 @@ def use_fake_raw(dbapi_connection, connection_record):
     assert rows == [(1, "x", 7)]
 
 
+def test_async_engine_keeps_values_of_columns_with_same_name():
+    """Keep positional values through the async SQLAlchemy adapter, too."""
+    engine = create_async_engine("cloudflare_d1+async://acct:token@db")
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def use_fake_raw(dbapi_connection, connection_record):
+        dbapi_connection._connection.client = httpx.AsyncClient(
+            transport=raw_transport(["id", "name", "id"], [[1, "x", 7]])
+        )
+
+    async def main():
+        try:
+            async with engine.connect() as conn:
+                result = await conn.execute(text("/* note */ " + JOIN_SQL))
+                assert list(result.keys()) == ["id", "name", "id"]
+                assert result.fetchall() == [(1, "x", 7)]
+        finally:
+            await engine.dispose()
+
+    asyncio.run(main())
+
+
 if __name__ == "__main__":
     pytest.main([__file__])