From: Mike Bayer Date: Fri, 22 Nov 2019 15:54:22 +0000 (-0500) Subject: Introduce flag combinations fixture X-Git-Tag: rel_1_3_12~14 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=3a8bdd3f6bb885b252412450431d3a106b7ee4b4;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git 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 5e303be00a..ed99b1eb2b 100644 --- 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 64738ad2f6..87c461fd2b 100644 --- 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 2d686c7e90..60cfde3c3d 100644 --- 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):