Source code for stoqlib.database.testsuite

# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4

## Copyright (C) 2006-2013 Async Open Source <>
## All rights reserved
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## GNU General Public License for more details.
## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., or visit:
## Author(s): Stoq Team <>
"""Database routines which are used by the testsuite"""

# FIXME: This should be moved/merged with the normal bootstrap

# This needs to before the other commits, so the externals/
# path is properly setup.
from stoqlib.lib.kiwilibrary import library
library  # pylint: disable=W0104

import logging
import os

from kiwi.component import provide_utility, utilities
from storm.expr import And
from storm.tracer import install_tracer, remove_tracer_type

from stoqlib.database.admin import initialize_system, ensure_admin_user
from stoqlib.database.interfaces import (
    ICurrentBranch, ICurrentBranchStation, ICurrentUser)
from stoqlib.database.runtime import new_store, get_default_store
from stoqlib.database.settings import db_settings
from stoqlib.domain.person import Branch, LoginUser, Person
from stoqlib.domain.station import BranchStation
from stoqlib.importers.stoqlibexamples import create
from stoqlib.lib.interfaces import IApplicationDescriptions, ISystemNotifier
from stoqlib.lib.message import DefaultSystemNotifier
from stoqlib.lib.osutils import get_username
from stoqlib.lib.pluginmanager import get_plugin_manager
from stoqlib.lib.settings import get_settings
from import get_hostname

log = logging.getLogger(__name__)

[docs]class StoqlibTestsuiteTracer(object): def __init__(self): self.reset()
[docs] def install(self): self.reset() install_tracer(self)
[docs] def remove(self): remove_tracer_type(type(self))
[docs] def reset(self): self.count = 0
[docs] def connection_raw_execute_success(self, connection, raw_cursor, statement, params): self.count += 1
[docs] def connection_raw_execute_error(self, connection, raw_cursor, statement, params, error): self.count += 1
# This notifier implementation is here to workaround trial; which # refuses to quit if SystemExit is raised or if sys.exit() is called. # For now it is assumed that errors() are fatal, that might change in # the near future
[docs]class TestsuiteNotifier(DefaultSystemNotifier): def __init__(self): self._messages = []
[docs] def reset(self): messages = self._messages self._messages = [] return messages
[docs] def message(self, name, short, description): self._messages.append((name, short, description))
[docs] def error(self, short, description): DefaultSystemNotifier.error(self, short, description) os._exit(1)
test_system_notifier = TestsuiteNotifier() def _provide_database_settings(): db_settings.username = os.environ.get(u'STOQLIB_TEST_USERNAME', get_username()) db_settings.hostname = os.environ.get(u'PGHOST', u'localhost') db_settings.port = int(os.environ.get(u'PGPORT', u'5432')) db_settings.dbname = os.environ.get(u'STOQLIB_TEST_DBNAME', u'%s_test' % db_settings.username) db_settings.password = u'' def _provide_current_user(): default_store = get_default_store() user = default_store.find(LoginUser, username=u'admin').one() assert user provide_utility(ICurrentUser, user, replace=True) def _provide_current_station(station_name=None, branch_name=None): if not station_name: station_name = get_hostname() store = new_store() if branch_name: branch = store.find(Person, And( == branch_name, Branch.person_id == else: branches = store.find(Branch) if branches.count() == 0: person = Person(name=u"test", store=store) branch = Branch(person=person, store=store) else: branch = branches[0] provide_utility(ICurrentBranch, branch) station = BranchStation.get_station(store, branch, station_name) if not station: station = BranchStation.create(store, branch, station_name) assert station assert station.is_active provide_utility(ICurrentBranchStation, station) store.commit(close=True) def _provide_app_info(): from stoqlib.lib.interfaces import IAppInfo from stoqlib.lib.appinfo import AppInfo info = AppInfo() info.set(u"name", u"Stoqlib") info.set(u"version", u"1.0.0") provide_utility(IAppInfo, info) # Public API
[docs]def provide_database_settings(dbname=None, address=None, port=None, username=None, password=None, createdb=True): """ Provide database settings. :param dbname: :param address: :param port: :param username: :param password: :param create: Create a new empty database if one is missing """ if not username: username = get_username() if not dbname: dbname = username + u'_test' if not address: address = os.environ.get(u'PGHOST', u'') if not port: port = os.environ.get(u'PGPORT', u'5432') if not password: password = u"" # Remove all old utilities pointing to the previous database. utilities.clean() provide_utility(ISystemNotifier, test_system_notifier, replace=True) _provide_application_descriptions() _provide_domain_slave_mapper() _provide_app_info() db_settings.address = address db_settings.port = port db_settings.dbname = dbname db_settings.username = username db_settings.password = password rv = False if createdb or not db_settings.database_exists(dbname): db_settings.clean_database(dbname, force=True) rv = True return rv
def _provide_domain_slave_mapper(): from stoqlib.gui.interfaces import IDomainSlaveMapper from stoqlib.gui.slaves.domainslavemapper import DefaultDomainSlaveMapper provide_utility(IDomainSlaveMapper, DefaultDomainSlaveMapper(), replace=True) def _provide_application_descriptions(): from stoq.lib.applist import ApplicationDescriptions provide_utility(IApplicationDescriptions, ApplicationDescriptions(), replace=True)
[docs]def provide_utilities(station_name, branch_name=None): """ Provide utilities like current user and current station. :param station_name: :param branch_name: """ _provide_current_user() _provide_current_station(station_name, branch_name) _provide_domain_slave_mapper()
def _enable_plugins(): manager = get_plugin_manager() for plugin in [u'ecf', u'nfe', u'optical']: if not manager.is_installed(plugin): # STOQLIB_TEST_QUICK won't let dropdb on testdb run. Just a # precaution to avoid trying to install it again with new_store() as store: manager.install_plugin(store, plugin) else: # Make sure that the plugin is imported so sys.path is properly # setup plugin = manager.get_plugin(plugin) plugin # pylint: disable=W0104
[docs]def bootstrap_suite(address=None, dbname=None, port=5432, username=None, password=u"", station_name=None, quick=False): """ Test. :param address: :param dbname: :param port: :param username: :param password: :param station_name: :param quick: """ os.environ['STOQ_TESTSUIT_RUNNING'] = '1' # This will only be required when we use uuid.UUID instances # for UUIDCol #import psycopg2.extras #psycopg2.extras.register_uuid() empty = provide_database_settings(dbname, address, port, username, password, createdb=not quick) # Reset the user settings (loaded from ~/.stoq/settings), so that user # preferences don't affect the tests. settings = get_settings() settings.reset() if quick and not empty: provide_utilities(station_name) _enable_plugins() return initialize_system(testsuite=True, force=True) # Commit before trying to apply patches which requires an exclusive lock # to all tables. _enable_plugins() ensure_admin_user(u"") create(utilities=True, create_users=True)