tests/dbtest_session.py

   1 #!/usr/bin/env python
   2
   3 from db_test import DBDakTestCase
   4
   5 from daklib.dbconn import DBConn, Uid
   6
   7 from sqlalchemy.exc import InvalidRequestError
   8
   9 import time
  10 import unittest
  11
  12 class SessionTestCase(DBDakTestCase):
  13     """
  14     This TestCase checks the behaviour of SQLAlchemy's session object. It should
  15     make sure the SQLAlchemy always works as we expect it. And it might help
  16     dak beginners to get a grasp on how the session works.
  17     """
  18
  19     def sleep(self):
  20         time.sleep(0.001)
  21
  22     def test_timestamps(self):
  23         '''
  24         Test the basic transaction behaviour. The session is not configured for
  25         autocommit mode and that is why we always have an open transaction that
  26         ends with either rollback() or commit().
  27         '''
  28
  29         # timestamps will always be the same in one transaction
  30         timestamp01 = self.now()
  31         self.sleep()
  32         timestamp02 = self.now()
  33         self.assertEqual(timestamp01, timestamp02)
  34         uid = Uid(uid = 'foobar')
  35         self.session.add(uid)
  36         self.session.flush()
  37         self.assertEqual(timestamp01, uid.created)
  38         # ... but different in multiple transactions
  39         self.session.rollback()
  40         timestamp03 = self.now()
  41         self.assertTrue(timestamp01 < timestamp03)
  42         uid = Uid(uid = 'foobar')
  43         self.session.add(uid)
  44         self.session.flush()
  45         self.assertTrue(timestamp01 < uid.created)
  46
  47     def test_crud(self):
  48         '''
  49         Test INSERT, UPDATE, DELETE, ROLLBACK, and COMMIT behaviour of the
  50         session.
  51         '''
  52
  53         # test INSERT
  54         uid = Uid(uid = 'foobar')
  55         self.assertTrue(uid not in self.session)
  56         self.session.add(uid)
  57         self.assertTrue(uid in self.session)
  58         self.assertTrue(uid in self.session.new)
  59         self.session.flush()
  60         self.assertTrue(uid in self.session)
  61         self.assertTrue(uid not in self.session.new)
  62         # test UPDATE
  63         uid.uid = 'foobar2'
  64         self.assertTrue(uid in self.session.dirty)
  65         self.session.flush()
  66         self.assertTrue(uid not in self.session.dirty)
  67         # test ROLLBACK
  68         self.session.rollback()
  69         self.assertTrue(uid not in self.session)
  70         # test COMMIT
  71         uid = Uid(uid = 'foobar')
  72         self.session.add(uid)
  73         self.assertTrue(uid in self.session.new)
  74         self.session.commit()
  75         self.assertTrue(uid in self.session)
  76         self.assertTrue(uid not in self.session.new)
  77         # test DELETE
  78         self.session.delete(uid)
  79         self.assertTrue(uid in self.session)
  80         self.assertTrue(uid in self.session.deleted)
  81         self.session.flush()
  82         self.assertTrue(uid not in self.session)
  83         self.assertTrue(uid not in self.session.deleted)
  84
  85     def test_expunge(self):
  86         '''
  87         Test expunge() of objects from session and the object_session()
  88         function.
  89         '''
  90
  91         # test expunge()
  92         uid = Uid(uid = 'foobar')
  93         self.session.add(uid)
  94         self.assertTrue(uid in self.session)
  95         self.session.expunge(uid)
  96         self.assertTrue(uid not in self.session)
  97         # test close()
  98         self.session.add(uid)
  99         self.assertTrue(uid in self.session)
 100         self.session.close()
 101         self.assertTrue(uid not in self.session)
 102         # make uid persistent
 103         self.session.add(uid)
 104         self.session.commit()
 105         self.assertTrue(uid in self.session)
 106         # test rollback() for persistent object
 107         self.session.rollback()
 108         self.assertTrue(uid in self.session)
 109         # test expunge() for persistent object
 110         self.session.expunge(uid)
 111         self.assertTrue(uid not in self.session)
 112         # test close() for persistent object
 113         self.session.add(uid)
 114         self.assertTrue(uid in self.session)
 115         self.session.close()
 116         self.assertTrue(uid not in self.session)
 117
 118     def refresh(self):
 119         '''
 120         Refreshes self.uid and should raise an exception is self.uid is not
 121         persistent.
 122         '''
 123         self.session.refresh(self.uid)
 124
 125     def test_refresh(self):
 126         '''
 127         Test the refresh() of an object.
 128         '''
 129
 130         self.uid = Uid(uid = 'foobar')
 131         self.assertEqual(None, self.uid.uid_id)
 132         self.session.add(self.uid)
 133         self.assertEqual(None, self.uid.uid_id)
 134         self.session.flush()
 135         self.assertTrue(self.uid.uid_id is not None)
 136         self.session.rollback()
 137         self.assertRaises(InvalidRequestError, self.refresh)
 138
 139     def test_session(self):
 140         '''
 141         Tests the ORMObject.session() method.
 142         '''
 143
 144         uid = Uid(uid = 'foobar')
 145         self.session.add(uid)
 146         self.assertEqual(self.session, uid.session())
 147
 148     def test_clone(self):
 149         '''
 150         Tests the ORMObject.clone() method.
 151         '''
 152
 153         uid1 = Uid(uid = 'foobar')
 154         # no session yet
 155         self.assertRaises(RuntimeError, uid1.clone)
 156         self.session.add(uid1)
 157         # object not persistent yet
 158         self.assertRaises(RuntimeError, uid1.clone)
 159         self.session.commit()
 160         # test without session parameter
 161         uid2 = uid1.clone()
 162         self.assertTrue(uid1 is not uid2)
 163         self.assertEqual(uid1.uid, uid2.uid)
 164         self.assertTrue(uid2 not in uid1.session())
 165         self.assertTrue(uid1 not in uid2.session())
 166         # test with explicit session parameter
 167         new_session = DBConn().session()
 168         uid3 = uid1.clone(session = new_session)
 169         self.assertEqual(uid1.uid, uid3.uid)
 170         self.assertTrue(uid3 in new_session)
 171         # test for ressource leaks with mass cloning
 172         for _ in xrange(1, 1000):
 173             uid1.clone()
 174
 175     def classes_to_clean(self):
 176         # We need to clean all Uid objects in case some test fails.
 177         return (Uid,)
 178
 179 if __name__ == '__main__':
 180     unittest.main()