← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/orm/test_update_delete.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/test_update_delete.py
1
 
from test.lib.testing import eq_, assert_raises, assert_raises_message
2
 
from test.lib import fixtures, testing
3
 
from sqlalchemy import Integer, String, ForeignKey, or_, and_, exc, select, func
4
 
from sqlalchemy.orm import mapper, relationship, backref, Session, joinedload
 
1
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
 
2
from sqlalchemy.testing import fixtures
 
3
from sqlalchemy import Integer, String, ForeignKey, or_, and_, exc, \
 
4
    select, func, Boolean, case
 
5
from sqlalchemy.orm import mapper, relationship, backref, Session, \
 
6
    joinedload, aliased
 
7
from sqlalchemy import testing
5
8
 
6
 
from test.lib.schema import Table, Column
 
9
from sqlalchemy.testing.schema import Table, Column
7
10
 
8
11
 
9
12
 
40
43
 
41
44
        mapper(User, users)
42
45
 
 
46
    def test_illegal_eval(self):
 
47
        User = self.classes.User
 
48
        s = Session()
 
49
        assert_raises_message(
 
50
            exc.ArgumentError,
 
51
            "Valid strategies for session synchronization "
 
52
            "are 'evaluate', 'fetch', False",
 
53
            s.query(User).update,
 
54
            {},
 
55
            synchronize_session="fake"
 
56
        )
 
57
 
43
58
    def test_illegal_operations(self):
44
59
        User = self.classes.User
45
60
 
63
78
                r"Can't call Query.delete\(\) when %s\(\) has been called" % mname,
64
79
                q.delete)
65
80
 
66
 
 
67
81
    def test_delete(self):
68
82
        User = self.classes.User
69
83
 
76
90
 
77
91
        eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
78
92
 
 
93
    def test_delete_against_metadata(self):
 
94
        User = self.classes.User
 
95
        users = self.tables.users
 
96
 
 
97
        sess = Session()
 
98
        sess.query(users).delete(synchronize_session=False)
 
99
        eq_(sess.query(User).count(), 0)
 
100
 
79
101
    def test_delete_with_bindparams(self):
80
102
        User = self.classes.User
81
103
 
183
205
        eq_([john.age, jack.age, jill.age, jane.age], [15,27,19,27])
184
206
        eq_(sess.query(User.age).order_by(User.id).all(), zip([15,27,19,27]))
185
207
 
 
208
    def test_update_against_metadata(self):
 
209
        User, users = self.classes.User, self.tables.users
 
210
 
 
211
        sess = Session()
 
212
 
 
213
        sess.query(users).update({users.c.age: 29}, synchronize_session=False)
 
214
        eq_(sess.query(User.age).order_by(User.id).all(), zip([29,29,29,29]))
 
215
 
186
216
    def test_update_with_bindparams(self):
187
217
        User = self.classes.User
188
218
 
196
226
        eq_([john.age, jack.age, jill.age, jane.age], [25,37,29,27])
197
227
        eq_(sess.query(User.age).order_by(User.id).all(), zip([25,37,29,27]))
198
228
 
 
229
    def test_update_without_load(self):
 
230
        User = self.classes.User
 
231
 
 
232
        sess = Session()
 
233
 
 
234
        sess.query(User).filter(User.id == 3).\
 
235
                update({'age': 44}, synchronize_session='fetch')
 
236
        eq_(sess.query(User.age).order_by(User.id).all(), zip([25,47,44,37]))
 
237
 
199
238
    def test_update_changes_resets_dirty(self):
200
239
        User = self.classes.User
201
240
 
411
450
                            synchronize_session='fetch')
412
451
        assert john not in sess
413
452
 
414
 
class UpdateDeleteRelatedTest(fixtures.MappedTest):
 
453
class UpdateDeleteIgnoresLoadersTest(fixtures.MappedTest):
415
454
    @classmethod
416
455
    def define_tables(cls, metadata):
