← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/orm/inheritance/test_basic.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/orm/inheritance/test_basic.py
1
1
import warnings
2
 
from test.lib.testing import eq_, assert_raises, assert_raises_message
 
2
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
3
3
from sqlalchemy import *
4
4
from sqlalchemy import exc as sa_exc, util, event
5
5
from sqlalchemy.orm import *
 
6
from sqlalchemy.orm.util import instance_str
6
7
from sqlalchemy.orm import exc as orm_exc, attributes
7
 
from test.lib.assertsql import AllOf, CompiledSQL
 
8
from sqlalchemy.testing.assertsql import AllOf, CompiledSQL
8
9
from sqlalchemy.sql import table, column
9
 
from test.lib import testing, engines
10
 
from test.lib import fixtures
 
10
from sqlalchemy import testing
 
11
from sqlalchemy.testing import engines
 
12
from sqlalchemy.testing import fixtures
11
13
from test.orm import _fixtures
12
 
from test.lib.schema import Table, Column
 
14
from sqlalchemy.testing.schema import Table, Column
 
15
from sqlalchemy import inspect
 
16
from sqlalchemy.ext.declarative import declarative_base
 
17
from sqlalchemy.testing.util import gc_collect
13
18
 
14
19
class O2MTest(fixtures.MappedTest):
15
20
    """deals with inheritance and one-to-many relationships"""
68
73
        l = sess.query(Blub).all()
69
74
        result = ','.join([repr(l[0]), repr(l[1]),
70
75
                          repr(l[0].parent_foo), repr(l[1].parent_foo)])
71
 
        print compare
72
 
        print result
73
 
        self.assert_(compare == result)
74
 
        self.assert_(l[0].parent_foo.data == 'foo #1'
75
 
                     and l[1].parent_foo.data == 'foo #1')
 
76
        eq_(compare, result)
 
77
        eq_(l[0].parent_foo.data, 'foo #1')
 
78
        eq_(l[1].parent_foo.data, 'foo #1')
 
79
 
 
80
class PolymorphicResolutionMultiLevel(fixtures.DeclarativeMappedTest,
 
81
                                        testing.AssertsCompiledSQL):
 
82
    run_setup_mappers = 'once'
 
83
    __dialect__ = 'default'
 
84
 
 
85
    @classmethod
 
86
    def setup_classes(cls):
 
87
        Base = cls.DeclarativeBasic
 
88
        class A(Base):
 
89
            __tablename__ = 'a'
 
90
            id = Column(Integer, primary_key=True)
 
91
        class B(A):
 
92
            __tablename__ = 'b'
 
93
            id = Column(Integer, ForeignKey('a.id'), primary_key=True)
 
94
        class C(A):
 
95
            __tablename__ = 'c'
 
96
            id = Column(Integer, ForeignKey('a.id'), primary_key=True)
 
97
        class D(B):
 
98
            __tablename__ = 'd'
 
99
            id = Column(Integer, ForeignKey('b.id'), primary_key=True)
 
100
 
 
101
    def test_ordered_b_d(self):
 
102
        a_mapper = inspect(self.classes.A)
 
103
        eq_(
 
104
            a_mapper._mappers_from_spec(
 
105
                    [self.classes.B, self.classes.D], None),
 
106
            [a_mapper, inspect(self.classes.B), inspect(self.classes.D)]
 
107
        )
 
108
 
 
109
    def test_a(self):
 
110
        a_mapper = inspect(self.classes.A)
 
111
        eq_(
 
112
            a_mapper._mappers_from_spec(
 
113
                    [self.classes.A], None),
 
114
            [a_mapper]
 
115
        )
 
116
 
 
117
    def test_b_d_selectable(self):
 
118
        a_mapper = inspect(self.classes.A)
 
119
        spec = [self.classes.D, self.classes.B]
 
