2
from sqlalchemy.test.testing import eq_, assert_raises, \
2
from test.lib.testing import eq_, assert_raises, \
3
3
assert_raises_message
4
4
from sqlalchemy.ext import declarative as decl
5
5
from sqlalchemy import exc
6
6
import sqlalchemy as sa
7
from sqlalchemy.test import testing
7
from test.lib import testing
8
8
from sqlalchemy import MetaData, Integer, String, ForeignKey, \
9
9
ForeignKeyConstraint, Index
10
from sqlalchemy.test.schema import Table, Column
10
from test.lib.schema import Table, Column
11
11
from sqlalchemy.orm import relationship, create_session, class_mapper, \
12
joinedload, compile_mappers, backref, clear_mappers, \
13
polymorphic_union, deferred, column_property
14
from sqlalchemy.test.testing import eq_
12
joinedload, configure_mappers, backref, clear_mappers, \
13
polymorphic_union, deferred, column_property, composite,\
15
from test.lib.testing import eq_
15
16
from sqlalchemy.util import classproperty
16
from test.orm._base import ComparableEntity, MappedTest
17
17
from sqlalchemy.ext.declarative import declared_attr
18
from test.lib import fixtures
19
class DeclarativeTestBase(testing.TestBase, testing.AssertsExecutionResults):
20
class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
22
23
Base = decl.declarative_base(testing.db)
24
25
def teardown(self):
26
28
Base.metadata.drop_all()
28
30
class DeclarativeTest(DeclarativeTestBase):
29
31
def test_basic(self):
30
class User(Base, ComparableEntity):
32
class User(Base, fixtures.ComparableEntity):
31
33
__tablename__ = 'users'
33
35
id = Column('id', Integer, primary_key=True,
231
233
assert_raises_message(exc.InvalidRequestError,
232
234
"'addresses' is not an instance of "
233
"ColumnProperty", compile_mappers)
235
"ColumnProperty", configure_mappers)
235
237
def test_string_dependency_resolution_two(self):
237
class User(Base, ComparableEntity):
239
class User(Base, fixtures.ComparableEntity):
238
240
__tablename__ = 'users'
239
241
id = Column(Integer, primary_key=True,
240
242
test_needs_autoincrement=True)
241
243
name = Column(String(50))
243
class Bar(Base, ComparableEntity):
245
class Bar(Base, fixtures.ComparableEntity):
244
246
__tablename__ = 'bar'
245
247
id = Column(Integer, primary_key=True)
246
248
rel = relationship('User',
249
251
assert_raises_message(exc.InvalidRequestError,
250
252
"does not have a mapped column named "
251
"'__table__'", compile_mappers)
253
"'__table__'", configure_mappers)
253
255
def test_string_dependency_resolution_no_magic(self):
254
256
"""test that full tinkery expressions work as written"""
256
class User(Base, ComparableEntity):
258
class User(Base, fixtures.ComparableEntity):
258
260
__tablename__ = 'users'
259
261
id = Column(Integer, primary_key=True)
261
263
primaryjoin='User.id==Address.user_id.prop.columns['
264
class Address(Base, ComparableEntity):
266
class Address(Base, fixtures.ComparableEntity):
266
268
__tablename__ = 'addresses'
267
269
id = Column(Integer, primary_key=True)
268
270
user_id = Column(Integer, ForeignKey('users.id'))
271
273
eq_(str(User.addresses.prop.primaryjoin),
272
274
'users.id = addresses.user_id')
274
276
def test_string_dependency_resolution_in_backref(self):
276
class User(Base, ComparableEntity):
278
class User(Base, fixtures.ComparableEntity):
278
280
__tablename__ = 'users'
279
281
id = Column(Integer, primary_key=True)
282
284
primaryjoin='User.id==Address.user_id',
285
class Address(Base, ComparableEntity):
287
class Address(Base, fixtures.ComparableEntity):
287
289
__tablename__ = 'addresses'
288
290
id = Column(Integer, primary_key=True)
289
291
email = Column(String(50))
290
292
user_id = Column(Integer, ForeignKey('users.id'))
293
295
eq_(str(User.addresses.property.primaryjoin),
294
296
str(Address.user.property.primaryjoin))
296
298
def test_string_dependency_resolution_tables(self):
298
class User(Base, ComparableEntity):
300
class User(Base, fixtures.ComparableEntity):
300
302
__tablename__ = 'users'
301
303
id = Column(Integer, primary_key=True)
316
318
Column('user_id', Integer,
317
319
ForeignKey('users.id')), Column('prop_id',
318
320
Integer, ForeignKey('props.id')))
322
assert class_mapper(User).get_property('props').secondary \
325
def test_string_dependency_resolution_schemas(self):
326
Base = decl.declarative_base()
330
__tablename__ = 'users'
331
__table_args__ = {'schema':'fooschema'}
333
id = Column(Integer, primary_key=True)
334
name = Column(String(50))
335
props = relationship('Prop', secondary='fooschema.user_to_prop',
336
primaryjoin='User.id==fooschema.user_to_prop.c.user_id',
337
secondaryjoin='fooschema.user_to_prop.c.prop_id==Prop.id',
342
__tablename__ = 'props'
343
__table_args__ = {'schema':'fooschema'}
345
id = Column(Integer, primary_key=True)
346
name = Column(String(50))
348
user_to_prop = Table('user_to_prop', Base.metadata,
349
Column('user_id', Integer, ForeignKey('fooschema.users.id')),
350
Column('prop_id',Integer, ForeignKey('fooschema.props.id')),
320
354
assert class_mapper(User).get_property('props').secondary \
323
357
def test_uncompiled_attributes_in_relationship(self):
325
class Address(Base, ComparableEntity):
359
class Address(Base, fixtures.ComparableEntity):
327
361
__tablename__ = 'addresses'
328
362
id = Column(Integer, primary_key=True,
386
420
__tablename__ = 'users'
387
421
id = Column('id', Integer, primary_key=True)
388
addresses = relationship('Addresss')
422
addresses = relationship('Address')
390
424
# hasattr() on a compile-loaded attribute
392
hasattr(User.addresses, 'property')
426
hasattr(User.addresses, 'property')
427
except exc.InvalidRequestError:
428
assert sa.util.compat.py32
394
430
# the exception is preserved. Remains the
395
431
# same through repeated calls.
589
625
eq_(sess.query(User).all(), [User(name='u1',
590
626
addresses=[Address(email='one'), Address(email='two')])])
592
@testing.uses_deprecated()
593
def test_custom_mapper(self):
595
class MyExt(sa.orm.MapperExtension):
597
def create_instance(self):
600
def mymapper(cls, tbl, **kwargs):
601
kwargs['extension'] = MyExt()
602
return sa.orm.mapper(cls, tbl, **kwargs)
604
from sqlalchemy.orm.mapper import Mapper
606
class MyMapper(Mapper):
608
def __init__(self, *args, **kwargs):
609
kwargs['extension'] = MyExt()
610
Mapper.__init__(self, *args, **kwargs)
612
from sqlalchemy.orm import scoping
613
ss = scoping.ScopedSession(create_session)
614
ss.extension = MyExt()
615
ss_mapper = ss.mapper
616
for mapperfunc in mymapper, MyMapper, ss_mapper:
617
base = decl.declarative_base()
621
__tablename__ = 'foo'
622
__mapper_cls__ = mapperfunc
623
id = Column(Integer, primary_key=True)
625
eq_(Foo.__mapper__.compile().extension.create_instance(),
627
base = decl.declarative_base(mapper=mapperfunc)
631
__tablename__ = 'foo'
632
id = Column(Integer, primary_key=True)
634
eq_(Foo.__mapper__.compile().extension.create_instance(),
628
def test_custom_mapper_attribute(self):
630
def mymapper(cls, tbl, **kwargs):
631
m = sa.orm.mapper(cls, tbl, **kwargs)
635
base = decl.declarative_base()
638
__tablename__ = 'foo'
639
__mapper_cls__ = mymapper
640
id = Column(Integer, primary_key=True)
642
eq_(Foo.__mapper__.CHECK, True)
644
def test_custom_mapper_argument(self):
646
def mymapper(cls, tbl, **kwargs):
647
m = sa.orm.mapper(cls, tbl, **kwargs)
651
base = decl.declarative_base(mapper=mymapper)
654
__tablename__ = 'foo'
655
id = Column(Integer, primary_key=True)
657
eq_(Foo.__mapper__.CHECK, True)
637
659
@testing.emits_warning('Ignoring declarative-like tuple value of '
652
674
'Mapper Mapper|User|users could not '
653
675
'assemble any primary key', define)
655
def test_table_args_bad_format(self):
660
__tablename__ = 'foo'
661
__table_args__ = ForeignKeyConstraint(['id'], ['foo.id'
663
id = Column('id', Integer, primary_key=True)
665
assert_raises_message(sa.exc.ArgumentError,
666
'Tuple form of __table_args__ is ', err)
677
def test_table_args_no_dict(self):
681
__tablename__ = 'foo'
682
__table_args__ = ForeignKeyConstraint(['id'], ['foo.bar']),
683
id = Column('id', Integer, primary_key=True)
684
bar = Column('bar', Integer)
686
assert Foo1.__table__.c.id.references(Foo1.__table__.c.bar)
668
688
def test_table_args_type(self):
834
854
def test_column_properties_2(self):
836
class Address(Base, ComparableEntity):
856
class Address(Base, fixtures.ComparableEntity):
838
858
__tablename__ = 'addresses'
839
859
id = Column(Integer, primary_key=True)
840
860
email = Column(String(50))
841
861
user_id = Column(Integer, ForeignKey('users.id'))
843
class User(Base, ComparableEntity):
863
class User(Base, fixtures.ComparableEntity):
845
865
__tablename__ = 'users'
846
866
id = Column('id', Integer, primary_key=True)
878
898
self.assert_sql_count(testing.db, go, 1)
900
def test_composite_inline(self):
901
class AddressComposite(fixtures.ComparableEntity):
902
def __init__(self, street, state):
905
def __composite_values__(self):
906
return [self.street, self.state]
908
class User(Base, fixtures.ComparableEntity):
909
__tablename__ = 'user'
910
id = Column(Integer, primary_key=True,
911
test_needs_autoincrement=True)
912
address = composite(AddressComposite,
913
Column('street', String(50)),
914
Column('state', String(2)),
917
Base.metadata.create_all()
920
address=AddressComposite('123 anywhere street',
925
sess.query(User).all(),
926
[User(address=AddressComposite('123 anywhere street',
930
def test_composite_separate(self):
931
class AddressComposite(fixtures.ComparableEntity):
932
def __init__(self, street, state):
935
def __composite_values__(self):
936
return [self.street, self.state]
938
class User(Base, fixtures.ComparableEntity):
939
__tablename__ = 'user'
940
id = Column(Integer, primary_key=True,
941
test_needs_autoincrement=True)
942
street = Column(String(50))
943
state = Column(String(2))
944
address = composite(AddressComposite,
947
Base.metadata.create_all()
950
address=AddressComposite('123 anywhere street',
955
sess.query(User).all(),
956
[User(address=AddressComposite('123 anywhere street',
960
def test_mapping_to_join(self):
961
users = Table('users', Base.metadata,
962
Column('id', Integer, primary_key=True)
964
addresses = Table('addresses', Base.metadata,
965
Column('id', Integer, primary_key=True),
966
Column('user_id', Integer, ForeignKey('users.id'))
968
usersaddresses = sa.join(users, addresses, users.c.id
969
== addresses.c.user_id)
971
__table__ = usersaddresses
972
__table_args__ = {'primary_key':[users.c.id]}
974
# need to use column_property for now
975
user_id = column_property(users.c.id, addresses.c.user_id)
976
address_id = addresses.c.id
978
assert User.__mapper__.get_property('user_id').columns[0] \
980
assert User.__mapper__.get_property('user_id').columns[1] \
981
is addresses.c.user_id
880
983
def test_synonym_inline(self):
882
class User(Base, ComparableEntity):
985
class User(Base, fixtures.ComparableEntity):
884
987
__tablename__ = 'users'
885
988
id = Column('id', Integer, primary_key=True,
1264
1367
any(Engineer.primary_language
1265
1368
== 'cobol')).first(), c2)
1267
# ensure that the Manager mapper was compiled with the Person id
1268
# column as higher priority. this ensures that "id" will get
1269
# loaded from the Person row and not the possibly non-present
1370
# ensure that the Manager mapper was compiled with the Manager id
1371
# column as higher priority. this ensures that "Manager.id"
1372
# is appropriately treated as the "id" column in the "manager"
1373
# table (reversed from 0.6's behavior.)
1272
assert Manager.id.property.columns == [Person.__table__.c.id,
1273
Manager.__table__.c.id]
1375
assert Manager.id.property.columns == [Manager.__table__.c.id, Person.__table__.c.id]
1275
1377
# assert that the "id" column is available without a second
1276
# load. this would be the symptom of the previous step not being
1378
# load. as of 0.7, the ColumnProperty tests all columns
1379
# in it's list to see which is present in the row.
1279
1381
sess.expunge_all()
1282
1384
assert sess.query(Manager).filter(Manager.name == 'dogbert'
1285
1386
self.assert_sql_count(testing.db, go, 1)
1286
1387
sess.expunge_all()
1473
1575
related_child1 = Column('c1', Integer)
1474
1576
__mapper_args__ = dict(polymorphic_identity='child2')
1476
sa.orm.compile_mappers() # no exceptions here
1578
sa.orm.configure_mappers() # no exceptions here
1580
def test_foreign_keys_with_col(self):
1581
"""Test that foreign keys that reference a literal 'id' subclass
1582
'id' attribute behave intuitively.
1588
class Booking(Base):
1589
__tablename__ = 'booking'
1590
id = Column(Integer, primary_key=True)
1592
class PlanBooking(Booking):
1593
__tablename__ = 'plan_booking'
1594
id = Column(Integer, ForeignKey(Booking.id),
1597
# referencing PlanBooking.id gives us the column
1598
# on plan_booking, not booking
1599
class FeatureBooking(Booking):
1600
__tablename__ = 'feature_booking'
1601
id = Column(Integer, ForeignKey(Booking.id),
1603
plan_booking_id = Column(Integer,
1604
ForeignKey(PlanBooking.id))
1606
plan_booking = relationship(PlanBooking,
1607
backref='feature_bookings')
1609
assert FeatureBooking.__table__.c.plan_booking_id.\
1610
references(PlanBooking.__table__.c.id)
1612
assert FeatureBooking.__table__.c.id.\
1613
references(Booking.__table__.c.id)
1478
1616
def test_single_colsonbase(self):
1479
1617
"""test single inheritance where all the columns are on the base
1482
class Company(Base, ComparableEntity):
1620
class Company(Base, fixtures.ComparableEntity):
1484
1622
__tablename__ = 'companies'
1485
1623
id = Column('id', Integer, primary_key=True,
1863
2001
assert_raises_message(sa.exc.ArgumentError,
1864
2002
'place __table_args__', go)
2004
@testing.emits_warning("The classname")
2005
def test_dupe_name_in_hierarchy(self):
2008
id = Column( Integer, primary_key=True)
2013
id = Column(Integer(),ForeignKey(a_1.id), primary_key = True)
2015
assert A.__mapper__.inherits is a_1.__mapper__
1866
2017
def test_concrete(self):
1867
2018
engineers = Table('engineers', Base.metadata, Column('id',
1868
2019
Integer, primary_key=True,
1951
2102
def _produce_test(inline, stringbased):
1953
class ExplicitJoinTest(MappedTest):
2104
class ExplicitJoinTest(fixtures.MappedTest):
1956
2107
def define_tables(cls, metadata):
1957
2108
global User, Address
1958
2109
Base = decl.declarative_base(metadata=metadata)
1960
class User(Base, ComparableEntity):
2111
class User(Base, fixtures.ComparableEntity):
1962
2113
__tablename__ = 'users'
1963
2114
id = Column(Integer, primary_key=True,
1964
2115
test_needs_autoincrement=True)
1965
2116
name = Column(String(50))
1967
class Address(Base, ComparableEntity):
2118
class Address(Base, fixtures.ComparableEntity):
1969
2120
__tablename__ = 'addresses'
1970
2121
id = Column(Integer, primary_key=True,
3023
3176
eq_(Model.__table__.c.keys(), ['col1', 'col3', 'col2', 'col4',
3179
def test_honor_class_mro_one(self):
3180
class HasXMixin(object):
3183
return Column(Integer)
3185
class Parent(HasXMixin, Base):
3186
__tablename__ = 'parent'
3187
id = Column(Integer, primary_key=True)
3189
class Child(Parent):
3190
__tablename__ = 'child'
3191
id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
3193
assert "x" not in Child.__table__.c
3195
def test_honor_class_mro_two(self):
3196
class HasXMixin(object):
3199
return Column(Integer)
3201
class Parent(HasXMixin, Base):
3202
__tablename__ = 'parent'
3203
id = Column(Integer, primary_key=True)
3209
id = Column(Integer, ForeignKey('parent.id'), primary_key=True)
3211
assert C().x() == 'hi'
3026
3214
class DeclarativeMixinPropertyTest(DeclarativeTestBase):
3028
3216
def test_column_property(self):