def test1(self):
     import transaction
     Base = sqlahelper.get_base()
     class Person(Base):
         __tablename__ = "people"
         id = sa.Column(sa.Integer, primary_key=True)
         first_name = sa.Column(sa.Unicode(100), nullable=False)
         last_name = sa.Column(sa.Unicode(100), nullable=False)
     engine = sa.create_engine(self.db1.url)
     sqlahelper.add_engine(engine)
     Base.metadata.create_all()
     fred = Person(id=1, first_name=u"Fred", last_name=u"Flintstone")
     wilma = Person(id=2, first_name=u"Wilma", last_name=u"Flintstone")
     barney = Person(id=3, first_name=u"Barney", last_name=u"Rubble")
     betty = Person(id=4, first_name=u"Betty", last_name=u"Rubble")
     Session = sqlahelper.get_session()
     sess = Session()
     sess.add_all([fred, wilma, barney, betty])
     transaction.commit()
     sess.expunge_all()
     del fred, wilma, barney, betty
     # Can we get back a record?
     barney2 = sess.query(Person).get(3)
     self.assertEqual(barney2.id, 3)
     self.assertEqual(barney2.first_name, u"Barney")
     self.assertEqual(barney2.last_name, u"Rubble")
     sa.select([Person.first_name])
     # Can we iterate the first names in reverse alphabetical order?
     q = sess.query(Person.first_name).order_by(Person.first_name.desc())
     result = [x.first_name for x in q]
     control = [u"Wilma", u"Fred", u"Betty", u"Barney"]
     self.assertEqual(result, control)

def create_engine(create_engine_dotted, dburl):
    if create_engine_dotted:
        return import_symbol(create_engine_dotted)(dburl)
    else:
        engine = sa.create_engine(dburl)
        sqlahelper.add_engine(engine)
        return engine

def setUpCommon():
    import c2cgeoportal
    c2cgeoportal.schema = 'main'
    c2cgeoportal.srid = 21781
    c2cgeoportal.caching.init_region({'backend': 'dogpile.cache.memory'})

    # if test.in does not exist (because the z3c.recipe.filetemplate
    # part hasn't been executed) then db_url is None
    if db_url is None:  # pragma: nocover
        return

    # verify that we have a working database connection before going
    # forward
    from sqlalchemy import create_engine
    from sqlalchemy.exc import OperationalError
    engine = create_engine(db_url)
    try:
        engine.connect()
    except OperationalError:  # pragma: nocover
        return

    import sqlahelper
    sqlahelper.add_engine(engine)

    from c2cgeoportal.models import Base
    Base.metadata.create_all()

def initialized(ev):
    PTAH = ptah.get_settings(ptah.CFG_ID_PTAH, ev.registry)

    # mail
    PTAH['Mailer'] = DummyMailer()
    PTAH['full_email_address'] = formataddr(
        (PTAH['email_from_name'], PTAH['email_from_address']))

    # sqla
    SQLA = ptah.get_settings(ptah.CFG_ID_SQLA, ev.registry)
    url = SQLA['url']
    if url:
        engine_args = {}
        if SQLA['cache']:
            cache = {}
            engine_args['execution_options'] = \
                {'compiled_cache': cache}
            SQLA['sqlalchemy_cache'] = cache
        try:
            engine = sqlahelper.get_engine()
        except: # pragma: no cover
            engine = sqlalchemy.engine_from_config(
                {'sqlalchemy.url': url}, 'sqlalchemy.', **engine_args)
            sqlahelper.add_engine(engine)

    # ptah manage
    if PTAH['manage']:
        ev.config.add_route(
            'ptah-manage', '/ptah-manage/*traverse',
            factory=ptah.manage.PtahManageRoute, use_global_views=True)
        ptah.manage.set_access_manager(
            ptah.manage.PtahAccessManager())

