Introduce flag combinations fixture

A helper for @testing.combinations when we just have lots of
true/false combinations as is the case with some ORM tests.

Change-Id: I9f2de97ce5b2487411ed610b8d41169c1052bd8f
(cherry picked from commit 431b019c435654162b0d61b8131a170c5b858fb0)

diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 5e303be00a31f09d3a628926fbf81b3fb16d2580..ed99b1eb2bb9d6a129dd8b6c79130c6e799c8066 100644 (file)
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -51,6 +51,7 @@ from .exclusions import skip  # noqa
 from .exclusions import skip_if  # noqa
 from .util import adict  # noqa
 from .util import fail  # noqa
+from .util import flag_combinations  # noqa
 from .util import force_drop_names  # noqa
 from .util import metadata_fixture  # noqa
 from .util import provide_metadata  # noqa

diff --git a/lib/sqlalchemy/testing/util.py b/lib/sqlalchemy/testing/util.py
index 64738ad2f64b2046f8165dead8add95ecb3bd7ce..87c461fd2b31ef2574f900786d57b5234f83a100 100644 (file)
--- a/lib/sqlalchemy/testing/util.py
+++ b/lib/sqlalchemy/testing/util.py
@@ -211,6 +211,56 @@ def provide_metadata(fn, *args, **kw):
         self.metadata = prev_meta
 
 
+def flag_combinations(*combinations):
+    """A facade around @testing.combinations() oriented towards boolean
+    keyword-based arguments.
+
+    Basically generates a nice looking identifier based on the keywords
+    and also sets up the argument names.
+
+    E.g.::
+
+        @testing.flag_combinations(
+            dict(lazy=False, passive=False),
+            dict(lazy=True, passive=False),
+            dict(lazy=False, passive=True),
+            dict(lazy=False, passive=True, raiseload=True),
+        )
+
+
+    would result in::
+
+        @testing.combinations(
+            ('', False, False, False),
+            ('lazy', True, False, False),
+            ('lazy_passive', True, True, False),
+            ('lazy_passive', True, True, True),
+            id_='iaaa',
+            argnames='lazy,passive,raiseload'
+        )
+
+    """
+
+    from . import config
+
+    keys = set()
+
+    for d in combinations:
+        keys.update(d)
+
+    keys = sorted(keys)
+
+    return config.combinations(
+        *[
+            ("_".join(k for k in keys if d.get(k, False)),)
+            + tuple(d.get(k, False) for k in keys)
+            for d in combinations
+        ],
+        id_="i" + ("a" * len(keys)),
+        argnames=",".join(keys)
+    )
+
+
 def metadata_fixture(ddl="function"):
     """Provide MetaData for a pytest fixture."""
 

diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 2d686c7e90765df9077cd7ff8ac0a9105b00e9ae..60cfde3c3dfb06e22fe9ad17718e65023541cffd 100644 (file)
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -5083,324 +5083,278 @@ class ActiveHistoryFlagTest(_fixtures.FixtureTest):
 class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
     run_inserts = None
 
