From 6f99bdf013f3a0637f0544c4c3daeac0392553d6 Mon Sep 17 00:00:00 2001
From: Gord Thompson <gord@gordthompson.com>
Date: Thu, 21 Nov 2019 09:43:40 -0500
Subject: Add sequence support for MariaDB 10.3+.

Added support for use of the :class:`.Sequence` construct with MariaDB 10.3
and greater, as this is now supported by this database.  The construct
integrates with the :class:`.Table` object in the same way that it does for
other databases like PostrgreSQL and Oracle; if is present on the integer
primary key "autoincrement" column, it is used to generate defaults.   For
backwards compatibility, to support a :class:`.Table` that has a
:class:`.Sequence` on it to support sequence only databases like Oracle,
while still not having the sequence fire off for MariaDB, the optional=True
flag should be set, which indicates the sequence should only be used to
generate the primary key if the target database offers no other option.

Fixes: #4976
Closes: #4996
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/4996
Pull-request-sha: cb2e1426ea0b6bc6c93dbe8f033a11df9d8c4915

Change-Id: I507bc405eee6cae2c5991345d0eac53a37fe7512
---
 test/sql/test_defaults.py | 74 +++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 62 insertions(+), 12 deletions(-)

(limited to 'test/sql/test_defaults.py')

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index ed7af2572..f033abab2 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -1147,34 +1147,67 @@ class AutoIncrementTest(fixtures.TablesTest):
         with expect_warnings(".*has no Python-side or server-side default.*"):
             go()
 
-    def test_col_w_sequence_non_autoinc_no_firing(self):
-        metadata = self.metadata
+    @testing.metadata_fixture(ddl="function")
+    def dataset_no_autoinc(self, metadata):
         # plain autoincrement/PK table in the actual schema
         Table("x", metadata, Column("set_id", Integer, primary_key=True))
-        metadata.create_all()
 
         # for the INSERT use a table with a Sequence
         # and autoincrement=False.  Using a ForeignKey
         # would have the same effect
+
+        some_seq = Sequence("some_seq")
+
         dataset_no_autoinc = Table(
             "x",
             MetaData(),
             Column(
                 "set_id",
                 Integer,
-                Sequence("some_seq"),
+                some_seq,
                 primary_key=True,
                 autoincrement=False,
             ),
         )
+        return dataset_no_autoinc
 
-        testing.db.execute(dataset_no_autoinc.insert())
-        eq_(
-            testing.db.scalar(
-                select([func.count("*")]).select_from(dataset_no_autoinc)
-            ),
-            1,
-        )
+    def test_col_w_optional_sequence_non_autoinc_no_firing(
+        self, dataset_no_autoinc
+    ):
+        """this is testing that a Table which includes a Sequence, when
+        run against a DB that does not support sequences, the Sequence
+        does not get in the way.
+
+        """
+        dataset_no_autoinc.c.set_id.default.optional = True
+
+        with testing.db.connect() as conn:
+            conn.execute(dataset_no_autoinc.insert())
+            eq_(
+                conn.scalar(
+                    select([func.count("*")]).select_from(dataset_no_autoinc)
+                ),
+                1,
+            )
+
+    @testing.fails_if(testing.requires.sequences)
+    def test_col_w_nonoptional_sequence_non_autoinc_no_firing(
+        self, dataset_no_autoinc
+    ):
+        """When the sequence is not optional and sequences are supported,
+        the test fails because we didn't create the sequence.
+
+        """
+        dataset_no_autoinc.c.set_id.default.optional = False
+
+        with testing.db.connect() as conn:
+            conn.execute(dataset_no_autoinc.insert())
+            eq_(
+                conn.scalar(
+                    select([func.count("*")]).select_from(dataset_no_autoinc)
+                ),
+                1,
+            )
 
 
 class SequenceDDLTest(fixtures.TestBase, testing.AssertsCompiledSQL):
@@ -1336,8 +1369,24 @@ class SequenceExecTest(fixtures.TestBase):
         testing.db.execute(t1.insert().values(x=s.next_value()))
         self._assert_seq_result(testing.db.scalar(t1.select()))
 
+    @testing.requires.supports_lastrowid
+    @testing.provide_metadata
+    def test_inserted_pk_no_returning_w_lastrowid(self):
+        """test inserted_primary_key contains the pk when
+        pk_col=next_value(), lastrowid is supported."""
+
+        metadata = self.metadata
+        e = engines.testing_engine(options={"implicit_returning": False})
+        s = Sequence("my_sequence")
+        metadata.bind = e
+        t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+        t1.create()
+        r = e.execute(t1.insert().values(x=s.next_value()))
+        self._assert_seq_result(r.inserted_primary_key[0])
+
+    @testing.requires.no_lastrowid_support
     @testing.provide_metadata
-    def test_inserted_pk_no_returning(self):
+    def test_inserted_pk_no_returning_no_lastrowid(self):
         """test inserted_primary_key contains [None] when
         pk_col=next_value(), implicit returning is not used."""
 
@@ -1397,6 +1446,7 @@ class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         for s in (Sequence("my_seq"), Sequence("my_seq", optional=True)):
             assert str(s.next_value().compile(dialect=testing.db.dialect)) in (
                 "nextval('my_seq')",
+                "nextval(my_seq)",
                 "gen_id(my_seq, 1)",
                 "my_seq.nextval",
             )
-- 
cgit v1.2.1