trove/trove/tests/config.py

# Copyright (c) 2011 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.

"""Handles configuration options for the tests.

The tests are capable of running in other contexts, such as in a VM or against
a real deployment. Using this configuration ensures we can run them in other
environments if we choose to.

"""

from collections import Mapping
from datetime import datetime
import json
import os


# TODO(tim.simpson): I feel like this class already exists somewhere in core
#                    Python.
class FrozenDict(Mapping):

    def __init__(self, original):
        self.original = original

    def __len__(self):
        return self.original.__len__()

    def __iter__(self, *args, **kwargs):
        return self.original.__iter__(self, *args, **kwargs)

    def __getitem__(self, *args, **kwargs):
        return self.original.__getitem__(*args, **kwargs)

    def __str__(self):
        return self.original.__str__()


USAGE_ENDPOINT = os.environ.get("USAGE_ENDPOINT",
                                "trove.tests.util.usage.UsageVerifier")


class TestConfig(object):
    """
    Holds test configuration values which can be accessed as attributes
    or using the values dictionary.
    """

    def __init__(self):
        """
        Create TestConfig, and set default values. These will be overwritten by
        the "load_from" methods below.
        """
        self._loaded_files = []
        self._values = {
            'clean_slate': os.environ.get("CLEAN_SLATE", "False") == "True",
            'fake_mode': os.environ.get("FAKE_MODE", "False") == "True",
            'nova_auth_url': "http://localhost/identity/v2.0",
            'trove_auth_url': "http://localhost/identity/v2.0/tokens",
            'dbaas_url': "http://localhost:8775/v1.0/dbaas",
            'version_url': "http://localhost:8775/",
            'nova_url': "http://localhost:8774/v2",
            'swift_url': "http://localhost:8080/v1/AUTH_",
            'dbaas_datastore': "mysql",
            'dbaas_datastore_id': "a00000a0-00a0-0a00-00a0-000a000000aa",
            'dbaas_datastore_name_no_versions': "Test_Datastore_1",
            'dbaas_datastore_version': "5.5",
            'dbaas_datastore_version_id': "b00000b0-00b0-0b00-00b0-"
                                          "000b000000bb",
            'dbaas_inactive_datastore_version': "mysql_inactive_version",
            'instance_create_time': 16 * 60,
            'mysql_connection_method': {"type": "direct"},
            'typical_nova_image_name': None,
            'white_box': os.environ.get("WHITE_BOX", "False") == "True",
            'test_mgmt': False,
            'use_local_ovz': False,
            "known_bugs": {},
            "in_proc_server": True,
            "report_directory": os.environ.get("REPORT_DIRECTORY", None),
            "trove_volume_support": True,
            "trove_volume_size": 1,
            "trove_max_volumes_per_tenant": 100,
            "trove_max_instances_per_tenant": 55,
            "usage_endpoint": USAGE_ENDPOINT,
            "root_on_create": False,
            "mysql": {
                "configurations": {
                    "valid_values": {
                        "connect_timeout": 120,
                        "local_infile": 0,
                        "collation_server": "latin1_swedish_ci"
                    },
                    "appending_values": {
                        "join_buffer_size": 1048576,
                        "connect_timeout": 15
                    },
                    "nondynamic_parameter": {
                        "join_buffer_size": 1048576,
                        "innodb_buffer_pool_size": 57671680
                    },
                    "out_of_bounds_under": {
                        "connect_timeout": -10
                    },
                    "out_of_bounds_over": {
                        "connect_timeout": 1000000
                    },
                    "parameters_list": [
                        "key_buffer_size",
                        "connect_timeout"
                    ]
                },
                "volume_support": True,
            },
            "redis": {"volume_support": False},
        }
        self._frozen_values = FrozenDict(self._values)
        self._users = None

    def get(self, name, default_value):
        return self.values.get(name, default_value)

    def get_report(self):
        return PrintReporter()

    def load_from_line(self, line):
        index = line.find("=")
        if index >= 0:
            key = line[:index]
            value = line[index + 1:]
            self._values[key] = value

    def load_include_files(self, original_file, files):
        directory = os.path.dirname(original_file)
        for file_sub_path in files:
            file_full_path = os.path.join(directory, file_sub_path)
            self.load_from_file(file_full_path)

    def load_from_file(self, file_path):
        if file_path in self._loaded_files:
            return
        file_contents = open(file_path, "r").read()
        try:
            contents = json.loads(file_contents)
        except Exception as exception:
            raise RuntimeError("Error loading conf file \"%s\"." % file_path,
                               exception)
        finally:
            self._loaded_files.append(file_path)

        if "include-files" in contents:
            self.load_include_files(file_path, contents['include-files'])
            del contents['include-files']
        self._values.update(contents)

    def __getattr__(self, name):
        if name not in self._values:
            raise AttributeError('Configuration value "%s" not found.' % name)
        else:
            return self._values[name]

    def python_cmd_list(self):
        """The start of a command list to use when running Python scripts."""
        commands = []
        if self.use_venv:
            commands.append("%s/tools/with_venv.sh" % self.nova_code_root)
            return list
        commands.append("python")
        return commands

    @property
    def users(self):
        if self._users is None:
            from trove.tests.util.users import Users
            self._users = Users(self.values['users'])
        return self._users

    @property
    def values(self):
        return self._frozen_values


class PrintReporter(object):

    def log(self, msg):
        print("%s [REPORT] %s" % (str(datetime.now()), msg))

    def update(self):
        pass  # Ignore. This is used in other reporters.


CONFIG = TestConfig()
del TestConfig.__init__