def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """

    config = Configurator(settings=settings)
    settings = config.get_settings()
    settings.update(yaml.load(file(settings.get("app.cfg"))))

    add_mako_renderer(config, ".html")
    add_mako_renderer(config, ".js")

    engine = engine_from_config(settings, 'sqlalchemy.')
    sqlahelper.add_engine(engine)
    dbreflection.init(engine)

    config.set_authorization_policy(ACLAuthorizationPolicy())

    config.add_request_method(
        _create_get_user_from_request(settings),
        name="user",
        property=True
    )

    authtkt_authentication_policy = AuthTktAuthenticationPolicy(
        settings.get('authtkt')['secret'],
        cookie_name=settings.get('authtkt')['cookie_name'],
        hashalg='sha512'
    )

    remote_user_authentication_policy = RemoteUserAuthenticationPolicy()
    policies = [remote_user_authentication_policy, authtkt_authentication_policy]
    authentication_policy = MultiAuthenticationPolicy(policies)
    config.set_authentication_policy(authentication_policy)

    config.include('pyramid_mako')
    config.include(pyramid_tm.includeme)

    config.add_static_view('static', 'static', cache_max_age=3600)

    config.add_route('home', '/')
    config.add_route('viewer', '/viewer.js')
    config.add_route('image_proxy', '/img/{type}/{id}')
    config.add_route('pdf_proxy', '/pdf/{type}/{id}.pdf')

    # historic parcels
    config.add_route('historic_parcel_get', '/historic_parcel/{id}')
    config.add_route('historic_parcel', '/historic_parcel')
    config.add_route('historic_parcel_doc', '/historic_parcel_doc')

    # print proxy routes
    config.add_route('printproxy', '/printproxy')
    config.add_route('printproxy_info', '/printproxy/info.json')
    config.add_route('printproxy_create', '/printproxy/create.json')
    config.add_route('printproxy_get', '/printproxy/{file}.printout')

    # mutation
    config.add_route('mutation_list', '/mutation/list')

    config.scan()
    return config.make_wsgi_app()

 def test_add_engine_twice(self):
     db1 = sa.create_engine(self.db1.url)
     db2 = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(db1)
     self.assertIs(sqlahelper.get_session().bind, db1)
     sqlahelper.add_engine(db2)
     self.assertIs(sqlahelper.get_session().bind, db2)
     self.assertIs(sqlahelper.get_session().bind, sqlahelper.sessions.default.registry.registry.value.bind)

    def setUp(self):
        try:
            engine = sqlahelper.get_engine()
        except:  # pragma: no cover
            engine = sqlalchemy.engine_from_config({"sqlalchemy.url": "sqlite://"})
            sqlahelper.add_engine(engine)

        self._setup_pyramid()

def init_engine(config):
    engine = create_engine(
        config.get('SQLALCHEMY_DATABASE_URI'),
            echo=config.get('SQLALCHEMY_ECHO'),
            pool_recycle=config.get('SQLALCHEMY_POOL_RECYCLE'),
            convert_unicode=True
        )

    add_engine(engine)

def main(global_config, **settings):
    engine = engine_from_config(settings)
    sqlahelper.add_engine(engine)
    config = Configurator(settings=settings,
        root_factory='.models.ShirlyResource')
    config.add_subscriber(add_renderer_globals, 'pyramid.events.BeforeRender')
    config.include('.')
    config.scan()
    return config.make_wsgi_app()

def init_engine(config):
    engine = create_engine(
        config['uri'],
        echo=config['echo'],
        pool_recycle=config['pool_recycle'],
        convert_unicode=True
    )

    add_engine(engine)

    def setUp(self):
        try:
            engine = sqlahelper.get_engine()
        except:
            engine = sqlalchemy.engine_from_config(
                {'sqlalchemy.url': 'sqlite://'})
            sqlahelper.add_engine(engine)

        self._setup_pyramid()
        self._setup_ptah()

def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """
    engine = engine_from_config(settings)
    sqlahelper.add_engine(engine)
    config = Configurator(settings=settings)
    config.include('.fainit')
    config.add_static_view('static', 'static', cache_max_age=3600)
    config.add_route('home', '/')
    config.scan()
    return config.make_wsgi_app()

    def loginchange(self):
        new_password = self.request.params.get('newPassword', None)
        new_password_confirm = self.request.params.get('confirmNewPassword', None)
        if new_password is None or new_password_confirm is None:
            raise HTTPBadRequest('"newPassword" and "confirmNewPassword" should be \
                   available in request params')

        # check if loggedin
        if not self.request.user:
            raise HTTPUnauthorized('bad credentials')
        if new_password != new_password_confirm:
            raise HTTPBadRequest("the new password and the new password \
                   confirmation don't match")

        u = self.request.user
        u._set_password(new_password)
        u.is_password_changed = True
        DBSession.flush()
        log.info("password changed for user: %s" % self.request.user.username)

        # handle replication
        if 'auth_replication_enabled' in self.request.registry.settings and \
                self.request.registry.settings['auth_replication_enabled'] == \
                'true':  # pragma: no cover
            try:
                log.debug("trying to find if engine set for replication exists")
                engine = sqlahelper.get_engine('replication')
            except RuntimeError:
                log.debug("engine for replication doesn't exist yet, trying \
                          to create")
                engine = engine_from_config(
                    self.request.registry.settings,
                    'sqlalchemy_replication.')
                sqlahelper.add_engine(engine, 'replication')

            DBSession2 = scoped_session(sessionmaker(bind=engine))

            dbuser_r = DBSession2.query(User).filter(User.id == self.request.user.id)
            result = dbuser_r.all()
            if len(result) == 0:
                msg = 'user not found in replication target database: %s' \
                    % self.request.user.username
                log.exception(msg)
                return HTTPBadRequest(msg)  # pragma nocover
            else:
                u_r = dbuser_r.all()[0]
                u_r._set_password(new_password)
                u_r.is_password_changed = True
                DBSession2.commit()
                log.info("password changed in replication target database \
                    for user: %s" % self.request.user.username)

        return Response('true', cache_control="no-cache")

 def test_multiple_engines_without_default(self):
     db1 = sa.create_engine(self.db1.url)
     db2 = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(db1, "db1")
     sqlahelper.add_engine(db2, "db2")
     # Can we retrieve the engines?
     self.assertIs(sqlahelper.get_engine("db1"), db1)
     self.assertIs(sqlahelper.get_engine("db2"), db2)
     # There should be no default engine
     self.assertIsNone(sqlahelper.get_session().bind)
     self.assertIsNone(sqlahelper.get_base().metadata.bind)
     self.assertIsNone(sqlahelper.get_engine())

 def test_multiple_engines(self):
     default = sa.create_engine(self.db1.url)
     stats = sa.create_engine(self.db2.url)
     sqlahelper.add_engine(default)
     sqlahelper.add_engine(stats, "stats")
     # Can we retrieve the engines?
     self.assertIs(sqlahelper.get_engine(), default)
     self.assertIs(sqlahelper.get_engine("default"), default)
     self.assertIs(sqlahelper.get_engine("stats"), stats)
     # Are the session binding and base binding set correctly?
     self.assertIs(sqlahelper.get_session().bind, default)
     self.assertIs(sqlahelper.get_base().metadata.bind, default)

