return combinations(*arg_iterable, **kw)
+class _variation_base(object):
+ __slots__ = ("name", "argname")
+
+ def __init__(self, case, argname, case_names):
+ self.name = case
+ self.argname = argname
+ for casename in case_names:
+ setattr(self, casename, casename == case)
+
+ def __bool__(self):
+ return self.name == self.argname
+
+ def __nonzero__(self):
+ return not self.__bool__()
+
+
+def variation(argname, cases):
+ """a helper around testing.combinations that provides a single namespace
+ that can be used as a switch.
+
+ e.g.::
+
+ @testing.variation("querytyp", ["select", "subquery", "legacy_query"])
+ @testing.variation("lazy", ["select", "raise", "raise_on_sql"])
+ def test_thing(
+ self,
+ querytyp,
+ lazy,
+ decl_base
+ ):
+ class Thing(decl_base):
+ __tablename__ = 'thing'
+
+ # use name directly
+ rel = relationship("Rel", lazy=lazy.name)
+
+ # use as a switch
+ if querytyp.select:
+ stmt = select(Thing)
+ elif querytyp.subquery:
+ stmt = select(Thing).subquery()
+ elif querytyp.legacy_query:
+ stmt = Session.query(Thing)
+ else:
+ assert False
+
+
+ The variable provided is a slots object of boolean variables, as well
+ as the name of the case itself under the attribute ".name"
+
+ """
+
+ case_names = [
+ argname if c is True else "not_" + argname if c is False else c
+ for c in cases
+ ]
+
+ typ = type(
+ argname,
+ (_variation_base,),
+ {
+ "__slots__": tuple(case_names),
+ },
+ )
+
+ return combinations(
+ *[
+ (casename, typ(casename, argname, case_names))
+ for casename in case_names
+ ],
+ id_="ia",
+ argnames=argname
+ )
+
+
def fixture(*arg, **kw):
return _fixture_functions.fixture(*arg, **kw)
except sa.exc.InvalidRequestError as e:
assert "load=False option does not support" in str(e)
- @testing.combinations("viewonly", "normal", argnames="viewonly")
- @testing.combinations("load", "noload", argnames="load")
- @testing.combinations("select", "raise", "raise_on_sql", argnames="lazy")
- @testing.combinations(
- "merge_persistent", "merge_detached", argnames="merge_persistent"
+ @testing.variation("viewonly", ["viewonly", "normal"])
+ @testing.variation("load", ["load", "noload"])
+ @testing.variation("lazy", ["select", "raise", "raise_on_sql"])
+ @testing.variation(
+ "merge_persistent", ["merge_persistent", "merge_detached"]
)
- @testing.combinations("detached", "persistent", argnames="detach_original")
- @testing.combinations("o2m", "m2o", argnames="direction")
+ @testing.variation("detach_original", ["detach", "persistent"])
+ @testing.variation("direction", ["o2m", "m2o"])
def test_relationship_population_maintained(
self,
viewonly,
properties={
"addresses": relationship(
Address,
- viewonly=viewonly == "viewonly",
- lazy=lazy,
+ viewonly=viewonly.viewonly,
+ lazy=lazy.name,
back_populates="user",
order_by=addresses.c.id,
)
properties={
"user": relationship(
User,
- viewonly=viewonly == "viewonly",
- lazy=lazy,
+ viewonly=viewonly.viewonly,
+ lazy=lazy.name,
back_populates="addresses",
)
},
)
s.commit()
- if direction == "o2m":
+ if direction.o2m:
cls_to_merge = User
obj_to_merge = (
s.scalars(select(User).options(joinedload(User.addresses)))
)
attrname = "addresses"
- elif direction == "m2o":
+ elif direction.m2o:
cls_to_merge = Address
obj_to_merge = (
s.scalars(
s2 = Session(testing.db)
- if merge_persistent == "merge_persistent":
+ if merge_persistent.merge_persistent:
target_persistent = s2.get(cls_to_merge, obj_to_merge.id) # noqa
- if detach_original == "detach":
+ if detach_original.detach:
s.expunge(obj_to_merge)
with self.sql_execution_asserter(testing.db) as assert_:
- merged_object = s2.merge(obj_to_merge, load=load == "load")
+ merged_object = s2.merge(obj_to_merge, load=load.load)
assert_.assert_(
CountStatements(
0
- if load == "noload"
+ if load.noload
else 1
- if merge_persistent == "merge_persistent"
+ if merge_persistent.merge_persistent
else 2
)
)
assert attrname in merged_object.__dict__
with self.sql_execution_asserter(testing.db) as assert_:
- if direction == "o2m":
+ if direction.o2m:
eq_(
merged_object.addresses,
[
for i in range(1, 4)
],
)
- elif direction == "m2o":
+ elif direction.m2o:
eq_(merged_object.user, User(id=1, name="u1"))
assert_.assert_(CountStatements(0))