Source code for panko.tests.functional.gabbi.fixtures
#
# Copyright 2015 Red Hat. 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.
"""Fixtures used during Gabbi-based test runs."""
import datetime
import os
from unittest import case
from gabbi import fixture
from oslo_config import cfg
from oslo_policy import opts
from oslo_utils import fileutils
from oslo_utils import uuidutils
import six
from six.moves.urllib import parse as urlparse
import sqlalchemy_utils
from panko.api import app
from panko import service
from panko import storage
from panko.storage import models
# NOTE(chdent): Hack to restore semblance of global configuration to
# pass to the WSGI app used per test suite. LOAD_APP_KWARGS are the olso
# configuration, and the pecan application configuration of
# which the critical part is a reference to the current indexer.
LOAD_APP_KWARGS = None
def setup_app():
global LOAD_APP_KWARGS
return app.load_app(**LOAD_APP_KWARGS)
[docs]class ConfigFixture(fixture.GabbiFixture):
"""Establish the relevant configuration for a test run."""
[docs] def start_fixture(self):
"""Set up config."""
global LOAD_APP_KWARGS
self.conf = None
# Determine the database connection.
db_url = os.environ.get('PIFPAF_URL', "sqlite://").replace(
"mysql://", "mysql+pymysql://")
if not db_url:
raise case.SkipTest('No database connection configured')
conf = self.conf = service.prepare_service([], [])
opts.set_defaults(self.conf)
content = ('{"default": ""}')
if six.PY3:
content = content.encode('utf-8')
self.tempfile = fileutils.write_to_tempfile(content=content,
prefix='policy',
suffix='.json')
conf.set_override("policy_file", self.tempfile,
group='oslo_policy')
conf.set_override(
'api_paste_config',
os.path.abspath(
'panko/tests/functional/gabbi/gabbi_paste.ini')
)
parsed_url = list(urlparse.urlparse(db_url))
parsed_url[2] += '-%s' % uuidutils.generate_uuid(dashed=False)
db_url = urlparse.urlunparse(parsed_url)
conf.set_override('connection', db_url, group='database')
if (parsed_url[0].startswith("mysql")
or parsed_url[0].startswith("postgresql")):
sqlalchemy_utils.create_database(conf.database.connection)
self.conn = storage.get_connection_from_config(self.conf)
self.conn.upgrade()
LOAD_APP_KWARGS = {
'conf': conf,
}
[docs] def stop_fixture(self):
"""Reset the config and remove data."""
if self.conn:
self.conn.clear()
if self.conf:
storage.get_connection_from_config(self.conf).clear()
[docs]class EventDataFixture(ConfigFixture):
"""Instantiate some sample event data for use in testing."""
[docs] def start_fixture(self):
"""Create some events."""
super(EventDataFixture, self).start_fixture()
events = []
name_list = ['chocolate.chip', 'peanut.butter', 'sugar']
for ix, name in enumerate(name_list):
timestamp = datetime.datetime.utcnow()
message_id = 'fea1b15a-1d47-4175-85a5-a4bb2c72924{}'.format(ix)
traits = [models.Trait('type', 1, name),
models.Trait('ate', 2, ix)]
event = models.Event(message_id,
'cookies_{}'.format(name),
timestamp,
traits, {'nested': {'inside': 'value'}})
events.append(event)
self.conn.record_events(events)
[docs]class CORSConfigFixture(fixture.GabbiFixture):
"""Inject mock configuration for the CORS middleware."""
[docs] def start_fixture(self):
# Here we monkeypatch GroupAttr.__getattr__, necessary because the
# paste.ini method of initializing this middleware creates its own
# ConfigOpts instance, bypassing the regular config fixture.
def _mock_getattr(instance, key):
if key != 'allowed_origin':
return self._original_call_method(instance, key)
return "http://valid.example.com"
self._original_call_method = cfg.ConfigOpts.GroupAttr.__getattr__
cfg.ConfigOpts.GroupAttr.__getattr__ = _mock_getattr
[docs] def stop_fixture(self):
"""Remove the monkeypatch."""
cfg.ConfigOpts.GroupAttr.__getattr__ = self._original_call_method