def includeme(config):
    settings = config.registry.settings

    # add default values for some global settings
    templates_dir = list(filter(
        None, settings.get('templates_dir', '').splitlines()))
    templates_dir.append('pygall:templates')
    settings['mako.directories'] = templates_dir
    if 'photos_dir' not in settings:
        settings['photos_dir'] = 'photos'
    if 'upload_dir' not in settings:
        settings['upload_dir'] = 'upload'
    allow_cdn = asbool(settings.get('allow_cdn', 'false'))
    settings['allow_cdn'] = allow_cdn
    ip.set_dest_dir(settings['photos_dir'])
    mkdir_p(settings['upload_dir'])

    init_security(settings)
    init_resources(settings)

    authentication_policy = AuthTktAuthenticationPolicy(
            settings['authtkt_secret'], callback=groupfinder)
    authorization_policy = ACLAuthorizationPolicy()

    config.add_settings(settings)
    config.set_root_factory(RootFactory)
    config.set_locale_negotiator(default_locale_negotiator)
    config.set_authentication_policy(authentication_policy)
    config.set_authorization_policy(authorization_policy)

    # initialize database
    engine = sqlalchemy.engine_from_config(settings, 'sqlalchemy.')
    sqlahelper.add_engine(engine)
    config.include(pyramid_tm.includeme)

    # formalchemy
    config.include('pyramid_formalchemy')
    config.include('fa.jquery')
    config.formalchemy_admin('admin', package='pygall',
            view='fa.jquery.pyramid.ModelView', factory=FAModelsFactory)

    # i18n
    config.add_translation_dirs('pygall:locale')

    # bind the mako renderer to other file extensions
    config.add_renderer('.mako', mako_renderer_factory)
    config.add_renderer('jsonp', JSONP(param_name='callback'))

    config.include(add_routes)
    config.include(add_views)

    config.scan('pygall')