-    def _run_test(self, detached, raiseload, backref, active_history, delete):
-
-        if delete:
-            assert not backref, "delete and backref are mutually exclusive"
-
-        Address, addresses, users, User = (
-            self.classes.Address,
-            self.tables.addresses,
-            self.tables.users,
-            self.classes.User,
-        )
-
-        opts = {}
-        if active_history:
-            opts["active_history"] = True
-        if raiseload:
-            opts["lazy"] = "raise"
-
-        mapper(
-            Address,
-            addresses,
-            properties={
-                "user": relationship(User, back_populates="addresses", **opts)
-            },
-        )
-        mapper(
-            User,
-            users,
-            properties={
-                "addresses": relationship(Address, back_populates="user")
-            },
-        )
-
-        s = Session()
-
-        a1 = Address(email_address="a1")
-        u1 = User(name="u1", addresses=[a1])
-        s.add_all([a1, u1])
-        s.commit()
-
-        if backref:
-            u1.addresses
-
-            if detached:
-                s.expunge(a1)
-
-            def go():
-                u1.addresses = []
-
-            if active_history:
-                if raiseload:
-                    assert_raises_message(
-                        exc.InvalidRequestError,
-                        "'Address.user' is not available due to lazy='raise'",
-                        go,
-                    )
-                    return
-                elif detached:
-                    assert_raises_message(
-                        orm_exc.DetachedInstanceError,
-                        "lazy load operation of attribute 'user' "
-                        "cannot proceed",
-                        go,
-                    )
-                    return
-            go()
-        else:
-            if detached:
-                s.expunge(a1)
-
-            if delete:
-
-                def go():
-                    del a1.user
-
-            else:
-
-                def go():
-                    a1.user = None
-
-            if active_history:
-                if raiseload:
-                    assert_raises_message(
-                        exc.InvalidRequestError,
-                        "'Address.user' is not available due to lazy='raise'",
-                        go,
-                    )
-                    return
-                elif detached:
-                    assert_raises_message(
-                        orm_exc.DetachedInstanceError,
-                        "lazy load operation of attribute 'user' "
-                        "cannot proceed",
-                        go,
-                    )
-                    return
-            go()
-
-        if detached:
-            s.add(a1)
-
-        s.commit()
-
-    def test_replace_m2o(self):
-        self._run_test(
+    @testing.flag_combinations(
+        dict(
             detached=False,
             raiseload=False,
             backref=False,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_detached(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=False,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_raiseload(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=False,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_detached_raiseload(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=False,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_backref(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=False,
             backref=True,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_detached_backref(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=True,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_raiseload_backref(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=True,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_detached_raiseload_backref(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=True,
             delete=False,
             active_history=False,
-        )
-
-    def test_replace_m2o_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=False,
             backref=False,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_detached_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=False,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_raiseload_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=False,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_detached_raiseload_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=False,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_backref_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=False,
             backref=True,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_detached_backref_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=True,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_raiseload_backref_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=True,
             delete=False,
             active_history=True,
-        )
-
-    def test_replace_m2o_detached_raiseload_backref_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=True,
             delete=False,
             active_history=True,
-        )
-
-    def test_delete_m2o(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=False,
             backref=False,
             delete=True,
             active_history=False,
-        )
-
-    def test_delete_m2o_detached(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=False,
             delete=True,
             active_history=False,
-        )
-
-    def test_delete_m2o_raiseload(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=False,
             delete=True,
             active_history=False,
-        )
-
-    def test_delete_m2o_detached_raiseload(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=False,
             delete=True,
             active_history=False,
-        )
-
-    def test_delete_m2o_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=False,
             backref=False,
             delete=True,
             active_history=True,
-        )
-
-    def test_delete_m2o_detached_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=False,
             backref=False,
             delete=True,
             active_history=True,
-        )
-
-    def test_delete_m2o_raiseload_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=False,
             raiseload=True,
             backref=False,
             delete=True,
             active_history=True,
-        )
-
-    def test_delete_m2o_detached_raiseload_activehistory(self):
-        self._run_test(
+        ),
+        dict(
             detached=True,
             raiseload=True,
             backref=False,
             delete=True,
             active_history=True,
+        ),
+    )
+    def test_m2o(self, detached, raiseload, backref, active_history, delete):
+
+        if delete:
+            assert not backref, "delete and backref are mutually exclusive"
+
+        Address, addresses, users, User = (
+            self.classes.Address,
+            self.tables.addresses,
+            self.tables.users,
+            self.classes.User,
+        )
+
+        opts = {}
+        if active_history:
+            opts["active_history"] = True
+        if raiseload:
+            opts["lazy"] = "raise"
+
+        mapper(
+            Address,
+            addresses,
+            properties={
+                "user": relationship(User, back_populates="addresses", **opts)
+            },
         )
+        mapper(
+            User,
+            users,
+            properties={
+                "addresses": relationship(Address, back_populates="user")
+            },
+        )
+
+        s = Session()
+
+        a1 = Address(email_address="a1")
+        u1 = User(name="u1", addresses=[a1])
+        s.add_all([a1, u1])
+        s.commit()
+
+        if backref:
+            u1.addresses
+
+            if detached:
+                s.expunge(a1)
+
+            def go():
+                u1.addresses = []
+
+            if active_history:
+                if raiseload:
+                    assert_raises_message(
+                        exc.InvalidRequestError,
+                        "'Address.user' is not available due to lazy='raise'",
+                        go,
+                    )
+                    return
+                elif detached:
+                    assert_raises_message(
+                        orm_exc.DetachedInstanceError,
+                        "lazy load operation of attribute 'user' "
+                        "cannot proceed",
+                        go,
+                    )
+                    return
+            go()
+        else:
+            if detached:
+                s.expunge(a1)
+
+            if delete:
+
+                def go():
+                    del a1.user
+
+            else:
+
+                def go():
+                    a1.user = None
+
+            if active_history:
+                if raiseload:
+                    assert_raises_message(
+                        exc.InvalidRequestError,
+                        "'Address.user' is not available due to lazy='raise'",
+                        go,
+                    )
+                    return
+                elif detached:
+                    assert_raises_message(
+                        orm_exc.DetachedInstanceError,
+                        "lazy load operation of attribute 'user' "
+                        "cannot proceed",
+                        go,
+                    )
+                    return
+            go()
+
+        if detached:
+            s.add(a1)
+
+        s.commit()
 
 
 class RelationDeprecationTest(fixtures.MappedTest):