4
4
"""Provides a context manager to run parts of a test as a different dbuser."""
6
from __future__ import absolute_import
15
12
from contextlib import contextmanager
17
from storm.database import (
21
from storm.zope.interfaces import IZStorm
23
from zope.component import getUtility
25
from canonical.config import dbconfig
28
def update_store_connections():
29
"""Update the connection settings for all active stores.
31
This is required for connection setting changes to be made visible.
33
Unlike disconnect_stores and reconnect_stores, this changes the
34
underlying connection of *existing* stores, leaving existing objects
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()
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')
55
connection._raw_connection = None
56
connection._state = STATE_DISCONNECTED
60
def switch_dbuser(new_name):
61
"""Change the current database user.
63
If new_name is None, the default will be restored.
66
dbconfig.override(dbuser=new_name)
67
update_store_connections()
16
from canonical.database.sqlbase import ZopelessTransactionManager
17
from canonical.testing.layers import LaunchpadZopelessLayer
76
26
temporary_name is the name of the dbuser that should be in place for the
77
27
code in the "with" block.
79
old_name = getattr(dbconfig.overrides, 'dbuser', None)
80
switch_dbuser(temporary_name)
29
restore_name = ZopelessTransactionManager._dbuser
31
# Note that this will raise an assertion error if the
32
# LaunchpadZopelessLayer is not already set up.
33
LaunchpadZopelessLayer.switchDbUser(temporary_name)
82
switch_dbuser(old_name)
36
LaunchpadZopelessLayer.switchDbUser(restore_name)