diff options
Diffstat (limited to 'test/orm/expire.py')
| -rw-r--r-- | test/orm/expire.py | 162 |
1 files changed, 81 insertions, 81 deletions
diff --git a/test/orm/expire.py b/test/orm/expire.py index d54e9fc2d..98963c012 100644 --- a/test/orm/expire.py +++ b/test/orm/expire.py @@ -1,6 +1,6 @@ """test attribute/instance expiration, deferral of attributes, etc.""" -import testbase +import testenv; testenv.configure_for_tests() from sqlalchemy import * from sqlalchemy import exceptions from sqlalchemy.orm import * @@ -10,25 +10,25 @@ from testlib.fixtures import * class ExpireTest(FixtureTest): keep_mappers = False refresh_data = True - + def test_expire(self): mapper(User, users, properties={ 'addresses':relation(Address, backref='user'), }) mapper(Address, addresses) - + sess = create_session() u = sess.query(User).get(7) assert len(u.addresses) == 1 u.name = 'foo' del u.addresses[0] sess.expire(u) - + assert 'name' not in u.__dict__ - + def go(): assert u.name == 'jack' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert 'name' in u.__dict__ u.name = 'foo' @@ -48,46 +48,46 @@ class ExpireTest(FixtureTest): def go(): assert u.name == 'jack' - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_doesntload_on_set(self): mapper(User, users) - + sess = create_session() u = sess.query(User).get(7) - + sess.expire(u, attribute_names=['name']) def go(): u.name = 'somenewname' - self.assert_sql_count(testbase.db, go, 0) + self.assert_sql_count(testing.db, go, 0) sess.flush() sess.clear() assert sess.query(User).get(7).name == 'somenewname' - + def test_no_session(self): mapper(User, users) sess = create_session() u = sess.query(User).get(7) - + sess.expire(u, attribute_names=['name']) sess.expunge(u) try: u.name except exceptions.InvalidRequestError, e: assert str(e) == "Instance <class 'testlib.fixtures.User'> is not bound to a Session, and no contextual session is established; attribute refresh operation cannot proceed" - + def test_expire_preserves_changes(self): """test that the expire load operation doesn't revert post-expire changes""" - + mapper(Order, orders) sess = create_session() o = sess.query(Order).get(3) sess.expire(o) - + o.description = "order 3 modified" def go(): assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o.description == 'order 3 modified' del o.description @@ -98,14 +98,14 @@ class ExpireTest(FixtureTest): assert "description" not in o.__dict__ assert o.description is None - + o.isopen=15 sess.expire(o, ['isopen', 'description']) o.description = 'some new description' sess.query(Order).all() assert o.isopen == 1 assert o.description == 'some new description' - + if False: # NOTYET: need to implement unconditional population # of expired attriutes in mapper._instances() @@ -114,12 +114,12 @@ class ExpireTest(FixtureTest): del o.isopen def go(): assert o.isopen is None - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_committed(self): """test that the committed state of the attribute receives the most recent DB data""" mapper(Order, orders) - + sess = create_session() o = sess.query(Order).get(3) sess.expire(o) @@ -129,8 +129,8 @@ class ExpireTest(FixtureTest): assert o._state.dict['description'] == 'order 3 modified' def go(): sess.flush() - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def test_expire_cascade(self): mapper(User, users, properties={ 'addresses':relation(Address, cascade="all, refresh-expire") @@ -162,8 +162,8 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address == 'jack@bean.com' assert u.name == 'jack' - # two loads - self.assert_sql_count(testbase.db, go, 2) + # two loads + self.assert_sql_count(testing.db, go, 2) assert 'name' in u.__dict__ assert 'addresses' in u.__dict__ @@ -183,9 +183,9 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address == 'jack@bean.com' assert u.name == 'jack' - # two loads, since relation() + scalar are + # two loads, since relation() + scalar are # separate right now - self.assert_sql_count(testbase.db, go, 2) + self.assert_sql_count(testing.db, go, 2) assert 'name' in u.__dict__ assert 'addresses' in u.__dict__ @@ -198,35 +198,35 @@ class ExpireTest(FixtureTest): sess = create_session() o = sess.query(Order).get(3) - + sess.expire(o, attribute_names=['description']) assert 'id' in o.__dict__ assert 'description' not in o.__dict__ assert o._state.dict['isopen'] == 1 - + orders.update(orders.c.id==3).execute(description='order 3 modified') - + def go(): assert o.description == 'order 3 modified' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o._state.dict['description'] == 'order 3 modified' - + o.isopen = 5 sess.expire(o, attribute_names=['description']) assert 'id' in o.__dict__ assert 'description' not in o.__dict__ assert o.__dict__['isopen'] == 5 assert o._state.committed_state['isopen'] == 1 - + def go(): assert o.description == 'order 3 modified' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) assert o.__dict__['isopen'] == 5 assert o._state.dict['description'] == 'order 3 modified' assert o._state.committed_state['isopen'] == 1 sess.flush() - + sess.expire(o, attribute_names=['id', 'isopen', 'description']) assert 'id' not in o.__dict__ assert 'isopen' not in o.__dict__ @@ -235,7 +235,7 @@ class ExpireTest(FixtureTest): assert o.description == 'order 3 modified' assert o.id == 3 assert o.isopen == 5 - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_partial_expire_lazy(self): mapper(User, users, properties={ @@ -245,41 +245,41 @@ class ExpireTest(FixtureTest): sess = create_session() u = sess.query(User).get(8) - + sess.expire(u, ['name', 'addresses']) assert 'name' not in u.__dict__ assert 'addresses' not in u.__dict__ - + # hit the lazy loader. just does the lazy load, # doesnt do the overall refresh def go(): assert u.addresses[0].email_address=='ed@wood.com' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + assert 'name' not in u.__dict__ - - # check that mods to expired lazy-load attributes + + # check that mods to expired lazy-load attributes # only do the lazy load sess.expire(u, ['name', 'addresses']) def go(): u.addresses = [Address(id=10, email_address='foo@bar.com')] - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + sess.flush() - - # flush has occurred, and addresses was modified, + + # flush has occurred, and addresses was modified, # so the addresses collection got committed and is # longer expired def go(): assert u.addresses[0].email_address=='foo@bar.com' assert len(u.addresses) == 1 - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + # but the name attribute was never loaded and so # still loads def go(): assert u.name == 'ed' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_partial_expire_eager(self): mapper(User, users, properties={ @@ -296,14 +296,14 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address=='ed@wood.com' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) - # check that mods to expired eager-load attributes + # check that mods to expired eager-load attributes # do the refresh sess.expire(u, ['name', 'addresses']) def go(): u.addresses = [Address(id=10, email_address='foo@bar.com')] - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.flush() # this should ideally trigger the whole load @@ -311,34 +311,34 @@ class ExpireTest(FixtureTest): def go(): assert u.addresses[0].email_address=='foo@bar.com' assert len(u.addresses) == 1 - self.assert_sql_count(testbase.db, go, 0) - + self.assert_sql_count(testing.db, go, 0) + def go(): assert u.name == 'ed' # scalar attributes have their own load - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) # ideally, this was already loaded, but we arent # doing it that way right now - #self.assert_sql_count(testbase.db, go, 0) + #self.assert_sql_count(testing.db, go, 0) def test_partial_expire_deferred(self): mapper(Order, orders, properties={ 'description':deferred(orders.c.description) }) - + sess = create_session() o = sess.query(Order).get(3) sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ - + # test that expired attribute access refreshes # the deferred def go(): assert o.isopen == 1 assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ @@ -347,10 +347,10 @@ class ExpireTest(FixtureTest): def go(): assert o.description == 'order 3' assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + clear_mappers() - + mapper(Order, orders) sess.clear() @@ -362,20 +362,20 @@ class ExpireTest(FixtureTest): # sanity check def go(): assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + assert 'description' in o.__dict__ assert 'isopen' in o.__dict__ sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ assert 'description' not in o.__dict__ - + # test that expired attribute access refreshes # the deferred def go(): assert o.isopen == 1 assert o.description == 'order 3' - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.expire(o, ['description', 'isopen']) assert 'isopen' not in o.__dict__ @@ -385,8 +385,8 @@ class ExpireTest(FixtureTest): def go(): assert o.description == 'order 3' assert o.isopen == 1 - self.assert_sql_count(testbase.db, go, 1) - + self.assert_sql_count(testing.db, go, 1) + class RefreshTest(FixtureTest): keep_mappers = False @@ -412,7 +412,7 @@ class RefreshTest(FixtureTest): # username is back to the DB assert u.name == 'jack' - + assert id(a) not in [id(x) for x in u.addresses] u.name = 'foo' @@ -438,28 +438,28 @@ class RefreshTest(FixtureTest): assert 'name' not in u.__dict__ s.refresh(u) assert u.name == 'jack' - + def test_refresh_with_lazy(self): - """test that when a lazy loader is set as a trigger on an object's attribute - (at the attribute level, not the class level), a refresh() operation doesnt + """test that when a lazy loader is set as a trigger on an object's attribute + (at the attribute level, not the class level), a refresh() operation doesnt fire the lazy loader or create any problems""" - + s = create_session() mapper(User, users, properties={'addresses':relation(mapper(Address, addresses))}) q = s.query(User).options(lazyload('addresses')) u = q.filter(users.c.id==8).first() def go(): s.refresh(u) - self.assert_sql_count(testbase.db, go, 1) + self.assert_sql_count(testing.db, go, 1) def test_refresh_with_eager(self): """test that a refresh/expire operation loads rows properly and sends correct "isnew" state to eager loaders""" - + mapper(User, users, properties={ 'addresses':relation(mapper(Address, addresses), lazy=False) }) - + s = create_session() u = s.get(User, 8) assert len(u.addresses) == 3 @@ -480,12 +480,12 @@ class RefreshTest(FixtureTest): mapper(Address, addresses) mapper(User, users, properties = dict(addresses=relation(Address,cascade="all, delete-orphan",lazy=False)) ) - + u=User() u.name='Justin' a = Address(id=10, email_address='lala') u.addresses.append(a) - + s.save(u) s.flush() s.clear() @@ -497,4 +497,4 @@ class RefreshTest(FixtureTest): s.refresh(u) if __name__ == '__main__': - testbase.main() + testenv.main() |