diff options
Diffstat (limited to 'test/orm/test_merge.py')
-rw-r--r-- | test/orm/test_merge.py | 99 |
1 files changed, 95 insertions, 4 deletions
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py index f69b07fe8..d4667d624 100644 --- a/test/orm/test_merge.py +++ b/test/orm/test_merge.py @@ -1,15 +1,15 @@ -from sqlalchemy.testing import assert_raises, assert_raises_message +from sqlalchemy.testing import assert_raises_message import sqlalchemy as sa -from sqlalchemy import Integer, PickleType, String, ForeignKey +from sqlalchemy import Integer, PickleType, String, ForeignKey, Text import operator from sqlalchemy import testing from sqlalchemy.util import OrderedSet from sqlalchemy.orm import mapper, relationship, create_session, \ PropComparator, synonym, comparable_property, sessionmaker, \ - attributes, Session, backref, configure_mappers, foreign + attributes, Session, backref, configure_mappers, foreign, deferred, defer from sqlalchemy.orm.collections import attribute_mapped_collection from sqlalchemy.orm.interfaces import MapperOption -from sqlalchemy.testing import eq_, ne_ +from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from test.orm import _fixtures from sqlalchemy import event, and_, case @@ -1378,6 +1378,97 @@ class M2ONoUseGetLoadingTest(fixtures.MappedTest): assert a2.user is u2 self.assert_sql_count(testing.db, go, 5) +class DeferredMergeTest(fixtures.MappedTest): + @classmethod + def define_tables(cls, metadata): + Table('book', metadata, + Column('id', Integer, primary_key=True), + Column('title', String(200), nullable=False), + Column('summary', String(2000)), + Column('excerpt', Text), + ) + + @classmethod + def setup_classes(cls): + class Book(cls.Basic): + pass + + def test_deferred_column_mapping(self): + # defer 'excerpt' at mapping level instead of query level + Book, book = self.classes.Book, self.tables.book + mapper(Book, book, properties={'excerpt': deferred(book.c.excerpt)}) + sess = sessionmaker()() + + b = Book( + title='Essential SQLAlchemy', + summary='some summary', + excerpt='some excerpt', + ) + sess.add(b) + sess.commit() + + b1 = sess.query(Book).first() + sess.expire(b1, ['summary']) + sess.close() + + def go(): + b2 = sess.merge(b1, load=False) + + # should not emit load for deferred 'excerpt' + eq_(b2.summary, 'some summary') + assert 'excerpt' not in b2.__dict__ + + # now it should emit load for deferred 'excerpt' + eq_(b2.excerpt, 'some excerpt') + assert 'excerpt' in b2.__dict__ + + self.sql_eq_(go, [ + ("SELECT book.summary AS book_summary " + "FROM book WHERE book.id = :param_1", + {u'param_1': 1}), + ("SELECT book.excerpt AS book_excerpt " + "FROM book WHERE book.id = :param_1", + {u'param_1': 1}) + ]) + + def test_deferred_column_query(self): + Book, book = self.classes.Book, self.tables.book + mapper(Book, book) + sess = sessionmaker()() + + b = Book( + title='Essential SQLAlchemy', + summary='some summary', + excerpt='some excerpt', + ) + sess.add(b) + sess.commit() + + # defer 'excerpt' at query level instead of mapping level + b1 = sess.query(Book).options(defer(Book.excerpt)).first() + sess.expire(b1, ['summary']) + sess.close() + + def go(): + b2 = sess.merge(b1, load=False) + + # should not emit load for deferred 'excerpt' + eq_(b2.summary, 'some summary') + assert 'excerpt' not in b2.__dict__ + + # now it should emit load for deferred 'excerpt' + eq_(b2.excerpt, 'some excerpt') + assert 'excerpt' in b2.__dict__ + + self.sql_eq_(go, [ + ("SELECT book.summary AS book_summary " + "FROM book WHERE book.id = :param_1", + {u'param_1': 1}), + ("SELECT book.excerpt AS book_excerpt " + "FROM book WHERE book.id = :param_1", + {u'param_1': 1}) + ]) + class MutableMergeTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): |