Produits Entr'ouvert » w.c.s.

import os

import psycopg2

import pytest

import random

from django.core.management import get_commands

from django.core.management import call_command

from django.core.management.base import CommandError

from utilities import clean_temporary_pub

from utilities import create_temporary_pub

from utilities import force_connections_close

from wcs.sql import cleanup_connection

@pytest.fixture(scope='module')

def cursor():

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

    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    cur = conn.cursor()

    yield cur

    cur.close()

@pytest.fixture

def database(cursor):

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

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

    yield dbname

    cleanup_connection()

    cursor.execute('DROP DATABASE %s' % dbname)

@pytest.fixture(params=['pickle', 'sql'])

def pub(request):

    pub = create_temporary_pub(sql_mode=(request.param == 'sql'))

    yield pub

    cleanup_connection()

    force_connections_close()

    clean_temporary_pub()

def test_command_exists():

    assert 'configure_connection' in get_commands()

def test_unknown_publisher():

    with pytest.raises(CommandError) as excinfo:

        call_command('configure_connection', '-d', 'unknown.net',

                     '--database', 'foobar')

    assert excinfo.value.message == 'unknown tenant'

def test_failing_connection(pub):

    with pytest.raises(psycopg2.OperationalError) as excinfo:

        call_command('configure_connection', '-d', 'example.net',

                     '--database', 'foobar', '--port', '666')

    assert 'could not connect' in excinfo.value.message

def test_database_does_not_exist(pub):

    new_database = 'test_{}'.format(random.randint(1000, 9999))

    with pytest.raises(psycopg2.OperationalError) as excinfo:

        call_command('configure_connection', '-d', 'example.net',

                     '--database', new_database)

    assert 'does not exist' in excinfo.value.message

def test_setup_database(pub, database):

    call_command('configure_connection', '-d', 'example.net',

                 '--database', database)

    pub.set_config()

    assert pub.cfg['postgresql'].get('database') == database

import os

from django.core.management.base import BaseCommand, CommandError

from qommon.publisher import get_publisher_class

class WcsTenantCommand(BaseCommand):

    def create_parser(self, *args):

        parser = super(WcsTenantCommand, self).create_parser(*args)

        parser.add_argument('-d', '--domain', required=True)

        return parser

    def execute(self, *args, **kwargs):

        self.publisher = self.get_publisher(kwargs['domain'])

        super(WcsTenantCommand, self).execute(*args, **kwargs)

    def get_publisher(self, domain):

        publisher_class = get_publisher_class()

        publisher = publisher_class.create_publisher()

        if domain not in publisher.get_tenants():

            raise CommandError('unknown tenant')

        publisher.app_dir = os.path.join(publisher.app_dir, domain)

        publisher.set_config()

        return publisher

import os

from wcs.ctl.management.commands import WcsTenantCommand

from wcs.sql import get_connection

from psycopg2 import OperationalError

class Command(WcsTenantCommand):

    help = 'Configure postgresql connection parameters.'

    def add_arguments(self, parser):

        parser.add_argument('--host')

        parser.add_argument('--port', type=int)

        parser.add_argument('--database', required=True)

        parser.add_argument('--user')

        parser.add_argument('--password')

        parser.add_argument('-f', '--force')

    def handle(self, **kwargs):

        self.setup_connection(**kwargs)

        self.publisher.write_cfg()

        self.publisher.load_site_options()

        self.enable()

        self.publisher.cleanup()

    def enable(self):

        if not self.publisher.site_options.has_option('options', 'postgresql'):

            self.publisher.site_options.set('options', 'postgresql', 'true')

            options_file = os.path.join(self.publisher.app_dir,

                                        'site-options.cfg')

            self.publisher.site_options.write(open(options_file, 'w'))

    def setup_connection(self, **kwargs):

        options = {}

        for k in ['host', 'port', 'database', 'user', 'password']:

            if k in kwargs:

                options[k] = kwargs.get(k)

        self.publisher.cfg['postgresql'] = options

        if not kwargs.get('force'):

            conn = self.test_connection()

            if isinstance(conn, OperationalError):

                raise(conn)

    def test_connection(self):

        try:

            return get_connection(new=True)

        except OperationalError as e:

            return e