Produits Entr'ouvert » w.c.s. » Au quotidien

import configparser

import os

import random

import shutil

import tempfile

import psycopg2

from webtest import TestApp

from quixote import cleanup, get_publisher

from django.conf import settings

from wcs.qommon.publisher import set_publisher_class

import wcs

import wcs.wsgi

from wcs.qommon.http_request import HTTPRequest

from wcs import publisher, compat, sql

import wcs.middleware

wcs.middleware.AfterJobsMiddleware.ASYNC = False

def create_temporary_pub():

    config = configparser.ConfigParser()

    APP_DIR = tempfile.mkdtemp()

    compat.CompatWcsPublisher.APP_DIR = APP_DIR

    compat.CompatWcsPublisher.DATA_DIR = os.path.abspath(

        os.path.join(os.path.dirname(wcs.__file__), '..', 'data')

    )

    compat.CompatWcsPublisher.cronjobs = None

    config.add_section('extra')

    config.set('extra', 'auquotidien', os.path.join(os.path.dirname(__file__), '..', 'auquotidien'))

    compat.CompatWcsPublisher._initialized = False

    compat.CompatWcsPublisher.configure(config)

    compat.CompatWcsPublisher.init_publisher_class()

    pub = compat.CompatWcsPublisher.create_publisher()

    pub.app_dir = os.path.join(APP_DIR, 'example.net')

    os.mkdir(pub.app_dir)

    # set classes

    pub.user_class = sql.SqlUser

    pub.role_class = sql.Role

    pub.token_class = sql.Token

    pub.tracking_code_class = sql.TrackingCode

    pub.session_class = sql.Session

    pub.custom_view_class = sql.CustomView

    pub.snapshot_class = sql.Snapshot

    pub.loggederror_class = sql.LoggedError

    conn = psycopg2.connect(user=os.environ['USER'], dbname='postgres')

    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    i = 0

    while True:

        dbname = 'wcstests%d' % random.randint(0, 100000)

        try:

            cur = conn.cursor()

            cur.execute('CREATE DATABASE %s' % dbname)

            break

        except psycopg2.Error:

            if i < 5:

                i += 1

                continue

            raise

        finally:

            cur.close()

    pub.cfg['postgresql'] = {'database': dbname, 'user': os.environ['USER']}

    pub.write_cfg()

    sql.do_user_table()

    sql.do_role_table()

    sql.do_tokens_table()

    sql.do_tracking_code_table()

    sql.do_session_table()

    sql.do_transient_data_table()

    sql.do_custom_views_table()

    sql.do_snapshots_table()

    sql.do_loggederrors_table()

    sql.do_meta_table()

    sql.init_global_table()

    conn.close()

    return pub

def clean_temporary_pub():

    pub = get_publisher()

    conn = psycopg2.connect(user=os.environ['USER'], dbname='postgres')

    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    try:

        cur = conn.cursor()

        cur.execute('DROP DATABASE %s' % pub.cfg['postgresql']['database'])

        cur.close()

    except psycopg2.Error as e:

        print(e)

    shutil.rmtree(pub.app_dir)

    pub.cleanup()

def get_app(pub, https=False):

    extra_environ = {'HTTP_HOST': 'example.net', 'REMOTE_ADDR': '127.0.0.1'}

    if https:

        settings.SECURE_PROXY_SSL_HEADER = ('HTTPS', 'on')

        extra_environ['HTTPS'] = 'on'

    else:

        extra_environ['HTTPS'] = 'off'

    return TestApp(wcs.wsgi.application, extra_environ=extra_environ)

def login(app, username='admin', password='admin'):

    login_page = app.get('/login/')

    login_form = login_page.forms['login-form']

    login_form['username'] = username

    login_form['password'] = password

    resp = login_form.submit()

    assert resp.status_int == 302

    return app