417
456
        Table('users', metadata,
501
540
 
502
541
        eq_(sess.query(Document.title).all(), zip(['baz']))
503
542
 
 
543
class UpdateDeleteFromTest(fixtures.MappedTest):
 
544
    @classmethod
 
545
    def define_tables(cls, metadata):
 
546
        Table('users', metadata,
 
547
              Column('id', Integer, primary_key=True),
 
548
            )
 
549
        Table('documents', metadata,
 
550
              Column('id', Integer, primary_key=True),
 
551
              Column('user_id', None, ForeignKey('users.id')),
 
552
              Column('title', String(32)),
 
553
              Column('flag', Boolean)
 
554
        )
 
555
 
 
556
    @classmethod
 
557
    def setup_classes(cls):
 
558
        class User(cls.Comparable):
 
559
            pass
 
560
 
 
561
        class Document(cls.Comparable):
 
562
            pass
 
563
 
 
564
    @classmethod
 
565
    def insert_data(cls):
 
566
        users = cls.tables.users
 
567
 
 
568
        users.insert().execute([
 
569
            dict(id=1, ),
 
570
            dict(id=2, ),
 
571
            dict(id=3, ),
 
572
            dict(id=4, ),
 
573
        ])
 
574
 
 
575
        documents = cls.tables.documents
 
576
 
 
577
        documents.insert().execute([
 
578
            dict(id=1, user_id=1, title='foo'),
 
579
            dict(id=2, user_id=1, title='bar'),
 
580
            dict(id=3, user_id=2, title='baz'),
 
581
            dict(id=4, user_id=2, title='hoho'),
 
582
            dict(id=5, user_id=3, title='lala'),
 
583
            dict(id=6, user_id=3, title='bleh'),
 
584
        ])
 
585
 
 
586
    @classmethod
 
587
    def setup_mappers(cls):
 
588
        documents, Document, User, users = (cls.tables.documents,
 
589
                                cls.classes.Document,
 
590
                                cls.classes.User,
 
591
                                cls.tables.users)
 
592
 
 
593
        mapper(User, users)
 
594
        mapper(Document, documents, properties={
 
595
            'user': relationship(User, backref='documents')
 
596
        })
 
597
 
 
598
    @testing.requires.update_from
 
599
    def test_update_from_joined_subq_test(self):
 
600
        Document = self.classes.Document
 
601
        s = Session()
 
602
 
 
603
        subq = s.query(func.max(Document.title).label('title')).\
 
604
                group_by(Document.user_id).subquery()
 
605
 
 
606
        s.query(Document).filter(Document.title == subq.c.title).\
 
607
                update({'flag': True}, synchronize_session=False)
 
608
 
 
609
        eq_(
 
610
            set(s.query(Document.id, Document.flag)),
 
611
            set([
 
612
                    (1, True), (2, None),
 
613
                    (3, None), (4, True),
 
614
                    (5, True), (6, None),
 
615
                ])
 
616
        )
 
617
 
 
618
    @testing.requires.update_where_target_in_subquery
 
619
    def test_update_using_in(self):
 
620
        Document = self.classes.Document
 
621
        s = Session()
 
622
 
 
623
        subq = s.query(func.max(Document.title).label('title')).\
 
624
                group_by(Document.user_id).subquery()
 
625
 
 
626
        s.query(Document).filter(Document.title.in_(subq)).\
 
627
                update({'flag': True}, synchronize_session=False)
 
628
 
 
629
        eq_(
 
630
            set(s.query(Document.id, Document.flag)),
 
631
            set([
 
632
                    (1, True), (2, None),
 
633
                    (3, None), (4, True),
 
634
                    (5, True), (6, None),
 
635
                ])
 
636
        )
 
637
 
 
638
    @testing.requires.update_where_target_in_subquery
 
639
    @testing.requires.standalone_binds
 
640
    def test_update_using_case(self):
 
641
        Document = self.classes.Document
 
642
        s = Session()
 
643
 
 
644
 
 
645
        subq = s.query(func.max(Document.title).label('title')).\
 
646
                group_by(Document.user_id).subquery()
 
647
 
 
648
        # this would work with Firebird if you do literal_column('1')
 
649
        # instead
 
650
        case_stmt = case([(Document.title.in_(subq), True)], else_=False)
 
651
        s.query(Document).update({'flag': case_stmt}, synchronize_session=False)
 
652
 
 
653
        eq_(
 
654
            set(s.query(Document.id, Document.flag)),
 
655
            set([
 
656
                    (1, True), (2, False),
 
657
                    (3, False), (4, True),
 
658
                    (5, True), (6, False),
 
659
                ])
 
660
        )
 
661
 
504
662
class ExpressionUpdateTest(fixtures.MappedTest):
505
663
    @classmethod
506
664
    def define_tables(cls, metadata):
541
699
        eq_(d1.cnt, 2)
542
700
        sess.close()
543
701
 
544
 
 
 
702
class InheritTest(fixtures.DeclarativeMappedTest):
 
703
 
 
704
    run_inserts = 'each'
 
705
 
 
706
    run_deletes = 'each'
 
707
 
 
708
    @classmethod
 
709
    def setup_classes(cls):
 
710
        Base = cls.DeclarativeBasic
 
711
 
 
712
        class Person(Base):
 
713
            __tablename__ = 'person'
 
714
            id = Column(Integer, primary_key=True, test_needs_autoincrement=True)
 
715
            type = Column(String(50))
 
716
            name = Column(String(50))
 
717
 
 
718
        class Engineer(Person):
 
719
            __tablename__ = 'engineer'
 
720
            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
 
721
            engineer_name = Column(String(50))
 
722
 
 
723
        class Manager(Person):
 
724
            __tablename__ = 'manager'
 
725
            id = Column(Integer, ForeignKey('person.id'), primary_key=True)
 
726
            manager_name = Column(String(50))
 
727
 
 
728
    @classmethod
 
729
    def insert_data(cls):
 
730
        Engineer, Person, Manager = cls.classes.Engineer, \
 
731
                    cls.classes.Person, cls.classes.Manager
 
732
        s = Session(testing.db)
 
733
        s.add_all([
 
734
            Engineer(name='e1', engineer_name='e1'),
 
735
            Manager(name='m1', manager_name='m1'),
 
736
            Engineer(name='e2', engineer_name='e2'),
 
737
            Person(name='p1'),
 
738
        ])
 
739
        s.commit()
 
740
 
 
741
    def test_illegal_metadata(self):
 
742
        person = self.classes.Person.__table__
 
743
        engineer = self.classes.Engineer.__table__
 
744
 
 
745
        sess = Session()
 
746
        assert_raises_message(
 
747
            exc.InvalidRequestError,
 
748
            "This operation requires only one Table or entity be "
 
749
            "specified as the target.",
 
750
            sess.query(person.join(engineer)).update, {}
 
751
        )
 
752
 
 
753
    def test_update_subtable_only(self):
 
754
        Engineer = self.classes.Engineer
 
755
        s = Session(testing.db)
 
756
        s.query(Engineer).update({'engineer_name': 'e5'})
 
757
 
 
758
        eq_(
 
759
            s.query(Engineer.engineer_name).all(),
 
760
            [('e5', ), ('e5', )]
 
761
        )
 
762
 
 
763
    @testing.requires.update_from
 
764
    def test_update_from(self):
 
765
        Engineer = self.classes.Engineer
 
766
        Person = self.classes.Person
 
767
        s = Session(testing.db)
 
768
        s.query(Engineer).filter(Engineer.id == Person.id).\
 
769
            filter(Person.name == 'e2').update({'engineer_name': 'e5'})
 
770
 
 
771
        eq_(
 
772
            set(s.query(Person.name, Engineer.engineer_name)),
 
773
            set([('e1', 'e1', ), ('e2', 'e5')])
 
774
        )
 
775
 
 
776
    @testing.only_on('mysql', 'Multi table update')
 
777
    def test_update_from_multitable(self):
 
778
        Engineer = self.classes.Engineer
 
779
        Person = self.classes.Person
 
780
        s = Session(testing.db)
 
781
        s.query(Engineer).filter(Engineer.id == Person.id).\
 
782
            filter(Person.name == 'e2').update({Person.name: 'e22',
 
783
                                Engineer.engineer_name: 'e55'})
 
784
 
 
785
        eq_(
 
786
            set(s.query(Person.name, Engineer.engineer_name)),
 
787
            set([('e1', 'e1', ), ('e22', 'e55')])
 
788
        )