Produits Entr'ouvert » w.c.s.

    cursor = mock.Mock(**{'fetchall.return_value': False})

    pgconn = mock.Mock(**{'cursor.return_value': cursor})

    with mock.patch('psycopg2.connect', **{'return_value': pgconn}) as connect:

    cursor = mock.Mock(**{'fetchall.return_value': False})

    pgconn = mock.Mock(**{'cursor.return_value': cursor})

    with mock.patch('psycopg2.connect', **{'return_value': pgconn}) as connect:

import collections

import copy

import json

import os

import random

import shutil

import tempfile

import zipfile

import psycopg2

import pytest

from quixote import cleanup

from wcs.ctl.check_hobos import CmdCheckHobos

from wcs.publisher import WcsPublisher

from wcs.sql import cleanup_connection

from .utilities import clean_temporary_pub, create_temporary_pub

CONFIG = {

    "postgresql": {

        "createdb-connection-params": {"database": "postgres", "user": os.environ['USER']},

        "database-template-name": "%s",

        "user": os.environ['USER'],

    }

}

WCS_BASE_TENANT = 'wcsteststenant%d' % random.randint(0, 100000)

WCS_TENANT = '%s.net' % WCS_BASE_TENANT

WCS_DB_NAME = '%s_net' % WCS_BASE_TENANT

NEW_WCS_BASE_TENANT = 'wcsteststenant%d' % random.randint(0, 100000)

NEW_WCS_TENANT = '%s.net' % NEW_WCS_BASE_TENANT

NEW_WCS_DB_NAME = '%s_net' % NEW_WCS_BASE_TENANT

HOBO_JSON = {

    'services': [

        {

            'title': 'Hobo',

            'slug': 'hobo',

            'service-id': 'hobo',

            'base_url': 'http://hobo.example.net/',

            'saml-sp-metadata-url': 'http://hobo.example.net/accounts/mellon/metadata/',

        },

        {

            'service-id': 'authentic',

            'saml-idp-metadata-url': 'http://authentic.example.net/idp/saml2/metadata',

            'template_name': '',

            'variables': {},

            'title': 'Authentic',

            'base_url': 'http://authentic.example.net/',

            'id': 3,

            'slug': 'authentic',

            'secret_key': '12345',

        },

        {

            'service-id': 'wcs',

            'template_name': 'publik.zip',

            'variables': {'xxx': 'HELLO WORLD'},

            'title': 'Test wcs',

            'saml-sp-metadata-url': 'http://%s/saml/metadata' % WCS_TENANT,

            'base_url': 'http://%s/' % WCS_TENANT,

            'backoffice-menu-url': 'http://%s/backoffice/menu.json' % WCS_TENANT,

            'id': 1,

            'secret_key': 'eiue7aa10nt6e9*#jg2bsfvdgl)cr%4(tafibfjx9i$pgnfj#v',

            'slug': 'test-wcs',

        },

        {

            'service-id': 'combo',

            'template_name': 'portal-agent',

            'title': 'Portal Agents',

            'base_url': 'http://agents.example.net/',

            'secret_key': 'aaa',

        },

        {

            'service-id': 'combo',

            'template_name': 'portal-user',

            'title': 'Portal',

            'base_url': 'http://portal.example.net/',

            'secret_key': 'bbb',

        },

    ],

    'timestamp': '1431420355.31',

}

@pytest.fixture

def setuptest():

    cleanup_connection()

    createdb_cfg = CONFIG['postgresql'].get('createdb-connection-params')

    conn = psycopg2.connect(**createdb_cfg)

    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    cursor = conn.cursor()

    for dbname in (WCS_DB_NAME, NEW_WCS_DB_NAME):

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

    pub = create_temporary_pub()

    pub.cfg['language'] = {'language': 'en'}

    hobo_cmd = CmdCheckHobos()

    hobo_cmd.all_services = HOBO_JSON

    WcsPublisher.APP_DIR = tempfile.mkdtemp()

    skeleton_dir = os.path.join(WcsPublisher.APP_DIR, 'skeletons')

    os.mkdir(skeleton_dir)

    with open(os.path.join(skeleton_dir, 'publik.zip'), 'wb') as f:

        with zipfile.ZipFile(f, 'w') as z:

            z.writestr('config.json', json.dumps(CONFIG))

            z.writestr('site-options.cfg', '[options]\npostgresql = true')

    yield pub, hobo_cmd

    clean_temporary_pub()

    shutil.rmtree(WcsPublisher.APP_DIR)

    cleanup_connection()

    for dbname in (WCS_DB_NAME, NEW_WCS_DB_NAME):

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

    conn.close()

