← 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: Jelmer Vernooij
  • Date: 2011-09-21 14:28:02 UTC
  • mfrom: (14006 devel)
  • mto: This revision was merged to the branch mainline in revision 14010.
  • Revision ID: jelmer@canonical.com-20110921142802-7ggkc204igsy532w
Merge lp:launchpad

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/canonical/database/ftests/test_isolation.py
12
12
from textwrap import dedent
13
13
import unittest
14
14
 
 
15
import transaction
 
16
 
15
17
from canonical.database.sqlbase import (
16
18
    cursor, ISOLATION_LEVEL_AUTOCOMMIT, ISOLATION_LEVEL_DEFAULT,
17
19
    ISOLATION_LEVEL_READ_COMMITTED, ISOLATION_LEVEL_SERIALIZABLE,
18
 
    connect)
 
20
    connect, ZopelessTransactionManager)
19
21
from canonical.testing.layers import LaunchpadZopelessLayer
20
22
 
21
23
 
 
24
def set_isolation_level(isolation):
 
25
    user = ZopelessTransactionManager._dbuser
 
26
    ZopelessTransactionManager.uninstall()
 
27
    ZopelessTransactionManager.initZopeless(dbuser=user, isolation=isolation)
 
28
 
 
29
 
22
30
class TestIsolation(unittest.TestCase):
23
31
    layer = LaunchpadZopelessLayer
24
32
 
25
 
    def setUp(self):
26
 
        self.txn = LaunchpadZopelessLayer.txn
27
 
 
28
33
    def getCurrentIsolation(self, con=None):
29
34
        if con is None:
30
35
            cur = cursor()
38
43
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
39
44
 
40
45
    def test_default2(self):
41
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_DEFAULT)
 
46
        set_isolation_level(ISOLATION_LEVEL_DEFAULT)
42
47
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
43
48
 
44
49
    def test_autocommit(self):
45
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
 
50
        set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
46
51
        # There is no actual 'autocommit' mode in PostgreSQL. psycopg
47
52
        # implements this feature by using read committed isolation and
48
53
        # issuing commit() statements after every query.
50
55
 
51
56
        # So we need to confirm we are actually in autocommit mode
52
57
        # by seeing if we an roll back
53
 
        con = self.txn.conn()
54
 
        cur = con.cursor()
 
58
        cur = cursor()
55
59
        cur.execute(
56
60
            "SELECT COUNT(*) FROM Person WHERE homepage_content IS NULL")
57
61
        self.failIfEqual(cur.fetchone()[0], 0)
58
62
        cur.execute("UPDATE Person SET homepage_content=NULL")
59
 
        con.rollback()
60
 
        cur = con.cursor()
 
63
        transaction.abort()
 
64
        cur = cursor()
61
65
        cur.execute(
62
66
            "SELECT COUNT(*) FROM Person WHERE homepage_content IS NOT NULL")
63
67
        self.failUnlessEqual(cur.fetchone()[0], 0)
64
68
 
65
69
    def test_readCommitted(self):
66
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
 
70
        set_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
67
71
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
68
72
 
69
73
    def test_serializable(self):
70
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
74
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
71
75
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
72
76
 
73
77
    def test_commit(self):
74
78
        # Change the isolation level
75
79
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
76
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
80
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
77
81
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
78
82
 
79
 
        con = self.txn.conn()
80
 
        cur = con.cursor()
 
83
        cur = cursor()
81
84
        cur.execute("UPDATE Person SET homepage_content=NULL")
82
 
        con.commit()
 
85
        transaction.commit()
83
86
        cur.execute("UPDATE Person SET homepage_content='foo'")
84
87
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
85
88
 
86
89
    def test_rollback(self):
87
90
        # Change the isolation level
88
91
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
89
 
        self.txn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
92
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
90
93
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
91
94
 
92
 
        con = self.txn.conn()
93
 
        cur = con.cursor()
 
95
        cur = cursor()
94
96
        cur.execute("UPDATE Person SET homepage_content=NULL")
95
 
        con.rollback()
 
97
        transaction.abort()
96
98
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
97
99
 
98
100
    def test_script(self):
109
111
                read committed
110
112
                serializable
111
113
                serializable
112
 
                serializable
113
 
                serializable
114
114
                """))
115
115
 
116
116
    def test_connect(self):