merge into cymysql branch...

1 files changed, 34 insertions, 1 deletions

diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index b5f50aeea..293e629c8 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -190,10 +190,27 @@ class QueryTest(fixtures.TestBase):
                 try:
                     table.create(bind=engine, checkfirst=True)
                     i = insert_values(engine, table, values)
-                    assert i == assertvalues, "tablename: %s %r %r" % (table.name, repr(i), repr(assertvalues))
+                    assert i == assertvalues, "tablename: %s %r %r" % \
+                            (table.name, repr(i), repr(assertvalues))
                 finally:
                     table.drop(bind=engine)
 
+    @testing.only_on('sqlite+pysqlite')
+    @testing.provide_metadata
+    def test_lastrowid_zero(self):
+        from sqlalchemy.dialects import sqlite
+        eng = engines.testing_engine()
+        class ExcCtx(sqlite.base.SQLiteExecutionContext):
+            def get_lastrowid(self):
+                return 0
+        eng.dialect.execution_ctx_cls = ExcCtx
+        t = Table('t', MetaData(), Column('x', Integer, primary_key=True),
+                            Column('y', Integer))
+        t.create(eng)
+        r = eng.execute(t.insert().values(y=5))
+        eq_(r.inserted_primary_key, [0])
+
+
     @testing.fails_on('sqlite', "sqlite autoincremnt doesn't work with composite pks")
     def test_misordered_lastrow(self):
         related = Table('related', metadata,
@@ -1011,6 +1028,22 @@ class QueryTest(fixtures.TestBase):
             lambda: row[u2.c.user_id]
         )
 
+    def test_ambiguous_column_contains(self):
+        # ticket 2702.  in 0.7 we'd get True, False.
+        # in 0.8, both columns are present so it's True;
+        # but when they're fetched you'll get the ambiguous error.
+        users.insert().execute(user_id=1, user_name='john')
+        result = select([
+                    users.c.user_id,
+                    addresses.c.user_id]).\
+                    select_from(users.outerjoin(addresses)).execute()
+        row = result.first()
+
+        eq_(
+            set([users.c.user_id in row, addresses.c.user_id in row]),
+            set([True])
+        )
+
     def test_ambiguous_column_by_col_plus_label(self):
         users.insert().execute(user_id=1, user_name='john')
         result = select([users.c.user_id,