Allow SchemaType and Variant to work together

Added support for the :class:`.Variant` and the :class:`.SchemaType`
objects to be compatible with each other.  That is, a variant
can be created against a type like :class:`.Enum`, and the instructions
to create constraints and/or database-specific type objects will
propagate correctly as per the variant's dialect mapping.

Also added testing for some potential double-event scenarios
on TypeDecorator but it seems usually this doesn't occur.

Change-Id: I4a7e7c26b4133cd14e870f5bc34a1b2f0f19a14a
Fixes: #2892

2 files changed, 97 insertions, 10 deletions

diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index bd67b6f69..6d0df3b5f 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -1607,7 +1607,7 @@ class SchemaTypeTest(fixtures.TestBase):
             impl = target_typ
 
         typ = MyType()
-        self._test_before_parent_attach(typ, target_typ)
+        self._test_before_parent_attach(typ, target_typ, double=True)
 
     def test_before_parent_attach_typedec_of_schematype(self):
         class MyType(TypeDecorator, sqltypes.SchemaType):
@@ -1623,17 +1623,52 @@ class SchemaTypeTest(fixtures.TestBase):
         typ = MyType()
         self._test_before_parent_attach(typ)
 
-    def _test_before_parent_attach(self, typ, evt_target=None):
+    def _test_before_parent_attach(self, typ, evt_target=None, double=False):
         canary = mock.Mock()
 
         if evt_target is None:
             evt_target = typ
 
-        event.listen(evt_target, "before_parent_attach", canary.go)
+        orig_set_parent = evt_target._set_parent
+        orig_set_parent_w_dispatch = evt_target._set_parent_with_dispatch
 
-        c = Column('q', typ)
+        def _set_parent(parent):
+            orig_set_parent(parent)
+            canary._set_parent(parent)
 
-        eq_(canary.mock_calls, [mock.call.go(evt_target, c)])
+        def _set_parent_w_dispatch(parent):
+            orig_set_parent_w_dispatch(parent)
+            canary._set_parent_with_dispatch(parent)
+
+        with mock.patch.object(evt_target, '_set_parent', _set_parent):
+            with mock.patch.object(
+                    evt_target, '_set_parent_with_dispatch',
+                    _set_parent_w_dispatch):
+                event.listen(evt_target, "before_parent_attach", canary.go)
+
+                c = Column('q', typ)
+
+        if double:
+            # no clean way yet to fix this, inner schema type is called
+            # twice, but this is a very unusual use case.
+            eq_(
+                canary.mock_calls,
+                [
+                    mock.call._set_parent(c),
+                    mock.call.go(evt_target, c),
+                    mock.call._set_parent(c),
+                    mock.call._set_parent_with_dispatch(c)
+                ]
+            )
+        else:
+            eq_(
+                canary.mock_calls,
+                [
+                    mock.call.go(evt_target, c),
+                    mock.call._set_parent(c),
+                    mock.call._set_parent_with_dispatch(c)
+                ]
+            )
 
     def test_independent_schema(self):
         m = MetaData()
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 8fbc65ef2..b417e6964 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -1315,11 +1315,7 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
                 non_native_enum_table.insert(), {"id": 1, "someenum": None})
             eq_(conn.scalar(select([non_native_enum_table.c.someenum])), None)
 
-    @testing.fails_on(
-        'mysql',
-        "The CHECK clause is parsed but ignored by all storage engines.")
-    @testing.fails_on(
-        'mssql', "FIXME: MS-SQL 2005 doesn't honor CHECK ?!?")
+    @testing.requires.enforces_check_constraints
     def test_check_constraint(self):
         assert_raises(
             (exc.IntegrityError, exc.ProgrammingError),
@@ -1327,6 +1323,62 @@ class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
             "insert into non_native_enum_table "
             "(id, someenum) values(1, 'four')")
 
+    @testing.requires.enforces_check_constraints
+    @testing.provide_metadata
+    def test_variant_we_are_default(self):
+        # test that the "variant" does not create a constraint
+        t = Table(
+            'my_table', self.metadata,
+            Column(
+                'data', Enum("one", "two", "three", name="e1").with_variant(
+                    Enum("four", "five", "six", name="e2"), "some_other_db"
+                )
+            )
+        )
+
+        eq_(
+            len([c for c in t.constraints if isinstance(c, CheckConstraint)]),
+            2
+        )
+
+        with testing.db.connect() as conn:
+            self.metadata.create_all(conn)
+            assert_raises(
+                (exc.IntegrityError, exc.ProgrammingError, exc.DataError),
+                conn.execute,
+                "insert into my_table "
+                "(data) values('four')")
+            conn.execute("insert into my_table (data) values ('two')")
+
+    @testing.requires.enforces_check_constraints
+    @testing.provide_metadata
+    def test_variant_we_are_not_default(self):
+        # test that the "variant" does not create a constraint
+        t = Table(
+            'my_table', self.metadata,
+            Column(
+                'data', Enum("one", "two", "three", name="e1").with_variant(
+                    Enum("four", "five", "six", name="e2"),
+                    testing.db.dialect.name
+                )
+            )
+        )
+
+        # ensure Variant isn't exploding the constraints
+        eq_(
+            len([c for c in t.constraints if isinstance(c, CheckConstraint)]),
+            2
+        )
+
+        with testing.db.connect() as conn:
+            self.metadata.create_all(conn)
+            assert_raises(
+                (exc.IntegrityError, exc.ProgrammingError, exc.DataError),
+                conn.execute,
+                "insert into my_table "
+                "(data) values('two')")
+            conn.execute("insert into my_table (data) values ('four')")
+
     def test_skip_check_constraint(self):
         with testing.db.connect() as conn:
             conn.execute(