import re from sqlalchemy import Column from sqlalchemy import event from sqlalchemy import ForeignKey from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import select from sqlalchemy import Table from sqlalchemy import testing from sqlalchemy.engine import result from sqlalchemy.ext.hybrid import hybrid_method from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.orm import aliased from sqlalchemy.orm import clear_mappers from sqlalchemy.orm import relationship from sqlalchemy.orm import Session from sqlalchemy.orm import synonym from sqlalchemy.orm import util as orm_util from sqlalchemy.orm import with_polymorphic from sqlalchemy.orm.path_registry import PathRegistry from sqlalchemy.orm.path_registry import PathToken from sqlalchemy.orm.path_registry import RootRegistry from sqlalchemy.testing import assert_raises from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import eq_ from sqlalchemy.testing import expect_raises from sqlalchemy.testing import expect_warnings from sqlalchemy.testing import fixtures from sqlalchemy.testing import is_ from sqlalchemy.testing.assertions import is_true from sqlalchemy.testing.fixtures import fixture_session from test.orm import _fixtures from .inheritance import _poly_fixtures class ContextualWarningsTest(fixtures.TestBase): """ Test for #7305 """ @testing.fixture def plain_fixture(cls, decl_base): class Foo(decl_base): __tablename__ = "foo" id = Column(Integer, primary_key=True) decl_base.metadata.create_all(testing.db) return Foo @testing.fixture def overlap_fixture(cls, decl_base): class Foo(decl_base): __tablename__ = "foo" id = Column(Integer, primary_key=True) bars = relationship( "Bar", primaryjoin="Foo.id==Bar.foo_id", ) class Bar(decl_base): __tablename__ = "bar" id = Column(Integer, primary_key=True) foo_id = Column(Integer, ForeignKey("foo.id")) foos = relationship( "Foo", primaryjoin="Bar.foo_id==Foo.id", ) return Foo, Bar def test_configure_mappers_explicit(self, overlap_fixture, decl_base): with expect_warnings( re.escape( "relationship 'Bar.foos' will copy column foo.id to column " "bar.foo_id, which conflicts with relationship(s): 'Foo.bars' " "(copies foo.id to bar.foo_id). " ), ): decl_base.registry.configure() def test_configure_mappers_implicit_aliased(self, overlap_fixture): Foo, Bar = overlap_fixture with expect_warnings( re.escape( "relationship 'Bar.foos' will copy column foo.id " "to column bar.foo_id, which conflicts with" ) + ".*" + re.escape( "(This warning originated from the `configure_mappers()` " "process, which was " "invoked automatically in response to a user-initiated " "operation.)" ), ): FooAlias = aliased(Foo) assert hasattr(FooAlias, "bars") def test_configure_mappers_implicit_instantiate(self, overlap_fixture): Foo, Bar = overlap_fixture with expect_warnings( re.escape( "relationship 'Bar.foos' will copy column foo.id " "to column bar.foo_id, which conflicts with" ) + ".*" + re.escape( "(This warning originated from the `configure_mappers()` " "process, which was " "invoked automatically in response to a user-initiated " "operation.)" ), ): foo = Foo() assert hasattr(foo, "bars") def test_autoflush_implicit(self, plain_fixture): Foo = plain_fixture sess = fixture_session() @event.listens_for(Foo, "before_insert") def emit_a_warning(mapper, connection, state): sess.add(Foo()) sess.add(Foo()) with expect_warnings( re.escape( "Usage of the 'Session.add()' operation is not " "currently supported within the execution stage of the flush " "process. Results may not be consistent. Consider using " "alternative event listeners or connection-level operations " "instead." ) + ".*" + re.escape( "(This warning originated from the Session 'autoflush' " "process, which was invoked automatically in response to a " "user-initiated operation. Consider using ``no_autoflush`` " "context manager if this warning happended while " "initializing objects.)" ), ): sess.execute(select(Foo)) class AliasedClassTest(fixtures.MappedTest, AssertsCompiledSQL): __dialect__ = "default" def _fixture(self, cls, properties={}): table = Table( "point", MetaData(), Column("id", Integer(), primary_key=True), Column("x", Integer), Column("y", Integer), ) clear_mappers() self.mapper_registry.map_imperatively( cls, table, properties=properties ) return table def test_simple(self): class Point: pass table = self._fixture(Point) alias = aliased(Point) assert alias.id assert alias.x assert alias.y assert Point.id.__clause_element__().table is table assert alias.id.__clause_element__().table is not table def test_named_entity(self): class Point: pass self._fixture(Point) alias = aliased(Point, name="pp") self.assert_compile( select(alias), "SELECT pp.id, pp.x, pp.y FROM point AS pp" ) def test_named_selectable(self): class Point: pass table = self._fixture(Point) alias = aliased(table, name="pp") self.assert_compile( select(alias), "SELECT pp.id, pp.x, pp.y FROM point AS pp" ) def test_not_instantiatable(self): class Point: pass self._fixture(Point) alias = aliased(Point) assert_raises(TypeError, alias) def test_instancemethod(self): class Point: def zero(self): self.x, self.y = 0, 0 self._fixture(Point) alias = aliased(Point) assert Point.zero assert getattr(alias, "zero") def test_classmethod(self): class Point: @classmethod def max_x(cls): return 100 self._fixture(Point) alias = aliased(Point) assert Point.max_x assert alias.max_x assert Point.max_x() == alias.max_x() == 100 def test_simple_property(self): class Point: @property def max_x(self): return 100 self._fixture(Point) alias = aliased(Point) assert Point.max_x assert Point.max_x != 100 assert alias.max_x assert Point.max_x is alias.max_x def test_descriptors(self): class descriptor: def __init__(self, fn): self.fn = fn def __get__(self, obj, owner): if obj is not None: return self.fn(obj, obj) else: return self def method(self): return "method" class Point: center = (0, 0) @descriptor def thing(self, arg): return arg.center self._fixture(Point) alias = aliased(Point) assert Point.thing != (0, 0) assert Point().thing == (0, 0) assert Point.thing.method() == "method" assert alias.thing != (0, 0) assert alias.thing.method() == "method" def _assert_has_table(self, expr, table): from sqlalchemy import Column # override testlib's override for child in expr.get_children(): if isinstance(child, Column): assert child.table is table def test_hybrid_descriptor_one(self): class Point: def __init__(self, x, y): self.x, self.y = x, y @hybrid_method def left_of(self, other): return self.x < other.x self._fixture(Point) alias = aliased(Point) sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.left_of(Point)), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x < point.x", ) def test_hybrid_descriptor_two(self): class Point: def __init__(self, x, y): self.x, self.y = x, y @hybrid_property def double_x(self): return self.x * 2 self._fixture(Point) alias = aliased(Point) eq_(str(Point.double_x), "Point.double_x") eq_(str(alias.double_x), "aliased(Point).double_x") eq_(str(Point.double_x.__clause_element__()), "point.x * :x_1") eq_(str(alias.double_x.__clause_element__()), "point_1.x * :x_1") sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.double_x > Point.x), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x * :x_1 > point.x", ) def test_hybrid_descriptor_three(self): class Point: def __init__(self, x, y): self.x, self.y = x, y @hybrid_property def x_alone(self): return self.x self._fixture(Point) alias = aliased(Point) eq_(str(Point.x_alone), "Point.x_alone") eq_(str(alias.x_alone), "aliased(Point).x_alone") # from __clause_element__() perspective, Point.x_alone # and Point.x return the same thing, so that's good eq_(str(Point.x.__clause_element__()), "point.x") eq_(str(Point.x_alone.__clause_element__()), "point.x") # same for the alias eq_(str(alias.x + 1), "point_1.x + :x_1") eq_(str(alias.x_alone + 1), "point_1.x + :x_1") point_mapper = inspect(Point) eq_( Point.x_alone._annotations, { "entity_namespace": point_mapper, "parententity": point_mapper, "parentmapper": point_mapper, "proxy_key": "x_alone", "proxy_owner": point_mapper, }, ) eq_( Point.x._annotations, { "entity_namespace": point_mapper, "parententity": point_mapper, "parentmapper": point_mapper, "proxy_key": "x", "proxy_owner": point_mapper, }, ) eq_(str(alias.x_alone == alias.x), "point_1.x = point_1.x") a2 = aliased(Point) eq_(str(a2.x_alone == alias.x), "point_1.x = point_2.x") eq_( a2.x._annotations, { "entity_namespace": inspect(a2), "parententity": inspect(a2), "parentmapper": point_mapper, "proxy_key": "x", "proxy_owner": inspect(a2), }, ) sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.x_alone > Point.x), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x > point.x", ) def test_proxy_descriptor_one(self): class Point: def __init__(self, x, y): self.x, self.y = x, y self._fixture(Point, properties={"x_syn": synonym("x")}) alias = aliased(Point) eq_(str(Point.x_syn), "Point.x_syn") eq_(str(alias.x_syn), "aliased(Point).x_syn") sess = fixture_session() self.assert_compile( sess.query(alias.x_syn).filter(alias.x_syn > Point.x_syn), "SELECT point_1.x AS point_1_x FROM point AS point_1, point " "WHERE point_1.x > point.x", ) def test_meta_getattr_one(self): class MetaPoint(type): def __getattr__(cls, key): if key == "x_syn": return cls.x raise AttributeError(key) class Point(metaclass=MetaPoint): pass self._fixture(Point) alias = aliased(Point) eq_(str(Point.x_syn), "Point.x") eq_(str(alias.x_syn), "aliased(Point).x") # from __clause_element__() perspective, Point.x_syn # and Point.x return the same thing, so that's good eq_(str(Point.x.__clause_element__()), "point.x") eq_(str(Point.x_syn.__clause_element__()), "point.x") # same for the alias eq_(str(alias.x + 1), "point_1.x + :x_1") eq_(str(alias.x_syn + 1), "point_1.x + :x_1") is_(Point.x_syn.__clause_element__(), Point.x.__clause_element__()) eq_(str(alias.x_syn == alias.x), "point_1.x = point_1.x") a2 = aliased(Point) eq_(str(a2.x_syn == alias.x), "point_1.x = point_2.x") sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.x_syn > Point.x), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x > point.x", ) def test_meta_getattr_two(self): class MetaPoint(type): def __getattr__(cls, key): if key == "double_x": return cls._impl_double_x raise AttributeError(key) class Point(metaclass=MetaPoint): @hybrid_property def _impl_double_x(self): return self.x * 2 self._fixture(Point) alias = aliased(Point) eq_(str(Point.double_x), "Point._impl_double_x") eq_(str(alias.double_x), "aliased(Point)._impl_double_x") eq_(str(Point.double_x.__clause_element__()), "point.x * :x_1") eq_(str(alias.double_x.__clause_element__()), "point_1.x * :x_1") sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.double_x > Point.x), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x * :x_1 > point.x", ) def test_meta_getattr_three(self): class MetaPoint(type): def __getattr__(cls, key): @hybrid_property def double_x(me): return me.x * 2 if key == "double_x": return double_x.__get__(None, cls) raise AttributeError(key) class Point(metaclass=MetaPoint): pass self._fixture(Point) alias = aliased(Point) eq_(str(Point.double_x.__clause_element__()), "point.x * :x_1") eq_(str(alias.double_x.__clause_element__()), "point_1.x * :x_1") sess = fixture_session() self.assert_compile( sess.query(alias).filter(alias.double_x > Point.x), "SELECT point_1.id AS point_1_id, point_1.x AS point_1_x, " "point_1.y AS point_1_y FROM point AS point_1, point " "WHERE point_1.x * :x_1 > point.x", ) def test_parententity_vs_parentmapper(self): class Point: pass self._fixture(Point, properties={"x_syn": synonym("x")}) pa = aliased(Point) is_(Point.x_syn._parententity, inspect(Point)) is_(Point.x._parententity, inspect(Point)) is_(Point.x_syn._parentmapper, inspect(Point)) is_(Point.x._parentmapper, inspect(Point)) is_( Point.x_syn.__clause_element__()._annotations["parententity"], inspect(Point), ) is_( Point.x.__clause_element__()._annotations["parententity"], inspect(Point), ) is_( Point.x_syn.__clause_element__()._annotations["parentmapper"], inspect(Point), ) is_( Point.x.__clause_element__()._annotations["parentmapper"], inspect(Point), ) pa = aliased(Point) is_(pa.x_syn._parententity, inspect(pa)) is_(pa.x._parententity, inspect(pa)) is_(pa.x_syn._parentmapper, inspect(Point)) is_(pa.x._parentmapper, inspect(Point)) is_( pa.x_syn.__clause_element__()._annotations["parententity"], inspect(pa), ) is_( pa.x.__clause_element__()._annotations["parententity"], inspect(pa) ) is_( pa.x_syn.__clause_element__()._annotations["parentmapper"], inspect(Point), ) is_( pa.x.__clause_element__()._annotations["parentmapper"], inspect(Point), ) class IdentityKeyTest(_fixtures.FixtureTest): run_inserts = None def _cases(): return testing.combinations( (orm_util,), (Session,), argnames="ormutil" ) @_cases() def test_identity_key_1(self, ormutil): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) key = ormutil.identity_key(User, [1]) eq_(key, (User, (1,), None)) key = ormutil.identity_key(User, ident=[1]) eq_(key, (User, (1,), None)) @_cases() def test_identity_key_scalar(self, ormutil): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) key = ormutil.identity_key(User, 1) eq_(key, (User, (1,), None)) key = ormutil.identity_key(User, ident=1) eq_(key, (User, (1,), None)) @_cases() def test_identity_key_2(self, ormutil): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) s = fixture_session() u = User(name="u1") s.add(u) s.flush() key = ormutil.identity_key(instance=u) eq_(key, (User, (u.id,), None)) @_cases() @testing.combinations("dict", "row", "mapping", argnames="rowtype") def test_identity_key_3(self, ormutil, rowtype): """test a real Row works with identity_key. this was broken w/ 1.4 future mode as we are assuming a mapping here. to prevent regressions, identity_key now accepts any of dict, RowMapping, Row for the "row". found_during_type_annotation """ User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) if rowtype == "dict": row = {users.c.id: 1, users.c.name: "Frank"} elif rowtype in ("mapping", "row"): row = result.result_tuple([users.c.id, users.c.name])((1, "Frank")) if rowtype == "mapping": row = row._mapping key = ormutil.identity_key(User, row=row) eq_(key, (User, (1,), None)) def test_identity_key_token(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) key = orm_util.identity_key(User, [1], identity_token="token") eq_(key, (User, (1,), "token")) key = orm_util.identity_key(User, ident=[1], identity_token="token") eq_(key, (User, (1,), "token")) class PathRegistryTest(_fixtures.FixtureTest): run_setup_mappers = "once" run_inserts = None run_deletes = None @classmethod def setup_mappers(cls): cls._setup_stock_mapping() def test_root_registry(self): umapper = inspect(self.classes.User) is_(RootRegistry()[umapper], umapper._path_registry) eq_(RootRegistry()[umapper], PathRegistry.coerce((umapper,))) def test_expand(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce((umapper,)) eq_( path[umapper.attrs.addresses][amapper][ amapper.attrs.email_address ], PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ), ) def test_entity_boolean(self): umapper = inspect(self.classes.User) path = PathRegistry.coerce((umapper,)) is_(bool(path), True) def test_key_boolean(self): umapper = inspect(self.classes.User) path = PathRegistry.coerce((umapper, umapper.attrs.addresses)) is_(bool(path), True) def test_aliased_class(self): User = self.classes.User ua = aliased(User) ua_insp = inspect(ua) path = PathRegistry.coerce((ua_insp, ua_insp.mapper.attrs.addresses)) assert path.parent.is_aliased_class def test_indexed_entity(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) is_(path[0], umapper) is_(path[2], amapper) def test_indexed_key_token(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, PathToken.intern(":*"), ) ) is_true(path.is_token) eq_(path[1], umapper.attrs.addresses) eq_(path[3], ":*") with expect_raises(IndexError): path[amapper] def test_slice_token(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, PathToken.intern(":*"), ) ) is_true(path.is_token) eq_(path[1:3], (umapper.attrs.addresses, amapper)) def test_indexed_key(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) eq_(path[1], umapper.attrs.addresses) eq_(path[3], amapper.attrs.email_address) def test_slice(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) path = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) eq_(path[1:3], (umapper.attrs.addresses, amapper)) def test_addition(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((amapper, amapper.attrs.email_address)) eq_( p1 + p2, PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ), ) def test_length(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) pneg1 = PathRegistry.coerce(()) p0 = PathRegistry.coerce((umapper,)) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) eq_(len(pneg1), 0) eq_(len(p0), 1) eq_(len(p1), 2) eq_(len(p2), 3) eq_(len(p3), 4) eq_(pneg1.length, 0) eq_(p0.length, 1) eq_(p1.length, 2) eq_(p2.length, 3) eq_(p3.length, 4) def test_eq(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) u_alias = inspect(aliased(self.classes.User)) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p3 = PathRegistry.coerce((umapper, umapper.attrs.name)) p4 = PathRegistry.coerce((u_alias, umapper.attrs.addresses)) p5 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p6 = PathRegistry.coerce( (amapper, amapper.attrs.user, umapper, umapper.attrs.addresses) ) p7 = PathRegistry.coerce( ( amapper, amapper.attrs.user, umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) is_(p1 == p2, True) is_(p1 == p3, False) is_(p1 == p4, False) is_(p1 == p5, False) is_(p6 == p7, False) is_(p6 == p7.parent.parent, True) is_(p1 != p2, False) is_(p1 != p3, True) is_(p1 != p4, True) is_(p1 != p5, True) def test_eq_non_path(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) u_alias = inspect(aliased(self.classes.User)) p1 = PathRegistry.coerce((umapper,)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p3 = PathRegistry.coerce((u_alias, umapper.attrs.addresses)) p4 = PathRegistry.coerce((u_alias, umapper.attrs.addresses, amapper)) p5 = PathRegistry.coerce((u_alias,)).token(":*") non_object = 54.1432 for obj in [p1, p2, p3, p4, p5]: with expect_warnings( "Comparison of PathRegistry to " "<.* 'float'> is not supported" ): is_(obj == non_object, False) with expect_warnings( "Comparison of PathRegistry to " "<.* 'float'> is not supported" ): is_(obj != non_object, True) def test_contains_mapper(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) assert p1.contains_mapper(umapper) assert not p1.contains_mapper(amapper) def test_path(self): umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((amapper, amapper.attrs.email_address)) eq_(p1.path, (umapper, umapper.attrs.addresses)) eq_(p2.path, (umapper, umapper.attrs.addresses, amapper)) eq_(p3.path, (amapper, amapper.attrs.email_address)) def test_registry_set(self): reg = {} umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((amapper, amapper.attrs.email_address)) p1.set(reg, "p1key", "p1value") p2.set(reg, "p2key", "p2value") p3.set(reg, "p3key", "p3value") eq_( reg, { ("p1key", p1.path): "p1value", ("p2key", p2.path): "p2value", ("p3key", p3.path): "p3value", }, ) def test_registry_get(self): reg = {} umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((amapper, amapper.attrs.email_address)) reg.update( { ("p1key", p1.path): "p1value", ("p2key", p2.path): "p2value", ("p3key", p3.path): "p3value", } ) eq_(p1.get(reg, "p1key"), "p1value") eq_(p2.get(reg, "p2key"), "p2value") eq_(p2.get(reg, "p1key"), None) eq_(p3.get(reg, "p3key"), "p3value") eq_(p3.get(reg, "p1key"), None) def test_registry_contains(self): reg = {} umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((amapper, amapper.attrs.email_address)) reg.update( { ("p1key", p1.path): "p1value", ("p2key", p2.path): "p2value", ("p3key", p3.path): "p3value", } ) assert p1.contains(reg, "p1key") assert not p1.contains(reg, "p2key") assert p3.contains(reg, "p3key") assert not p2.contains(reg, "fake") def test_registry_setdefault(self): reg = {} umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) reg.update({("p1key", p1.path): "p1value"}) p1.setdefault(reg, "p1key", "p1newvalue_a") p1.setdefault(reg, "p1key_new", "p1newvalue_b") p2.setdefault(reg, "p2key", "p2newvalue") eq_( reg, { ("p1key", p1.path): "p1value", ("p1key_new", p1.path): "p1newvalue_b", ("p2key", p2.path): "p2newvalue", }, ) def test_serialize(self): User = self.classes.User Address = self.classes.Address umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) eq_(p1.serialize(), [(User, "addresses"), (Address, "email_address")]) eq_(p2.serialize(), [(User, "addresses"), (Address, None)]) eq_(p3.serialize(), [(User, "addresses")]) def test_deseralize(self): User = self.classes.User Address = self.classes.Address umapper = inspect(self.classes.User) amapper = inspect(self.classes.Address) p1 = PathRegistry.coerce( ( umapper, umapper.attrs.addresses, amapper, amapper.attrs.email_address, ) ) p2 = PathRegistry.coerce((umapper, umapper.attrs.addresses, amapper)) p3 = PathRegistry.coerce((umapper, umapper.attrs.addresses)) eq_( PathRegistry.deserialize( [(User, "addresses"), (Address, "email_address")] ), p1, ) eq_( PathRegistry.deserialize([(User, "addresses"), (Address, None)]), p2, ) eq_(PathRegistry.deserialize([(User, "addresses")]), p3) class PathRegistryInhTest(_poly_fixtures._Polymorphic): run_setup_mappers = "once" run_inserts = None run_deletes = None def test_plain(self): Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer pmapper = inspect(Person) emapper = inspect(Engineer) p1 = PathRegistry.coerce((pmapper, emapper.attrs.machines)) # given a mapper and an attribute on a subclass, # the path converts what you get to be against that subclass eq_(p1.path, (emapper, emapper.attrs.machines)) def test_plain_compound(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer cmapper = inspect(Company) pmapper = inspect(Person) emapper = inspect(Engineer) p1 = PathRegistry.coerce( (cmapper, cmapper.attrs.employees, pmapper, emapper.attrs.machines) ) # given a mapper and an attribute on a subclass, # the path converts what you get to be against that subclass eq_( p1.path, ( cmapper, cmapper.attrs.employees, emapper, emapper.attrs.machines, ), ) def test_plain_aliased(self): Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer emapper = inspect(Engineer) p_alias = aliased(Person) p_alias = inspect(p_alias) p1 = PathRegistry.coerce((p_alias, emapper.attrs.machines)) # plain AliasedClass - the path keeps that AliasedClass directly # as is in the path eq_(p1.path, (p_alias, emapper.attrs.machines)) def test_plain_aliased_compound(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer cmapper = inspect(Company) emapper = inspect(Engineer) c_alias = aliased(Company) p_alias = aliased(Person) c_alias = inspect(c_alias) p_alias = inspect(p_alias) p1 = PathRegistry.coerce( (c_alias, cmapper.attrs.employees, p_alias, emapper.attrs.machines) ) # plain AliasedClass - the path keeps that AliasedClass directly # as is in the path eq_( p1.path, ( c_alias, cmapper.attrs.employees, p_alias, emapper.attrs.machines, ), ) def test_with_poly_sub(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer emapper = inspect(Engineer) cmapper = inspect(Company) p_poly = with_polymorphic(Person, [Engineer]) e_poly_insp = inspect(p_poly.Engineer) # noqa - used by comment below p_poly_insp = inspect(p_poly) p1 = PathRegistry.coerce((p_poly_insp, emapper.attrs.machines)) # changes as of #5082: when a with_polymorphic is in the middle # of a path, the natural path makes sure it uses the base mappers, # however when it's at the root, the with_polymorphic stays in # the natural path # this behavior is the same as pre #5082, it was temporarily changed # but this proved to be incorrect. The path starts on a # with_polymorphic(), so a Query will "naturally" construct a path # that comes from that wp. eq_(p1.path, (e_poly_insp, emapper.attrs.machines)) eq_(p1.natural_path, (e_poly_insp, emapper.attrs.machines)) # this behavior is new as of the final version of #5082. # the path starts on a normal entity and has a with_polymorphic # in the middle, for this to match what Query will generate it needs # to use the non aliased mappers in the natural path. p2 = PathRegistry.coerce( ( cmapper, cmapper.attrs.employees, p_poly_insp, emapper.attrs.machines, ) ) eq_( p2.path, ( cmapper, cmapper.attrs.employees, e_poly_insp, emapper.attrs.machines, ), ) eq_( p2.natural_path, ( cmapper, cmapper.attrs.employees, emapper, emapper.attrs.machines, ), ) def test_with_poly_base_two(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer cmapper = inspect(Company) pmapper = inspect(Person) p_poly = with_polymorphic(Person, [Engineer]) e_poly_insp = inspect(p_poly.Engineer) # noqa - used by comment below p_poly_insp = inspect(p_poly) p1 = PathRegistry.coerce( ( cmapper, cmapper.attrs.employees, p_poly_insp, pmapper.attrs.paperwork, ) ) eq_( p1.path, ( cmapper, cmapper.attrs.employees, p_poly_insp, pmapper.attrs.paperwork, ), ) eq_( p1.natural_path, ( cmapper, cmapper.attrs.employees, pmapper, pmapper.attrs.paperwork, ), ) def test_nonpoly_oftype_aliased_subclass_onroot(self): Engineer = _poly_fixtures.Engineer eng_alias = aliased(Engineer) ea_insp = inspect(eng_alias) p1 = PathRegistry.coerce((ea_insp, ea_insp.mapper.attrs.paperwork)) eq_(p1.path, (ea_insp, ea_insp.mapper.attrs.paperwork)) eq_(p1.natural_path, (ea_insp, ea_insp.mapper.attrs.paperwork)) def test_nonpoly_oftype_aliased_subclass(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer cmapper = inspect(Company) pmapper = inspect(Person) eng_alias = aliased(Engineer) ea_insp = inspect(eng_alias) p1 = PathRegistry.coerce( ( cmapper, cmapper.attrs.employees, ea_insp, ea_insp.mapper.attrs.paperwork, ) ) eq_( p1.path, ( cmapper, cmapper.attrs.employees, ea_insp, ea_insp.mapper.attrs.paperwork, ), ) eq_( p1.natural_path, ( cmapper, cmapper.attrs.employees, pmapper, pmapper.attrs.paperwork, ), ) def test_nonpoly_oftype_subclass(self): Company = _poly_fixtures.Company Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer emapper = inspect(Engineer) cmapper = inspect(Company) pmapper = inspect(Person) p1 = PathRegistry.coerce( ( cmapper, cmapper.attrs.employees, emapper, emapper.attrs.paperwork, ) ) eq_( p1.path, ( cmapper, cmapper.attrs.employees, pmapper, pmapper.attrs.paperwork, ), ) eq_( p1.natural_path, ( cmapper, cmapper.attrs.employees, pmapper, pmapper.attrs.paperwork, ), ) def test_with_poly_base_one(self): Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer pmapper = inspect(Person) emapper = inspect(Engineer) p_poly = with_polymorphic(Person, [Engineer]) p_poly = inspect(p_poly) # "name" is actually on Person, not Engineer p1 = PathRegistry.coerce((p_poly, emapper.attrs.name)) # polymorphic AliasedClass - because "name" is on Person, # we get Person, not Engineer eq_(p1.path, (p_poly, pmapper.attrs.name)) def test_with_poly_use_mapper(self): Person = _poly_fixtures.Person Engineer = _poly_fixtures.Engineer emapper = inspect(Engineer) p_poly = with_polymorphic(Person, [Engineer], _use_mapper_path=True) p_poly = inspect(p_poly) p1 = PathRegistry.coerce((p_poly, emapper.attrs.machines)) # polymorphic AliasedClass with the "use_mapper_path" flag - # the AliasedClass acts just like the base mapper eq_(p1.path, (emapper, emapper.attrs.machines))