1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests confirming that changing isolation levels does what we expect."""
__metaclass__ = type
__all__ = []
import os.path
from subprocess import Popen, PIPE, STDOUT
import sys
from textwrap import dedent
import unittest
import transaction
from canonical.config import dbconfig
from canonical.database.sqlbase import (
cursor,
connect,
ISOLATION_LEVEL_SERIALIZABLE,
)
from canonical.testing.layers import (
disconnect_stores,
LaunchpadZopelessLayer,
)
def set_isolation_level(isolation):
dbconfig.override(isolation_level=isolation)
disconnect_stores()
class TestIsolation(unittest.TestCase):
layer = LaunchpadZopelessLayer
def getCurrentIsolation(self, con=None):
if con is None:
cur = cursor()
else:
cur = con.cursor()
cur.execute("SELECT * FROM Person")
cur.execute("SHOW transaction_isolation")
return cur.fetchone()[0]
def test_default(self):
self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
def test_autocommit(self):
set_isolation_level('autocommit')
# There is no actual 'autocommit' mode in PostgreSQL. psycopg
# implements this feature by using read committed isolation and
# issuing commit() statements after every query.
self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
# So we need to confirm we are actually in autocommit mode
# by seeing if we an roll back
cur = cursor()
cur.execute(
"SELECT COUNT(*) FROM Person WHERE homepage_content IS NULL")
self.failIfEqual(cur.fetchone()[0], 0)
cur.execute("UPDATE Person SET homepage_content=NULL")
transaction.abort()
cur = cursor()
cur.execute(
"SELECT COUNT(*) FROM Person WHERE homepage_content IS NOT NULL")
self.failUnlessEqual(cur.fetchone()[0], 0)
def test_readCommitted(self):
set_isolation_level('read_committed')
self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
def test_serializable(self):
set_isolation_level('serializable')
self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
def test_commit(self):
# Change the isolation level
self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
set_isolation_level('serializable')
self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
cur = cursor()
cur.execute("UPDATE Person SET homepage_content=NULL")
transaction.commit()
cur.execute("UPDATE Person SET homepage_content='foo'")
self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
def test_rollback(self):
# Change the isolation level
self.failUnlessEqual(self.getCurrentIsolation(), 'read committed')
set_isolation_level('serializable')
self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
cur = cursor()
cur.execute("UPDATE Person SET homepage_content=NULL")
transaction.abort()
self.failUnlessEqual(self.getCurrentIsolation(), 'serializable')
def test_script(self):
# Ensure that things work in stand alone scripts too, in case out
# test infrustructure is faking something.
script = os.path.join(
os.path.dirname(__file__), 'script_isolation.py')
cmd = [sys.executable, script]
process = Popen(cmd, stdout=PIPE, stderr=STDOUT, stdin=PIPE)
(script_output, _empty) = process.communicate()
self.failUnlessEqual(process.returncode, 0, 'Error: ' + script_output)
self.failUnlessEqual(script_output, dedent("""\
read committed
read committed
serializable
serializable
"""))
def test_connect(self):
# Ensure connect() method returns a connection with the correct
# default isolation
con = connect()
self.failUnlessEqual(self.getCurrentIsolation(con), 'read committed')
con.rollback()
self.failUnlessEqual(self.getCurrentIsolation(con), 'read committed')
# Ensure that changing the isolation sticks.
con = connect(isolation=ISOLATION_LEVEL_SERIALIZABLE)
self.failUnlessEqual(self.getCurrentIsolation(con), 'serializable')
con.rollback()
self.failUnlessEqual(self.getCurrentIsolation(con), 'serializable')
# But on a fresh connection, it works just fine.
con = connect()
con.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
self.failUnlessEqual(self.getCurrentIsolation(con), 'serializable')
con.rollback()
self.failUnlessEqual(self.getCurrentIsolation(con), 'serializable')
|