tests/dbtest_session.py

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