~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/testing/dbuser.py

  • Committer: Julian Edwards
  • Date: 2011-09-21 11:24:53 UTC
  • mto: This revision was merged to the branch mainline in revision 14004.
  • Revision ID: julian.edwards@canonical.com-20110921112453-6132qxw0gl2q2306
remove unneeded interpreter in txlongpoll recipe

Show diffs side-by-side

added added

removed removed

Lines of Context:
3
3
 
4
4
"""Provides a context manager to run parts of a test as a different dbuser."""
5
5
 
6
 
from __future__ import absolute_import
7
 
 
8
6
__metaclass__ = type
9
7
__all__ = [
10
8
    'dbuser',
11
9
    'lp_dbuser',
12
 
    'switch_dbuser',
13
10
    ]
14
11
 
15
12
from contextlib import contextmanager
16
13
 
17
 
from storm.database import (
18
 
    STATE_CONNECTED,
19
 
    STATE_DISCONNECTED,
20
 
    )
21
 
from storm.zope.interfaces import IZStorm
22
14
import transaction
23
 
from zope.component import getUtility
24
 
 
25
 
from canonical.config import dbconfig
26
 
 
27
 
 
28
 
def update_store_connections():
29
 
    """Update the connection settings for all active stores.
30
 
 
31
 
    This is required for connection setting changes to be made visible.
32
 
 
33
 
    Unlike disconnect_stores and reconnect_stores, this changes the
34
 
    underlying connection of *existing* stores, leaving existing objects
35
 
    functional.
36
 
    """
37
 
    for name, store in getUtility(IZStorm).iterstores():
38
 
        connection = store._connection
39
 
        if connection._state == STATE_CONNECTED:
40
 
            if connection._raw_connection is not None:
41
 
                connection._raw_connection.close()
42
 
 
43
 
            # This method assumes that calling transaction.abort() will
44
 
            # call rollback() on the store, but this is no longer the
45
 
            # case as of jamesh's fix for bug 230977; Stores are not
46
 
            # registered with the transaction manager until they are
47
 
            # used. While storm doesn't provide an API which does what
48
 
            # we want, we'll go under the covers and emit the
49
 
            # register-transaction event ourselves. This method is
50
 
            # only called by the test suite to kill the existing
51
 
            # connections so the Store's reconnect with updated
52
 
            # connection settings.
53
 
            store._event.emit('register-transaction')
54
 
 
55
 
            connection._raw_connection = None
56
 
            connection._state = STATE_DISCONNECTED
57
 
    transaction.abort()
58
 
 
59
 
 
60
 
def switch_dbuser(new_name):
61
 
    """Change the current database user.
62
 
 
63
 
    If new_name is None, the default will be restored.
64
 
    """
65
 
    transaction.commit()
66
 
    dbconfig.override(dbuser=new_name)
67
 
    update_store_connections()
 
15
 
 
16
from canonical.database.sqlbase import ZopelessTransactionManager
 
17
from canonical.testing.layers import LaunchpadZopelessLayer
68
18
 
69
19
 
70
20
@contextmanager
76
26
    temporary_name is the name of the dbuser that should be in place for the
77
27
    code in the "with" block.
78
28
    """
79
 
    old_name = getattr(dbconfig.overrides, 'dbuser', None)
80
 
    switch_dbuser(temporary_name)
 
29
    restore_name = ZopelessTransactionManager._dbuser
 
30
    transaction.commit()
 
31
    # Note that this will raise an assertion error if the
 
32
    # LaunchpadZopelessLayer is not already set up.
 
33
    LaunchpadZopelessLayer.switchDbUser(temporary_name)
81
34
    yield
82
 
    switch_dbuser(old_name)
 
35
    transaction.commit()
 
36
    LaunchpadZopelessLayer.switchDbUser(restore_name)
83
37
 
84
38
 
85
39
def lp_dbuser():