120
        eq_(
 
121
            a_mapper._mappers_from_spec(
 
122
                    spec,
 
123
                    self.classes.B.__table__.join(self.classes.D.__table__)
 
124
            ),
 
125
            [inspect(self.classes.B), inspect(self.classes.D)]
 
126
        )
 
127
 
 
128
    def test_d_selectable(self):
 
129
        a_mapper = inspect(self.classes.A)
 
130
        spec = [self.classes.D]
 
131
        eq_(
 
132
            a_mapper._mappers_from_spec(
 
133
                    spec,
 
134
                    self.classes.B.__table__.join(self.classes.D.__table__)
 
135
            ),
 
136
            [inspect(self.classes.D)]
 
137
        )
 
138
 
 
139
    def test_reverse_d_b(self):
 
140
        a_mapper = inspect(self.classes.A)
 
141
        spec = [self.classes.D, self.classes.B]
 
142
        eq_(
 
143
            a_mapper._mappers_from_spec(
 
144
                    spec, None),
 
145
            [a_mapper, inspect(self.classes.B), inspect(self.classes.D)]
 
146
        )
 
147
        mappers, selectable = a_mapper._with_polymorphic_args(spec=spec)
 
148
        self.assert_compile(selectable,
 
149
            "a LEFT OUTER JOIN b ON a.id = b.id "
 
150
            "LEFT OUTER JOIN d ON b.id = d.id")
 
151
 
 
152
    def test_d_b_missing(self):
 
153
        a_mapper = inspect(self.classes.A)
 
154
        spec = [self.classes.D]
 
155
        eq_(
 
156
            a_mapper._mappers_from_spec(
 
157
                    spec, None),
 
158
            [a_mapper, inspect(self.classes.B), inspect(self.classes.D)]
 
159
        )
 
160
        mappers, selectable = a_mapper._with_polymorphic_args(spec=spec)
 
161
        self.assert_compile(selectable,
 
162
            "a LEFT OUTER JOIN b ON a.id = b.id "
 
163
            "LEFT OUTER JOIN d ON b.id = d.id")
 
164
 
 
165
    def test_d_c_b(self):
 
166
        a_mapper = inspect(self.classes.A)
 
167
        spec = [self.classes.D, self.classes.C, self.classes.B]
 
168
        ms = a_mapper._mappers_from_spec(spec, None)
 
169
 
 
170
        eq_(
 
171
            ms[-1], inspect(self.classes.D)
 
172
        )
 
173
        eq_(ms[0], a_mapper)
 
174
        eq_(
 
175
            set(ms[1:3]), set(a_mapper._inheriting_mappers)
 
176
        )
76
177
 
77
178
class PolymorphicOnNotLocalTest(fixtures.MappedTest):
78
179
    @classmethod
325
426
            [Child]
326
427
        )
327
428
 
 
429
class SortOnlyOnImportantFKsTest(fixtures.MappedTest):
 
430
    @classmethod
 
431
    def define_tables(cls, metadata):
 
432
        Table('a', metadata,
 
433
                Column('id', Integer, primary_key=True,
 
434
                                    test_needs_autoincrement=True),
 
435
                Column('b_id', Integer,
 
436
                        ForeignKey('b.id', use_alter=True, name='b'))
 
437
            )
 
438
        Table('b', metadata,
 
439
            Column('id', Integer, ForeignKey('a.id'), primary_key=True)
 
440
            )
 
441
 
 
442
    @classmethod
 
443
    def setup_classes(cls):
 
444
        Base = declarative_base()
 
445
 
 
446
        class A(Base):
 
447
            __tablename__ = "a"
 
448
 
 
449
            id = Column(Integer, primary_key=True,
 
450
                                    test_needs_autoincrement=True)
 
451
            b_id = Column(Integer, ForeignKey('b.id'))
 
452
 
 
453
        class B(A):
 
454
            __tablename__ = "b"
 
