← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/orm/test_unitofworkv2.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_unitofworkv2.py
1
 
from test.lib.testing import eq_, assert_raises, assert_raises_message
2
 
from test.lib import testing, engines
3
 
from test.lib.schema import Table, Column
 
1
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
 
2
from sqlalchemy import testing
 
3
from sqlalchemy.testing import engines
 
4
from sqlalchemy.testing.schema import Table, Column
4
5
from test.orm import _fixtures
5
 
from test.lib import fixtures
 
6
from sqlalchemy.testing import fixtures
6
7
from sqlalchemy import Integer, String, ForeignKey, func
7
8
from sqlalchemy.orm import mapper, relationship, backref, \
8
9
                            create_session, unitofwork, attributes,\
9
10
                            Session, class_mapper, sync, exc as orm_exc
10
11
 
11
 
from test.lib.assertsql import AllOf, CompiledSQL
 
12
from sqlalchemy.testing.assertsql import AllOf, CompiledSQL
12
13
 
13
14
class AssertsUOW(object):
14
15
    def _get_test_uow(self, session):
1322
1323
            ),
1323
1324
        )
1324
1325
 
 
1326
class LoadersUsingCommittedTest(UOWTest):
 
1327
        """Test that events which occur within a flush()
 
1328
        get the same attribute loading behavior as on the outside
 
1329
        of the flush, and that the unit of work itself uses the
 
1330
        "committed" version of primary/foreign key attributes
 
1331
        when loading a collection for historical purposes (this typically
 
1332
        has importance for when primary key values change).
 
1333
 
 
1334
        """
 
1335
 
 
1336
        def _mapper_setup(self, passive_updates=True):
 
1337
            users, Address, addresses, User = (self.tables.users,
 
1338
                                    self.classes.Address,
 
1339
                                    self.tables.addresses,
 
1340
                                    self.classes.User)
 
1341
 
 
1342
            mapper(User, users, properties={
 
1343
                'addresses': relationship(Address,
 
1344
                    order_by=addresses.c.email_address,
 
1345
                    passive_updates=passive_updates,
 
1346
                    backref='user')
 
1347
            })
 
1348
            mapper(Address, addresses)
 
1349
            return create_session(autocommit=False)
 
1350
 
 
1351
        def test_before_update_m2o(self):
 
1352
            """Expect normal many to one attribute load behavior
 
1353
            (should not get committed value)
 
1354
            from within public 'before_update' event"""
 
1355
            sess = self._mapper_setup()
 
1356
 
 
1357
            Address, User = self.classes.Address, self.classes.User
 
1358
 
 
1359
            def before_update(mapper, connection, target):
 
1360
                # if get committed is used to find target.user, then
 
1361
                # it will be still be u1 instead of u2
 
1362
                assert target.user.id == target.user_id == u2.id
 
1363
            from sqlalchemy import event
 
1364
            event.listen(Address, 'before_update', before_update)
 
1365
 
 
1366
            a1 = Address(email_address='a1')
 
1367
            u1 = User(name='u1', addresses=[a1])
 
1368
            sess.add(u1)
 
1369
 
 
1370
            u2 = User(name='u2')
 
1371
            sess.add(u2)
 
1372
            sess.commit()
 
1373
 
 
1374
            sess.expunge_all()
 
1375
            # lookup an address and move it to the other user
 
1376
            a1 = sess.query(Address).get(a1.id)
 
1377
 
 
1378
            # move address to another user's fk
 
1379
            assert a1.user_id == u1.id
 
1380
            a1.user_id = u2.id
 
1381
 
 
1382
            sess.flush()
 
1383
 
 
1384
        def test_before_update_o2m_passive(self):
 
1385
            """Expect normal one to many attribute load behavior
 
1386
            (should not get committed value)
 
1387
            from within public 'before_update' event"""
 
1388
            self._test_before_update_o2m(True)
 
1389
 
 
1390
        def test_before_update_o2m_notpassive(self):
 
1391
            """Expect normal one to many attribute load behavior
 
1392
            (should not get committed value)
 
1393
            from within public 'before_update' event with
 
1394
            passive_updates=False
 
1395
 
 
1396
            """
 
1397
            self._test_before_update_o2m(False)
 
1398
 
 
1399
        def _test_before_update_o2m(self, passive_updates):
 
1400
            sess = self._mapper_setup(passive_updates=passive_updates)
 
1401
 
 
1402
            Address, User = self.classes.Address, self.classes.User
 
1403
 
 
1404
            class AvoidReferencialError(Exception):
 
1405
                """the test here would require ON UPDATE CASCADE on FKs
 
1406
                for the flush to fully succeed; this exception is used
 
1407
                to cancel the flush before we get that far.
 
1408
 
 
1409
                """
 
1410
 
 
1411
            def before_update(mapper, connection, target):
 
1412
                if passive_updates:
 
1413
                    # we shouldn't be using committed value.
 
1414
                    # so, having switched target's primary key,
 
1415
                    # we expect no related items in the collection
 
1416
                    # since we are using passive_updates
 
1417
                    # this is a behavior change since #2350
 
1418
                    assert 'addresses' not in target.__dict__
 
1419
                    eq_(target.addresses, [])
 
1420
                else:
 
1421
                    # in contrast with passive_updates=True,
 
1422
                    # here we expect the orm to have looked up the addresses
 
1423
                    # with the committed value (it needs to in order to
 
1424
                    # update the foreign keys).  So we expect addresses
 
1425
                    # collection to move with the user,
 
1426
                    # (just like they will be after the update)
 
1427
 
 
1428
                    # collection is already loaded
 
1429
                    assert 'addresses' in target.__dict__
 
1430
                    eq_([a.id for a in target.addresses],
 
1431
                            [a.id for a in [a1, a2]])
 
1432
                raise AvoidReferencialError()
 
1433
            from sqlalchemy import event
 
1434
            event.listen(User, 'before_update', before_update)
 
1435
 
 
1436
            a1 = Address(email_address='jack1')
 
1437
            a2 = Address(email_address='jack2')
 
1438
            u1 = User(id=1, name='jack', addresses=[a1, a2])
 
1439
            sess.add(u1)
 
1440
            sess.commit()
 
1441
 
 
1442
            sess.expunge_all()
 
1443
            u1 = sess.query(User).get(u1.id)
 
1444
            u1.id = 2
 
1445
            try:
 
1446
                sess.flush()
 
1447
            except AvoidReferencialError:
 
1448
                pass