def database_exists(database):

    res = False

    cleanup_connection()

    createdb_cfg = CONFIG['postgresql'].get('createdb-connection-params')

    conn = psycopg2.connect(**createdb_cfg)

    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    cursor = conn.cursor()

    cursor.execute("SELECT 1 AS result FROM pg_database WHERE datname='%s'" % database)

    if cursor.fetchall():

        res = True

    conn.close()

    return res

def test_deploy(setuptest):

    _, hobo_cmd = setuptest

    assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert not database_exists(WCS_DB_NAME)

    assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))

    assert not database_exists(NEW_WCS_DB_NAME)

    cleanup()

    with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:

        fd.write(json.dumps(HOBO_JSON))

    hobo_cmd = CmdCheckHobos()

    base_options = {}

    sub_options_class = collections.namedtuple('Options', ['ignore_timestamp', 'redeploy', 'extra'])

    sub_options = sub_options_class(True, False, None)

    hobo_cmd.execute(

        base_options,

        sub_options,

        ['http://%s/' % WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],

    )

    assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert database_exists(WCS_DB_NAME)

    # deploy a new tenant

    with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:

        hobo_json = copy.deepcopy(HOBO_JSON)

        wcs_service = hobo_json['services'][2]

        wcs_service['saml-sp-metadata-url'] = ('http://%s/saml/metadata' % WCS_TENANT,)

        wcs_service['base_url'] = 'http://%s/' % NEW_WCS_TENANT

        wcs_service['backoffice-menu-url'] = 'http://%s/backoffice/menu.json' % NEW_WCS_TENANT

        fd.write(json.dumps(hobo_json))

    hobo_cmd.execute(

        base_options,

        sub_options,

        ['http://%s/' % NEW_WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],

    )

    assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert database_exists(WCS_DB_NAME)

    assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))

    assert database_exists(NEW_WCS_DB_NAME)

