summaryrefslogtreecommitdiff
path: root/oslo_db/sqlalchemy/test_base.py
blob: 601a2c87f7fb89fb2838b7bab11f5b76e39bba97 (plain)
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
# Copyright (c) 2013 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import fixtures

try:
    from oslotest import base as test_base
except ImportError:
    raise NameError('Oslotest is not installed. Please add oslotest in your'
                    ' test-requirements')


import six

from oslo_db import exception
from oslo_db.sqlalchemy import provision
from oslo_db.sqlalchemy import session
from oslo_db.sqlalchemy import utils


class DbFixture(fixtures.Fixture):
    """Basic database fixture.

    Allows to run tests on various db backends, such as SQLite, MySQL and
    PostgreSQL. By default use sqlite backend. To override default backend
    uri set env variable OS_TEST_DBAPI_CONNECTION with database admin
    credentials for specific backend.
    """

    DRIVER = "sqlite"

    # these names are deprecated, and are not used by DbFixture.
    # they are here for backwards compatibility with test suites that
    # are referring to them directly.
    DBNAME = PASSWORD = USERNAME = 'openstack_citest'

    def __init__(self, test):
        super(DbFixture, self).__init__()

        self.test = test

    def setUp(self):
        super(DbFixture, self).setUp()

        try:
            self.provision = provision.ProvisionedDatabase(self.DRIVER)
            self.addCleanup(self.provision.dispose)
        except exception.BackendNotAvailable:
            msg = '%s backend is not available.' % self.DRIVER
            return self.test.skip(msg)
        else:
            self.test.provision = self.provision
            self.test.engine = self.provision.engine
            self.addCleanup(setattr, self.test, 'engine', None)
            self.test.sessionmaker = session.get_maker(self.test.engine)
            self.addCleanup(setattr, self.test, 'sessionmaker', None)


class DbTestCase(test_base.BaseTestCase):
    """Base class for testing of DB code.

    Using `DbFixture`. Intended to be the main database test case to use all
    the tests on a given backend with user defined uri. Backend specific
    tests should be decorated with `backend_specific` decorator.
    """

    FIXTURE = DbFixture

    def setUp(self):
        super(DbTestCase, self).setUp()
        self.useFixture(self.FIXTURE(self))


class OpportunisticTestCase(DbTestCase):
    """Placeholder for backwards compatibility."""

ALLOWED_DIALECTS = ['sqlite', 'mysql', 'postgresql']


def backend_specific(*dialects):
    """Decorator to skip backend specific tests on inappropriate engines.

    ::dialects: list of dialects names under which the test will be launched.
    """
    def wrap(f):
        @six.wraps(f)
        def ins_wrap(self):
            if not set(dialects).issubset(ALLOWED_DIALECTS):
                raise ValueError(
                    "Please use allowed dialects: %s" % ALLOWED_DIALECTS)
            if self.engine.name not in dialects:
                msg = ('The test "%s" can be run '
                       'only on %s. Current engine is %s.')
                args = (utils.get_callable_name(f), ' '.join(dialects),
                        self.engine.name)
                self.skip(msg % args)
            else:
                return f(self)
        return ins_wrap
    return wrap


class MySQLOpportunisticFixture(DbFixture):
    DRIVER = 'mysql'


class PostgreSQLOpportunisticFixture(DbFixture):
    DRIVER = 'postgresql'


class MySQLOpportunisticTestCase(OpportunisticTestCase):
    FIXTURE = MySQLOpportunisticFixture


class PostgreSQLOpportunisticTestCase(OpportunisticTestCase):
    FIXTURE = PostgreSQLOpportunisticFixture