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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
# Copyright 2009-2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
import os
from fixtures import (
EnvironmentVariableFixture,
TestWithFixtures,
)
import testtools
from canonical.config import dbconfig
from canonical.config.fixture import ConfigUseFixture
from canonical.testing.layers import BaseLayer
from lp.testing.pgsql import (
ConnectionWrapper,
PgTestSetup,
)
class TestPgTestSetup(testtools.TestCase, TestWithFixtures):
def assertDBName(self, expected_name, fixture):
"""Check that fixture uses expected_name as its dbname."""
self.assertEqual(expected_name, fixture.dbname)
fixture.setUp()
self.addCleanup(fixture.dropDb)
self.addCleanup(fixture.tearDown)
cur = fixture.connect().cursor()
cur.execute('SELECT current_database()')
where = cur.fetchone()[0]
self.assertEqual(expected_name, where)
def test_db_naming_LP_TEST_INSTANCE_set(self):
# when LP_TEST_INSTANCE is set, it is used for dynamic db naming.
BaseLayer.setUp()
self.addCleanup(BaseLayer.tearDown)
fixture = PgTestSetup(dbname=PgTestSetup.dynamic)
expected_name = "%s_%d" % (PgTestSetup.dbname, os.getpid())
self.assertDBName(expected_name, fixture)
def test_db_naming_without_LP_TEST_INSTANCE_is_static(self):
self.useFixture(EnvironmentVariableFixture('LP_TEST_INSTANCE'))
fixture = PgTestSetup(dbname=PgTestSetup.dynamic)
expected_name = PgTestSetup.dbname
self.assertDBName(expected_name, fixture)
def test_db_naming_stored_in_BaseLayer_configs(self):
BaseLayer.setUp()
self.addCleanup(BaseLayer.tearDown)
fixture = PgTestSetup(dbname=PgTestSetup.dynamic)
fixture.setUp()
self.addCleanup(fixture.dropDb)
self.addCleanup(fixture.tearDown)
expected_value = 'dbname=%s host=localhost' % fixture.dbname
self.assertEqual(expected_value, dbconfig.rw_main_master)
self.assertEqual(expected_value, dbconfig.rw_main_slave)
with ConfigUseFixture(BaseLayer.appserver_config_name):
self.assertEqual(expected_value, dbconfig.rw_main_master)
self.assertEqual(expected_value, dbconfig.rw_main_slave)
class TestPgTestSetupTuning(testtools.TestCase, TestWithFixtures):
layer = BaseLayer
def testOptimization(self):
# Test to ensure that the database is destroyed only when necessary
# Make a change to a database
fixture = PgTestSetup()
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('CREATE TABLE foo (x int)')
con.commit()
# Fake it so the harness doesn't know a change has been made
ConnectionWrapper.committed = False
finally:
fixture.tearDown()
# Now check to ensure that the table we just created is still there if
# we reuse the fixture.
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
# This tests that the table still exists, as well as modifying the
# db
cur.execute('INSERT INTO foo VALUES (1)')
con.commit()
finally:
fixture.tearDown()
# Now ensure that the table is gone - the commit must have been rolled
# back.
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('CREATE TABLE foo (x int)')
con.commit()
# Leave the table.
ConnectionWrapper.committed = False
finally:
fixture.tearDown()
# The database should *always* be recreated if a new template had been
# chosen.
PgTestSetup._last_db = ('different-template', fixture.dbname)
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
# If this fails, TABLE foo still existed and the DB wasn't rebuilt
# correctly.
cur.execute('CREATE TABLE foo (x int)')
con.commit()
finally:
fixture.tearDown()
def test_sequences(self):
# Sequences may be affected by connections even if the connection
# is rolled back. So ensure the database is reset fully, in the
# cases where we just rollback the changes we also need to reset all
# the sequences.
# Setup a table that uses a sequence
fixture = PgTestSetup()
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('CREATE TABLE foo (x serial, y integer)')
con.commit()
con.close()
# Fake it so the harness doesn't know a change has been made
ConnectionWrapper.committed = False
finally:
fixture.tearDown()
sequence_values = []
# Insert a row into it and roll back the changes. Each time, we
# should end up with the same sequence value
for i in range(3):
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('INSERT INTO foo (y) VALUES (1)')
cur.execute("SELECT currval('foo_x_seq')")
sequence_values.append(cur.fetchone()[0])
con.rollback()
con.close()
finally:
fixture.tearDown()
# Fail if we got a diffent sequence value at some point
for v in sequence_values:
self.failUnlessEqual(v, sequence_values[0])
# Repeat the test, but this time with some data already in the
# table
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('INSERT INTO foo (y) VALUES (1)')
con.commit()
con.close()
# Fake it so the harness doesn't know a change has been made
ConnectionWrapper.committed = False
finally:
fixture.tearDown()
sequence_values = []
# Insert a row into it and roll back the changes. Each time, we
# should end up with the same sequence value
for i in range(1, 3):
fixture.setUp()
try:
con = fixture.connect()
cur = con.cursor()
cur.execute('INSERT INTO foo (y) VALUES (1)')
cur.execute("SELECT currval('foo_x_seq')")
sequence_values.append(cur.fetchone()[0])
con.rollback()
con.close()
finally:
fixture.tearDown()
# Fail if we got a diffent sequence value at some point
for v in sequence_values:
self.failUnlessEqual(v, sequence_values[0])
|