def main(argv=sys.argv):
    if len(argv) != 2:
        usage(argv)
    config_uri = argv[1]
    setup_logging(config_uri)
    settings = get_appsettings(config_uri)
    engine = engine_from_config(settings, 'sqlalchemy.')
    add_engine(engine)
    session = get_session()
    Base.metadata.create_all(engine)
    with transaction.manager:
        model = MyModel(name='one', value=1)
        session.add(model)

    def setUp(self):
        if self._init_sqla:
            try:
                engine = sqlahelper.get_engine()
            except: # pragma: no cover
                engine = sqlalchemy.engine_from_config(
                    {'sqlalchemy.url': 'sqlite://'})
                sqlahelper.add_engine(engine)

        self.init_pyramid()

        if self._init_ptah: # pragma: no cover
            self.init_ptah()

def sqla_initializing(ev):
    url = SQLA.url
    if url:
        engine_args = {}
        if SQLA.cache:
            engine_args['execution_options'] = \
                {'compiled_cache': SQL_compiled_cache}
        try:
            engine = sqlahelper.get_engine()
        except:
            engine = sqlalchemy.engine_from_config(
                {'sqlalchemy.url': url}, 'sqlalchemy.', **engine_args)
            sqlahelper.add_engine(engine)

def initializing(ev):
    # auth
    if not SECURITY.secret:
        SECURITY.secret = uuid.uuid4().get_hex()

    pname = SECURITY.policy
    if pname not in ('', 'no-policy'):
        policyFactory, attrs, kw = types[pname]

        settings = []
        for attr in attrs:
            settings.append(SECURITY.get(attr))

        kwargs = {'wild_domain': False}
        for attr in kw:
            kwargs[attr] = SECURITY.get(attr)

        policy = policyFactory(*settings, **kwargs)
        config.registry.registerUtility(policy, IAuthenticationPolicy)

    if SECURITY.authorization:
        config.registry.registerUtility(
            ACLAuthorizationPolicy(), IAuthorizationPolicy)

    # mail
    smtp_mailer = SMTPMailer(
        hostname = MAIL.host,
        port = MAIL.port,
        username = MAIL.username or None,
        password = MAIL.password or None,
        no_tls = MAIL.no_tls,
        force_tls = MAIL.force_tls,
        debug_smtp = MAIL.debug)

    MAIL.Mailer = DirectMailDelivery(smtp_mailer)
    MAIL.full_from_address = formataddr((MAIL.from_name, MAIL.from_address))

    # sqla
    url = SQLA.url
    if url:
        engine_args = {}
        if SQLA.cache:
            engine_args['execution_options'] = \
                {'compiled_cache': SQL_compiled_cache}
        try:
            engine = sqlahelper.get_engine()
        except:
            engine = sqlalchemy.engine_from_config(
                {'sqlalchemy.url': url}, 'sqlalchemy.', **engine_args)
            sqlahelper.add_engine(engine)