← Back to branch summary

~ubuntu-branches/ubuntu/vivid/ceilometer/vivid

« back to all changes in this revision

Viewing changes to ceilometer/openstack/common/db/sqlalchemy/test_migrations.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2014-03-06 14:44:28 UTC
  • mto: (28.1.1 utopic-proposed) (1.2.1)
  • mto: This revision was merged to the branch mainline in revision 19.
  • Revision ID: package-import@ubuntu.com-20140306144428-rvphsh4igwyulzf0
Tags: upstream-2014.1~b3
Import upstream version 2014.1~b3

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/openstack/common/db/sqlalchemy/test_migrations.py
15
15
#    under the License.
16
16
 
17
17
import functools
 
18
import logging
18
19
import os
19
20
import subprocess
20
21
 
23
24
import sqlalchemy
24
25
import sqlalchemy.exc
25
26
 
 
27
from ceilometer.openstack.common.db.sqlalchemy import utils
26
28
from ceilometer.openstack.common.gettextutils import _
27
 
from ceilometer.openstack.common import log as logging
28
29
from ceilometer.openstack.common.py3kcompat import urlutils
29
30
from ceilometer.openstack.common import test
30
31
 
31
32
LOG = logging.getLogger(__name__)
32
33
 
33
34
 
34
 
def _get_connect_string(backend, user, passwd, database):
35
 
    """Get database connection
36
 
 
37
 
    Try to get a connection with a very specific set of values, if we get
38
 
    these then we'll run the tests, otherwise they are skipped
39
 
    """
40
 
    if backend == "postgres":
41
 
        backend = "postgresql+psycopg2"
42
 
    elif backend == "mysql":
43
 
        backend = "mysql+mysqldb"
44
 
    else:
45
 
        raise Exception("Unrecognized backend: '%s'" % backend)
46
 
 
47
 
    return ("%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s"
48
 
            % {'backend': backend, 'user': user, 'passwd': passwd,
49
 
               'database': database})
50
 
 
51
 
 
52
 
def _is_backend_avail(backend, user, passwd, database):
53
 
    try:
54
 
        connect_uri = _get_connect_string(backend, user, passwd, database)
55
 
        engine = sqlalchemy.create_engine(connect_uri)
56
 
        connection = engine.connect()
57
 
    except Exception:
58
 
        # intentionally catch all to handle exceptions even if we don't
59
 
        # have any backend code loaded.
60
 
        return False
61
 
    else:
62
 
        connection.close()
63
 
        engine.dispose()
64
 
        return True
65
 
 
66
 
 
67
35
def _have_mysql(user, passwd, database):
68
36
    present = os.environ.get('TEST_MYSQL_PRESENT')
69
37
    if present is None:
70
 
        return _is_backend_avail('mysql', user, passwd, database)
 
38
        return utils.is_backend_avail(backend='mysql',
 
39
                                      user=user,
 
40
                                      passwd=passwd,
 
41
                                      database=database)
71
42
    return present.lower() in ('', 'true')
72
43
 
73
44
 
74
45
def _have_postgresql(user, passwd, database):
75
46
    present = os.environ.get('TEST_POSTGRESQL_PRESENT')
76
47
    if present is None:
77
 
        return _is_backend_avail('postgres', user, passwd, database)
 
48
        return utils.is_backend_avail(backend='postgres',
 
49
                                      user=user,
 
50
                                      passwd=passwd,
 
51
                                      database=database)
78
52
    return present.lower() in ('', 'true')
79
53
 
80
54
 
81
 
def get_db_connection_info(conn_pieces):
82
 
    database = conn_pieces.path.strip('/')
83
 
    loc_pieces = conn_pieces.netloc.split('@')
84
 
    host = loc_pieces[1]
85
 
 
86
 
    auth_pieces = loc_pieces[0].split(':')
87
 
    user = auth_pieces[0]
88
 
    password = ""
89
 
    if len(auth_pieces) > 1:
90
 
        password = auth_pieces[1].strip()
91
 
 
92
 
    return (user, password, database, host)
93
 
 
94
 
 
95
55
def _set_db_lock(lock_path=None, lock_prefix=None):
96
56
    def decorator(f):
97
57
        @functools.wraps(f)
166
126
                         "Failed to run: %s\n%s" % (cmd, output))
167
127
 
168
128
    def _reset_pg(self, conn_pieces):
169
 
        (user, password, database, host) = get_db_connection_info(conn_pieces)
 
129
        (user,
 
130
         password,
 
131
         database,
 
132
         host) = utils.get_db_connection_info(conn_pieces)
170
133
        os.environ['PGPASSWORD'] = password
171
134
        os.environ['PGUSER'] = user
172
135
        # note(boris-42): We must create and drop database, we can't
205
168
                # the MYSQL database, which is easier and less error-prone
206
169
                # than using SQLAlchemy to do this via MetaData...trust me.
207
170
                (user, password, database, host) = \
208
 
                    get_db_connection_info(conn_pieces)
 
171
                    utils.get_db_connection_info(conn_pieces)
209
172
                sql = ("drop database if exists %(db)s; "
210
173
                       "create database %(db)s;") % {'db': database}
211
174
                cmd = ("mysql -u \"%(user)s\" -p\"%(password)s\" -h %(host)s "