trove/run_tests.py

# Copyright 2013 OpenStack Foundation
# Copyright 2013 Rackspace Hosting
# Copyright 2013 Hewlett-Packard Development Company, L.P.
# 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 functools
import gettext
import os
import sys
import traceback

import eventlet
from oslo_log import log as logging
import proboscis
import six
from six.moves import urllib
import wsgi_intercept
from wsgi_intercept.httplib2_intercept import install as wsgi_install

from trove.common import cfg
from trove.common.rpc import service as rpc_service
from trove.common.rpc import version as rpc_version
from trove.common import utils
from trove import rpc
from trove.tests.config import CONFIG
from trove.tests import root_logger

eventlet.monkey_patch(thread=False)

CONF = cfg.CONF
original_excepthook = sys.excepthook


def add_support_for_localization():
    """Adds support for localization in the logging.

    If ../nova/__init__.py exists, add ../ to Python search path, so that
    it will override what happens to be installed in
    /usr/(local/)lib/python...

    """
    path = os.path.join(os.path.abspath(sys.argv[0]), os.pardir, os.pardir)
    possible_topdir = os.path.normpath(path)
    if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
        sys.path.insert(0, possible_topdir)

    if six.PY3:
        gettext.install('nova')
    else:
        gettext.install('nova', unicode=True)


def initialize_trove(config_file):
    from trove.common import pastedeploy

    root_logger.DefaultRootLogger()

    cfg.CONF(args=[],
             project='trove',
             default_config_files=[config_file])
    logging.setup(CONF, None)
    topic = CONF.taskmanager_queue
    rpc.init(CONF)

    taskman_service = rpc_service.RpcService(
        CONF.taskmanager_rpc_encr_key, topic=topic,
        rpc_api_version=rpc_version.RPC_API_VERSION,
        manager='trove.taskmanager.manager.Manager')
    taskman_service.start()

    return pastedeploy.paste_deploy_app(config_file, 'trove', {})


def datastore_init():
    # Adds the datastore for mysql (needed to make most calls work).
    from trove.configuration.models import DatastoreConfigurationParameters
    from trove.datastore import models

    models.DBDatastore.create(
        id=CONFIG.dbaas_datastore_id, name=CONFIG.dbaas_datastore,
        default_version_id=CONFIG.dbaas_datastore_version_id)

    models.DBDatastore.create(id=utils.generate_uuid(),
                              name=CONFIG.dbaas_datastore_name_no_versions,
                              default_version_id=None)

    main_dsv = models.DBDatastoreVersion.create(
        id=CONFIG.dbaas_datastore_version_id,
        datastore_id=CONFIG.dbaas_datastore_id,
        name=CONFIG.dbaas_datastore_version,
        manager="mysql",
        image_id='c00000c0-00c0-0c00-00c0-000c000000cc',
        packages='test packages',
        active=1)
    models.DBDatastoreVersion.create(
        id="d00000d0-00d0-0d00-00d0-000d000000dd",
        datastore_id=CONFIG.dbaas_datastore_id,
        name='mysql_inactive_version', manager="mysql",
        image_id='c00000c0-00c0-0c00-00c0-000c000000cc',
        packages=None, active=0)

    def add_parm(name, data_type, max_size, min_size=0, restart_required=0):
        DatastoreConfigurationParameters.create(
            datastore_version_id=main_dsv.id,
            name=name,
            restart_required=restart_required,
            max_size=max_size,
            min_size=0,
            data_type=data_type,
            deleted=0,
            deleted_at=None)

    add_parm('key_buffer_size', 'integer', 4294967296)
    add_parm('connect_timeout', 'integer', 65535)
    add_parm('join_buffer_size', 'integer', 4294967296)
    add_parm('local_infile', 'integer', 1)
    add_parm('collation_server', 'string', None, None)
    add_parm('innodb_buffer_pool_size', 'integer', 57671680,
             restart_required=1)


def initialize_database():
    from trove.db import get_db_api
    from trove.db.sqlalchemy import session
    db_api = get_db_api()
    db_api.drop_db(CONF)  # Destroys the database, if it exists.
    db_api.db_sync(CONF)
    session.configure_db(CONF)
    datastore_init()
    db_api.configure_db(CONF)