455
 
 
456
            id = Column(Integer, ForeignKey('a.id'), primary_key=True)
 
457
 
 
458
            __mapper_args__ = {'inherit_condition': id == A.id}
 
459
 
 
460
        cls.classes.A = A
 
461
        cls.classes.B = B
 
462
 
 
463
    def test_flush(self):
 
464
        s = Session(testing.db)
 
465
        s.add(self.classes.B())
 
466
        s.flush()
328
467
 
329
468
class FalseDiscriminatorTest(fixtures.MappedTest):
330
469
    @classmethod
335
474
            Column('type', Boolean, nullable=False))
336
475
 
337
476
    def test_false_on_sub(self):
338
 
        class Foo(object):pass
339
 
        class Bar(Foo):pass
 
477
        class Foo(object):
 
478
            pass
 
479
        class Bar(Foo):
 
480
            pass
340
481
        mapper(Foo, t1, polymorphic_on=t1.c.type, polymorphic_identity=True)
341
482
        mapper(Bar, inherits=Foo, polymorphic_identity=False)
342
483
        sess = create_session()
414
555
        Table('table_b', metadata,
415
556
           Column('id', Integer, ForeignKey('table_a.id'),
416
557
                                primary_key=True),
417
 
           Column('class_name', String(50))
 
558
           Column('class_name', String(50)),
418
559
        )
419
560
        Table('table_c', metadata,
420
561
           Column('id', Integer, ForeignKey('table_b.id'),
421
 
                                primary_key=True)
 
562
                                primary_key=True),
 
563
           Column('data', String(10))
422
564
        )
423
565
 
424
566
    @classmethod
433
575
            pass
434
576
        class C(B):
435
577
            pass
 
578
        class D(B):
 
579
            pass
436
580
 
437
581
        mapper(A, table_a,
438
582
                        polymorphic_on=table_a.c.class_name,
442
586
                        polymorphic_identity='b')
443
587
        mapper(C, table_c, inherits=B,
444
588
                        polymorphic_identity='c')
 
589
        mapper(D, inherits=B,
 
590
                        polymorphic_identity='d')
445
591
 
446
592
    def test_poly_configured_immediate(self):
