← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/testing/tests/test_fixture.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2011-09-05 16:05:38 UTC
  • mfrom: (13813.1.7 disco)
  • Revision ID: launchpad@pqm.canonical.com-20110905160538-kyybbc6zai8958z9
[r=wgrant, allenap,
 jtv][bug=836662] pgbouncer test fixture and db disconnection fix for
 branch-rewrite.py

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/testing/tests/test_fixture.py
8
8
from textwrap import dedent
9
9
 
10
10
from fixtures import EnvironmentVariableFixture
 
11
import psycopg2
 
12
from storm.exceptions import DisconnectionError
11
13
from zope.component import (
12
14
    adapts,
13
15
    queryAdapter,
17
19
    Interface,
18
20
    )
19
21
 
20
 
from canonical.testing.layers import BaseLayer
 
22
from canonical.config import dbconfig
 
23
from canonical.launchpad.interfaces.lpstorm import IMasterStore
 
24
from canonical.testing.layers import (
 
25
    BaseLayer,
 
26
    DatabaseLayer,
 
27
    LaunchpadZopelessLayer,
 
28
    )
 
29
from lp.registry.model.person import Person
21
30
from lp.testing import TestCase
22
31
from lp.testing.fixture import (
 
32
    PGBouncerFixture,
23
33
    RabbitServer,
24
34
    ZopeAdapterFixture,
25
35
    )
89
99
            self.assertIsInstance(adapter, FooToBar)
90
100
        # The adapter is no longer registered.
91
101
        self.assertIs(None, queryAdapter(context, IBar))
 
102
 
 
103
 
 
104
class TestPGBouncerFixtureWithCA(TestCase):
 
105
    """PGBouncerFixture reconnect tests for Component Architecture layers.
 
106
 
 
107
    Registered Storm Stores should be reconnected through pgbouncer.
 
108
    """
 
109
    layer = LaunchpadZopelessLayer
 
110
 
 
111
    def is_connected(self):
 
112
        # First rollback any existing transaction to ensure we attempt
 
113
        # to reconnect. We currently rollback the store explicitely
 
114
        # rather than call transaction.abort() due to Bug #819282.
 
115
        store = IMasterStore(Person)
 
116
        store.rollback()
 
117
 
 
118
        try:
 
119
            store.find(Person).first()
 
120
            return True
 
121
        except DisconnectionError:
 
122
            return False
 
123
 
 
124
    def test_stop_and_start(self):
 
125
        # Database is working.
 
126
        assert self.is_connected()
 
127
 
 
128
        # And database with the fixture is working too.
 
129
        pgbouncer = PGBouncerFixture()
 
130
        with PGBouncerFixture() as pgbouncer:
 
131
            assert self.is_connected()
 
132
 
 
133
            # pgbouncer is transparant. To confirm we are connecting via
 
134
            # pgbouncer, we need to shut it down and confirm our
 
135
            # connections are dropped.
 
136
            pgbouncer.stop()
 
137
            assert not self.is_connected()
 
138
 
 
139
            # If we restart it, things should be back to normal.
 
140
            pgbouncer.start()
 
141
            assert self.is_connected()
 
142
 
 
143
        # Database is still working.
 
144
        assert self.is_connected()
 
145
 
 
146
    def test_stop_no_start(self):
 
147
        # Database is working.
 
148
        assert self.is_connected()
 
149
 
 
150
        # And database with the fixture is working too.
 
151
        with PGBouncerFixture() as pgbouncer:
 
152
            assert self.is_connected()
 
153
 
 
154
            # pgbouncer is transparant. To confirm we are connecting via
 
155
            # pgbouncer, we need to shut it down and confirm our
 
156
            # connections are dropped.
 
157
            pgbouncer.stop()
 
158
            assert not self.is_connected()
 
159
 
 
160
        # Database is working again.
 
161
        assert self.is_connected()
 
162
 
 
163
 
 
164
class TestPGBouncerFixtureWithoutCA(TestCase):
 
165
    """PGBouncerFixture tests for non-Component Architecture layers."""
 
166
    layer = DatabaseLayer
 
167
 
 
168
    def is_db_available(self):
 
169
        # Direct connection to the DB.
 
170
        con_str = dbconfig.rw_main_master + ' user=launchpad_main'
 
171
        try:
 
172
            con = psycopg2.connect(con_str)
 
173
            cur = con.cursor()
 
174
            cur.execute("SELECT id FROM Person LIMIT 1")
 
175
            con.close()
 
176
            return True
 
177
        except psycopg2.OperationalError:
 
178
            return False
 
179
 
 
180
    def test_install_fixture(self):
 
181
        self.assert_(self.is_db_available())
 
182
 
 
183
        with PGBouncerFixture() as pgbouncer:
 
184
            self.assertTrue(self.is_db_available())
 
185
 
 
186
            pgbouncer.stop()
 
187
            self.assertFalse(self.is_db_available())
 
188
 
 
189
        # This confirms that we are again connecting directly to the
 
190
        # database, as the pgbouncer process was shutdown.
 
191
        self.assertTrue(self.is_db_available())
 
192
 
 
193
    def test_install_fixture_with_restart(self):
 
194
        self.assert_(self.is_db_available())
 
195
 
 
196
        with PGBouncerFixture() as pgbouncer:
 
197
            self.assertTrue(self.is_db_available())
 
198
 
 
199
            pgbouncer.stop()
 
200
            self.assertFalse(self.is_db_available())
 
201
 
 
202
            pgbouncer.start()
 
203
            self.assertTrue(self.is_db_available())
 
204
 
 
205
        # Note that because pgbouncer was left running, we can't confirm
 
206
        # that we are now connecting directly to the database.
 
207
        self.assertTrue(self.is_db_available())