def test_deploy_url_change(setuptest):

    _, hobo_cmd = setuptest

    assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert not database_exists(WCS_DB_NAME)

    assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))

    assert not database_exists(NEW_WCS_DB_NAME)

    cleanup()

    with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:

        fd.write(json.dumps(HOBO_JSON))

    hobo_cmd = CmdCheckHobos()

    base_options = {}

    sub_options_class = collections.namedtuple('Options', ['ignore_timestamp', 'redeploy', 'extra'])

    sub_options = sub_options_class(True, False, None)

    hobo_cmd.execute(

        base_options,

        sub_options,

        ['http://%s/' % WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],

    )

    assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert database_exists(WCS_DB_NAME)

    # domain change request

    with open(os.path.join(WcsPublisher.APP_DIR, 'hobo.json'), 'w') as fd:

        hobo_json = copy.deepcopy(HOBO_JSON)

        wcs_service = hobo_json['services'][2]

        wcs_service['legacy_urls'] = [

            {

                'saml-sp-metadata-url': wcs_service['saml-sp-metadata-url'],

                'base_url': wcs_service['base_url'],

                'backoffice-menu-url': wcs_service['backoffice-menu-url'],

            }

        ]

        wcs_service['saml-sp-metadata-url'] = ('http://%s/saml/metadata' % WCS_TENANT,)

        wcs_service['base_url'] = 'http://%s/' % NEW_WCS_TENANT

        wcs_service['backoffice-menu-url'] = 'http://%s/backoffice/menu.json' % NEW_WCS_TENANT

        fd.write(json.dumps(hobo_json))

    hobo_cmd.execute(

        base_options,

        sub_options,

        ['http://%s/' % NEW_WCS_TENANT, os.path.join(WcsPublisher.APP_DIR, 'hobo.json')],

    )

    assert not os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', WCS_TENANT))

    assert not database_exists(WCS_DB_NAME)

    assert os.path.exists(os.path.join(WcsPublisher.APP_DIR, 'tenants', NEW_WCS_TENANT))

    assert database_exists(NEW_WCS_DB_NAME)

            pub.set_tenant_by_hostname(self.get_instance_path(service.get('base_url')), skip_sql=True)

            tenant_app_dir = os.path.join(global_tenants_dir, self.get_instance_path(service.get('base_url')))

            # check in legacy_urls for domain change

            for legacy_urls in service.get('legacy_urls', []):

                legacy_base_url = legacy_urls.get('base_url')

                if legacy_base_url.endswith('/'):  # wcs doesn't expect a trailing slash

                    legacy_base_url = legacy_base_url[:-1]

                legacy_instance_path = self.get_instance_path(legacy_base_url)

                try:

                    pub.set_tenant_by_hostname(legacy_instance_path, skip_sql=True)

                    # rename tenant directory

                    legacy_tenant_dir = os.path.join(global_tenants_dir, legacy_instance_path)

                    print('rename tenant directory %s to %s' % (legacy_tenant_dir, tenant_app_dir))

                    os.rename(legacy_tenant_dir, tenant_app_dir)

                    break

                except UnknownTenantError:

                    pass

                # new tenant

                os.mkdir(tenant_app_dir)

            pub.set_tenant_by_hostname(self.get_instance_path(service.get('base_url')))

        self.configure_sql(service, pub)

    def get_instance_path(self, base_url):

        parsed_url = urllib.parse.urlsplit(base_url)

        def get_domain_table_name(base_url):

            return base_url.replace('-', '_').replace('.', '_').replace('+', '_')

        domain_table_name = get_domain_table_name(self.get_instance_path(service.get('base_url')))

        legacy_domain_table_names = []

        for legacy_urls in service.get('legacy_urls', []):

            legacy_base_url = legacy_urls.get('base_url')

            if legacy_base_url.endswith('/'):  # wcs doesn't expect a trailing slash

                legacy_base_url = legacy_base_url[:-1]

            legacy_domain_table_names.append(get_domain_table_name(self.get_instance_path(legacy_base_url)))

        legacy_database_names = []

            for legacy_domain_table_name in legacy_domain_table_names:

                legacy_database_names.append(

                    self.normalize_database_name(

                        (database_template_name % legacy_domain_table_name).strip('_')

                    )

                )

        # check if database already exists

        cur.execute("SELECT 1 AS result FROM pg_database WHERE datname=%s", (database_name,))

        if cur.fetchall():

            # database exists, check if it is ready to go

            cur.execute(

                """SELECT table_name

            )

            if cur.fetchall():

                new_database = False

        else:

            # database does not exists, check if a legacy database exists

            for legacy_database_name in legacy_database_names:

                cur.execute("SELECT 1 AS result FROM pg_database WHERE datname=%s", (legacy_database_name,))

                    # legacy database exists, rename it

                    try:

                        # drop existing connections to DB

                        cur.execute(

                            '''SELECT pg_terminate_backend(pg_stat_activity.pid)

                            FROM pg_stat_activity

                            WHERE pg_stat_activity.datname = %s

                            AND pid <> pg_backend_pid();''',

                            (legacy_database_name,),

                        )

                        cur.execute("ALTER DATABASE %s RENAME TO %s" % (legacy_database_name, database_name))

                        new_database = False

                        break

                    except psycopg2.Error as e:

                        print(

                            'failed to rename database %s to %s (%s)'

                            % (legacy_database_name, database_name, psycopg2.errorcodes.lookup(e.pgcode)),

                            file=sys.stderr,

                        )

                        return

                try:

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

                except psycopg2.Error as e:

                    print(

                        'failed to create database (%s)' % psycopg2.errorcodes.lookup(e.pgcode),

                        file=sys.stderr,

                    )

                    return

        cur.close()