def initialize_fakes(app):
    # Set up WSGI interceptor. This sets up a fake host that responds each
    # time httplib tries to communicate to localhost, port 8779.
    def wsgi_interceptor(*args, **kwargs):

        def call_back(env, start_response):
            path_info = env.get('PATH_INFO')
            if path_info:
                env['PATH_INFO'] = urllib.parse.unquote(path_info)
            return app.__call__(env, start_response)

        return call_back

    wsgi_intercept.add_wsgi_intercept('localhost',
                                      CONF.bind_port,
                                      wsgi_interceptor)
    from trove.tests.util import event_simulator
    event_simulator.monkey_patch()
    from trove.tests.fakes import taskmanager
    taskmanager.monkey_patch()


def parse_args_for_test_config():
    test_conf = 'etc/tests/localhost.test.conf'
    repl = False
    new_argv = []
    for index in range(len(sys.argv)):
        arg = sys.argv[index]
        print(arg)
        if arg[:14] == "--test-config=":
            test_conf = arg[14:]
        elif arg == "--repl":
            repl = True
        else:
            new_argv.append(arg)
    sys.argv = new_argv
    return test_conf, repl


def run_tests(repl):
    """Runs all of the tests."""

    if repl:
        # Actually show errors in the repl.
        sys.excepthook = original_excepthook

        def no_thanks(exit_code):
            print("Tests finished with exit code %d." % exit_code)
        sys.exit = no_thanks

    proboscis.TestProgram().run_and_exit()

    if repl:
        import code
        code.interact()


def import_tests():
    # F401 unused imports needed for tox tests
    from trove.tests.api import backups  # noqa
    from trove.tests.api import configurations  # noqa
    from trove.tests.api import databases  # noqa
    from trove.tests.api import datastores  # noqa
    from trove.tests.api import flavors  # noqa
    from trove.tests.api import header  # noqa
    from trove.tests.api import instances as rd_instances  # noqa
    from trove.tests.api import instances_actions as rd_actions  # noqa
    from trove.tests.api import instances_delete  # noqa
    from trove.tests.api import instances_mysql_down  # noqa
    from trove.tests.api import instances_resize  # noqa
    from trove.tests.api import limits  # noqa
    from trove.tests.api.mgmt import accounts  # noqa
    from trove.tests.api.mgmt import admin_required  # noqa
    from trove.tests.api.mgmt import hosts  # noqa
    from trove.tests.api.mgmt import instances as mgmt_instances  # noqa
    from trove.tests.api.mgmt import instances_actions as mgmt_actions  # noqa
    from trove.tests.api.mgmt import malformed_json  # noqa
    from trove.tests.api.mgmt import storage  # noqa
    from trove.tests.api import replication  # noqa
    from trove.tests.api import root  # noqa
    from trove.tests.api import root_on_create  # noqa
    from trove.tests.api import user_access  # noqa
    from trove.tests.api import users  # noqa
    from trove.tests.api import versions  # noqa
    from trove.tests.db import migrations  # noqa


def main(import_func):
    try:
        wsgi_install()
        add_support_for_localization()
        # Load Trove app
        # Paste file needs absolute path
        config_file = os.path.realpath('etc/trove/trove.conf.test')
        # 'etc/trove/test-api-paste.ini'
        app = initialize_trove(config_file)
        # Initialize sqlite database.
        initialize_database()
        # Swap out WSGI, httplib, and other components with test doubles.
        initialize_fakes(app)

        # Initialize the test configuration.
        test_config_file, repl = parse_args_for_test_config()
        CONFIG.load_from_file(test_config_file)

        import_func()

        from trove.tests.util import event_simulator
        event_simulator.run_main(functools.partial(run_tests, repl))

    except Exception as e:
        # Printing the error manually like this is necessary due to oddities
        # with sys.excepthook.
        print("Run tests failed: %s" % e)
        traceback.print_exc()
        raise


if __name__ == "__main__":
    main(import_tests)