← 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: Jeroen Vermeulen
  • Date: 2011-09-26 07:21:53 UTC
  • mfrom: (14031 devel)
  • mto: This revision was merged to the branch mainline in revision 14049.
  • Revision ID: jeroen.vermeulen@canonical.com-20110926072153-0lj7d2924rf5xpoc
Merge devel, resolve conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/canonical/database/ftests/test_isolation.py
1
 
# Copyright 2009 Canonical Ltd.  This software is licensed under the
 
1
# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
2
2
# GNU Affero General Public License version 3 (see the file LICENSE).
3
3
 
4
4
"""Tests confirming that changing isolation levels does what we expect."""
7
7
__all__ = []
8
8
 
9
9
import os.path
10
 
from subprocess import Popen, PIPE, STDOUT
 
10
from subprocess import (
 
11
    PIPE,
 
12
    Popen,
 
13
    STDOUT,
 
14
    )
11
15
import sys
12
16
from textwrap import dedent
13
17
import unittest
14
18
 
15
19
import transaction
16
20
 
 
21
from canonical.config import dbconfig
17
22
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
 
23
    connect,
 
24
    cursor,
 
25
    ISOLATION_LEVEL_SERIALIZABLE,
 
26
    )
 
27
from canonical.testing.layers import (
 
28
    disconnect_stores,
 
29
    LaunchpadZopelessLayer,
 
30
    )
22
31
 
23
32
 
24
33
def set_isolation_level(isolation):
25
 
    user = ZopelessTransactionManager._dbuser
26
 
    ZopelessTransactionManager.uninstall()
27
 
    ZopelessTransactionManager.initZopeless(dbuser=user, isolation=isolation)
 
34
    dbconfig.override(isolation_level=isolation)
 
35
    disconnect_stores()
28
36
 
29
37
 
30
38
class TestIsolation(unittest.TestCase):
42
50
    def test_default(self):
43
51
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
44
52
 
45
 
    def test_default2(self):
46
 
        set_isolation_level(ISOLATION_LEVEL_DEFAULT)
47
 
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
48
 
 
49
53
    def test_autocommit(self):
50
 
        set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
 
54
        set_isolation_level('autocommit')
51
55
        # There is no actual 'autocommit' mode in PostgreSQL. psycopg
52
56
        # implements this feature by using read committed isolation and
53
57
        # issuing commit() statements after every query.
67
71
        self.failUnlessEqual(cur.fetchone()[0], 0)
68
72
 
69
73
    def test_readCommitted(self):
70
 
        set_isolation_level(ISOLATION_LEVEL_READ_COMMITTED)
 
74
        set_isolation_level('read_committed')
71
75
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
72
76
 
73
77
    def test_serializable(self):
74
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
78
        set_isolation_level('serializable')
75
79
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
76
80
 
77
81
    def test_commit(self):
78
82
        # Change the isolation level
79
83
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
80
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
84
        set_isolation_level('serializable')
81
85
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
82
86
 
83
87
        cur = cursor()
89
93
    def test_rollback(self):
90
94
        # Change the isolation level
91
95
        self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
92
 
        set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
96
        set_isolation_level('serializable')
93
97
        self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
94
98
 
95
99
        cur = cursor()