← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/canonical/database/ftests/test_isolation.py

  • Committer: William Grant
  • Date: 2011-09-24 03:56:19 UTC
  • mto: This revision was merged to the branch mainline in revision 14030.
  • Revision ID: william.grant@canonical.com-20110924035619-ht8qu83op75xhdgr
LaunchpadScript no longer uses initZopeless.

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/canonical/database/ftests/test_isolation.py
14
14
 
15
15
import transaction
16
16
 
 
17
from canonical.config import dbconfig
17
18
from canonical.database.sqlbase import (
18
 
    cursor, ISOLATION_LEVEL_AUTOCOMMIT, ISOLATION_LEVEL_DEFAULT,
19
 
    ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_SERIALIZABLE,
20
 
    connect, ZopelessTransactionManager)
21
 
from canonical.testing.layers import LaunchpadZopelessLayer
 
19
    cursor,
 
20
    connect,
 
21
    ISOLATION_LEVEL_SERIALIZABLE,
 
22
    )
 
23
from canonical.testing.layers import (
 
24
    disconnect_stores,
 
25
    LaunchpadZopelessLayer,
 
26
    )
22
27
 
23
28
 
24
29
def set_isolation_level(isolation):
25
 
    user = ZopelessTransactionManager._dbuser
26
 
    ZopelessTransactionManager.uninstall()
27
 
    ZopelessTransactionManager.initZopeless(dbuser=user, isolation=isolation)
 
30
    dbconfig.override(isolation_level=isolation)
 
31
    disconnect_stores()
28
32
 
29
33
 
30
34
class TestIsolation(unittest.TestCase):
42
46
    def test_default(self):
43
47
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
44
48
 
45
 
    def test_default2(self):
46
 
        set_isolation_level(ISOLATION_LEVEL_DEFAULT)
47
 
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
48
 
 
49
49
    def test_autocommit(self):
50
 
        set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
 
50
        set_isolation_level('autocommit')
51
51
        # There is no actual 'autocommit' mode in PostgreSQL. psycopg
52
52
        # implements this feature by using read committed isolation and
53
53
        # issuing commit() statements after every query.
67
67
        self.failUnlessEqual(cur.fetchone()[0], 0)
68
68
 
69
69
    def test_readCommitted(self):
70
 
        set_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
 
70
        set_isolation_level('read_committed')
71
71
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
72
72
 
73
73
    def test_serializable(self):
74
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
74
        set_isolation_level('serializable')
75
75
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
76
76
 
77
77
    def test_commit(self):
78
78
        # Change the isolation level
79
79
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
80
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
80
        set_isolation_level('serializable')
81
81
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
82
82
 
83
83
        cur = cursor()
89
89
    def test_rollback(self):
90
90
        # Change the isolation level
91
91
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
92
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
92
        set_isolation_level('serializable')
93
93
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
94
94
 
95
95
        cur = cursor()