447
593
        A, C, B = (self.classes.A,
471
617
 
472
618
        assert isinstance(sess.query(A).first(), C)
473
619
 
474
 
    def test_assignment(self):
475
 
        C, B = self.classes.C, self.classes.B
 
620
    def test_valid_assignment_upwards(self):
 
621
        """test that we can assign 'd' to a B, since B/D
 
622
        both involve the same set of tables.
 
623
        """
 
624
        D, B = self.classes.D, self.classes.B
 
625
 
 
626
        sess = Session()
 
627
        b1 = B()
 
628
        b1.class_name = 'd'
 
629
        sess.add(b1)
 
630
        sess.commit()
 
631
        sess.close()
 
632
        assert isinstance(sess.query(B).first(), D)
 
633
 
 
634
    def test_invalid_assignment_downwards(self):
 
635
        """test that we warn on assign of 'b' to a C, since this adds
 
636
        a row to the C table we'd never load.
 
637
        """
 
638
        C = self.classes.C
 
639
 
 
640
        sess = Session()
 
641
        c1 = C()
 
642
        c1.class_name = 'b'
 
643
        sess.add(c1)
 
644
        assert_raises_message(
 
645
            sa_exc.SAWarning,
 
646
            "Flushing object %s with incompatible "
 
647
            "polymorphic identity 'b'; the object may not "
 
648
            "refresh and/or load correctly" % instance_str(c1),
 
649
            sess.flush
 
650
        )
 
651
 
 
652
    def test_invalid_assignment_upwards(self):
 
653
        """test that we warn on assign of 'c' to a B, since we will have a
 
654
        "C" row that has no joined row, which will cause object
 
655
        deleted errors.
 
656
        """
 
657
        B = self.classes.B
476
658
 
477
659
        sess = Session()
478
660
        b1 = B()
479
661
        b1.class_name = 'c'
480
662
        sess.add(b1)
481
 
        sess.commit()
482
 
        sess.close()
483
 
        assert isinstance(sess.query(B).first(), C)
 
663
        assert_raises_message(
 
664
            sa_exc.SAWarning,
 
665
            "Flushing object %s with incompatible "
 
666
            "polymorphic identity 'c'; the object may not "
 
667
            "refresh and/or load correctly" % instance_str(b1),
 
668
            sess.flush
 
669
        )
 
670
 
 
671
    def test_entirely_oob_assignment(self):
 
672
        """test warn on an unknown polymorphic identity.
 
673
        """
 
674
        B = self.classes.B
 
675
 
 
676
        sess = Session()
 
677
        b1 = B()
 
678
        b1.class_name = 'xyz'
 
679
        sess.add(b1)
 
680
        assert_raises_message(
 
681
            sa_exc.SAWarning,
 
682
            "Flushing object %s with incompatible "
 
683
            "polymorphic identity 'xyz'; the object may not "
 
684
            "refresh and/or load correctly" % instance_str(b1),
 
685
            sess.flush
 
686
        )
 
687
 
 
688
    def test_not_set_on_upate(self):
 
689
        C = self.classes.C
 
690
 
 
691
        sess = Session()
 
692
        c1 = C()
 
693
        sess.add(c1)
 
694
        sess.commit()
 
695
        sess.expire(c1)
 
696
 
 
697
        c1.data = 'foo'
 
698
        sess.flush()
 
699
 
 
700
    def test_validate_on_upate(self):
 
701
        C = self.classes.C
 
702
 
 
703
        sess = Session()
 
704
        c1 = C()
 
705
        sess.add(c1)
 
706
        sess.commit()
 
707
        sess.expire(c1)
 
708
 
 
709
        c1.class_name = 'b'
 
710
        assert_raises_message(
 
711
            sa_exc.SAWarning,
 
712
            "Flushing object %s with incompatible "
 
713
            "polymorphic identity 'b'; the object may not "
 
714
            "refresh and/or load correctly" % instance_str(c1),
 
715
            sess.flush
 
716
        )
484
717
 
485
718
class CascadeTest(fixtures.MappedTest):
486
719
    """that cascades on polymorphic relationships continue
915
1148
        sess.flush()
916
1149
        assert user_roles.count().scalar() == 1
917
1150
 
 
1151
class JoinedNoFKSortingTest(fixtures.MappedTest):
 
1152
    @classmethod
 
1153
    def define_tables(cls, metadata):
 
1154
        Table("a", metadata,
 
1155
                Column('id', Integer, primary_key=True,
 
1156
                    test_needs_autoincrement=True)
 
1157
            )
 
1158
        Table("b", metadata,
 
1159
                Column('id', Integer, primary_key=True)
 
1160
            )
 
1161
        Table("c", metadata,
 
1162
                Column('id', Integer, primary_key=True)
 
1163
            )
 
1164
 
 
1165
    @classmethod
 
1166
    def setup_classes(cls):
 
1167
        class A(cls.Basic):
 
1168
            pass
 
1169
        class B(A):
 
1170
            pass
 
1171
        class C(A):
 
1172
            pass
 
1173
 
 
1174
    @classmethod
 
1175
    def setup_mappers(cls):
 
1176
        A, B, C = cls.classes.A, cls.classes.B, cls.classes.C
 
1177
        mapper(A, cls.tables.a)
 
1178
        mapper(B, cls.tables.b, inherits=A,
 
1179
                    inherit_condition=cls.tables.a.c.id == cls.tables.b.c.id)
 
1180
        mapper(C, cls.tables.c, inherits=A,
 
1181
                    inherit_condition=cls.tables.a.c.id == cls.tables.c.c.id)
 
1182
 
 
1183
    def test_ordering(self):
 
1184
        B, C = self.classes.B, self.classes.C
 
1185
        sess = Session()
 
1186
        sess.add_all([B(), C(), B(), C()])
 
1187
        self.assert_sql_execution(
 
1188
                testing.db,
 
1189
                sess.flush,
 
1190
                CompiledSQL(
 
1191
                    "INSERT INTO a () VALUES ()",
 
1192
                    {}
 
1193
                ),
 
1194
                CompiledSQL(
 
1195
                    "INSERT INTO a () VALUES ()",
 
1196
                    {}
 
1197
                ),
 
1198
                CompiledSQL(
 
1199
                    "INSERT INTO a () VALUES ()",
 
1200
                    {}
 
1201
                ),
 
1202
                CompiledSQL(
 
1203
                    "INSERT INTO a () VALUES ()",
 
1204
                    {}
 
1205
                ),
 
1206
                AllOf(
 
1207
                    CompiledSQL(
 
1208
                        "INSERT INTO b (id) VALUES (:id)",
 
1209
                        [{"id": 1}, {"id": 3}]
 
1210
                    ),
 
1211
                    CompiledSQL(
 
1212
                        "INSERT INTO c (id) VALUES (:id)",
 
1213
                        [{"id": 2}, {"id": 4}]
 
1214
                    )
 
1215
                )
 
1216
        )
 
1217
 
918
1218
class VersioningTest(fixtures.MappedTest):
919
1219
    @classmethod
920
1220
    def define_tables(cls, metadata):
1430
1730
            Column('b', String(10))
1431
1731
        )
1432
1732
 
 
1733
    def test_no_optimize_on_map_to_join(self):
 
1734
        base, sub = self.tables.base, self.tables.sub
 
1735
 
 
1736
        class Base(fixtures.ComparableEntity):
 
1737
            pass
 
1738
 
 
1739
        class JoinBase(fixtures.ComparableEntity):
 
1740
            pass
 
1741
        class SubJoinBase(JoinBase):
 
1742
            pass
 
1743
 
 
1744
        mapper(Base, base)
 
1745
        mapper(JoinBase, base.outerjoin(sub), properties=util.OrderedDict(
 
1746
                [('id', [base.c.id, sub.c.id]),
 
1747
                ('counter', [base.c.counter, sub.c.counter])])
 
1748
            )
 
1749
        mapper(SubJoinBase, inherits=JoinBase)
 
1750
 
 
1751
        sess = Session()
 
1752
        sess.add(Base(data='data'))
 
1753
        sess.commit()
 
1754
 
 
1755
        sjb = sess.query(SubJoinBase).one()
 
1756
        sjb_id = sjb.id
 
1757
        sess.expire(sjb)
 
1758
 
 
1759
        # this should not use the optimized load,
 
1760
        # which assumes discrete tables
 
1761
        def go():
 
1762
            eq_(sjb.data, 'data')
 
1763
 
 
1764
        self.assert_sql_execution(
 
1765
            testing.db,
 
1766
            go,
 
1767
            CompiledSQL(
 
1768
                "SELECT base.id AS base_id, sub.id AS sub_id, "
 
1769
                "base.counter AS base_counter, sub.counter AS sub_counter, "
 
1770
                "base.data AS base_data, "
 
1771
                "base.type AS base_type, sub.sub AS sub_sub, "
 
1772
                "sub.counter2 AS sub_counter2 FROM base "
 
1773
                "LEFT OUTER JOIN sub ON base.id = sub.id "
 
1774
                "WHERE base.id = :param_1",
 
1775
                {'param_1': sjb_id}
 
1776
            ),
 
1777
        )
 
1778
 
 
1779
 
1433
1780
    def test_optimized_passes(self):
1434
1781
        """"test that the 'optimized load' routine doesn't crash when
1435
1782
        a column in the join condition is not available."""
1471
1818
            pass
1472
1819
        mapper(Base, base, polymorphic_on=base.c.type, polymorphic_identity='base')
1473
1820
        mapper(Sub, sub, inherits=Base, polymorphic_identity='sub', properties={
1474
 
            'concat':column_property(sub.c.sub + "|" + sub.c.sub)
 
1821
            'concat': column_property(sub.c.sub + "|" + sub.c.sub)
1475
1822
        })
1476
1823
        sess = sessionmaker()()
1477
1824
        s1 = Sub(data='s1data', sub='s1sub')
1490
1837
            pass
1491
1838
        mapper(Base, base, polymorphic_on=base.c.type, polymorphic_identity='base')
1492
1839
        mapper(Sub, sub, inherits=Base, polymorphic_identity='sub', properties={
1493
 
            'concat':column_property(base.c.data + "|" + sub.c.sub)
 
1840
            'concat': column_property(base.c.data + "|" + sub.c.sub)
1494
1841
        })
1495
1842
        sess = sessionmaker()()
1496
1843
        s1 = Sub(data='s1data', sub='s1sub')
1603
1950
                                ['counter2']) is None
1604
1951
 
1605
1952
        s1.id = 1
1606
 
        attributes.instance_state(s1).commit_all(s1.__dict__, None)
 
1953
        attributes.instance_state(s1)._commit_all(s1.__dict__, None)
1607
1954
        assert m._optimized_get_statement(attributes.instance_state(s1),
1608
1955
                                ['counter2']) is not None
1609
1956
 
1660
2007
            ),
1661
2008
        )
1662
2009
 
 
2010
class TransientInheritingGCTest(fixtures.TestBase):
 
2011
    __requires__ = ('cpython',)
 
2012
 
 
2013
    def _fixture(self):
 
2014
        Base = declarative_base()
 
2015
 
 
2016
        class A(Base):
 
2017
            __tablename__ = 'a'
 
2018
            id = Column(Integer, primary_key=True,
 
2019
                                    test_needs_autoincrement=True)
 
2020
            data = Column(String(10))
 
2021
        self.A = A
 
2022
        return Base
 
2023
 
 
2024
    def setUp(self):
 
2025
        self.Base = self._fixture()
 
2026
 
 
2027
    def tearDown(self):
 
2028
        self.Base.metadata.drop_all(testing.db)
 
2029
        #clear_mappers()
 
2030
        self.Base = None
 
2031
 
 
2032
    def _do_test(self, go):
 
2033
        B = go()
 
2034
        self.Base.metadata.create_all(testing.db)
 
2035
        sess = Session(testing.db)
 
2036
        sess.add(B(data='some b'))
 
2037
        sess.commit()
 
2038
 
 
2039
        b1 = sess.query(B).one()
 
2040
        assert isinstance(b1, B)
 
2041
        sess.close()
 
2042
        del sess
 
2043
        del b1
 
2044
        del B
 
2045
 
 
2046
        gc_collect()
 
2047
 
 
2048
        eq_(
 
2049
            len(self.A.__subclasses__()),
 
2050
            0)
 
2051
 
 
2052
    def test_single(self):
 
2053
        def go():
 
2054
            class B(self.A):
 
2055
                pass
 
2056
            return B
 
2057
        self._do_test(go)
 
2058
 
 
2059
    @testing.fails_if(lambda: True,
 
2060
                "not supported for joined inh right now.")
 
2061
    def test_joined(self):
 
2062
        def go():
 
2063
            class B(self.A):
 
2064
                __tablename__ = 'b'
 
2065
                id = Column(Integer, ForeignKey('a.id'),
 
2066
                        primary_key=True)
 
2067
            return B
 
2068
        self._do_test(go)
 
2069
 
1663
2070
class NoPKOnSubTableWarningTest(fixtures.TestBase):
1664
2071
 
1665
2072
    def _fixture(self):