Produits Entr'ouvert » w.c.s.

import psycopg2

from wcs.ctl.delete_tenant import CmdDeleteTenant

from wcs.sql import get_connection_and_cursor, cleanup_connection

def test_delete_tenant():

    pub = create_temporary_pub(sql_mode=True)

    delete_cmd = CmdDeleteTenant()

    assert os.path.isdir(pub.app_dir)

    sub_options_class = collections.namedtuple('Options', ['force_drop'])

    sub_options = sub_options_class(False)

    delete_cmd.delete_tenant(pub, sub_options, [])

    assert not os.path.isdir(pub.app_dir)

    parent_dir = os.path.dirname(pub.app_dir)

    if not [filename for filename in os.listdir(parent_dir) if 'removed' in filename]:

        assert False

    conn, cur = get_connection_and_cursor()

    cur.execute("""SELECT schema_name

                FROM information_schema.schemata

                WHERE schema_name like '%removed%'""")

    assert len(cur.fetchall()) == 1

    clean_temporary_pub()

    pub = create_temporary_pub(sql_mode=True)

    sub_options = sub_options_class(True)

    delete_cmd.delete_tenant(pub, sub_options, [])

    conn, cur = get_connection_and_cursor(new=True)

    assert not os.path.isdir(pub.app_dir)

    cur.execute("""SELECT table_name

                FROM information_schema.tables

                WHERE table_schema = 'public' AND

                table_type = 'BASE TABLE'""")

    assert not cur.fetchall()

    cur.execute("""SELECT datname

                FROM pg_database

                WHERE datname = '%s'""" % pub.cfg['postgresql']['database'])

    assert cur.fetchall()

    clean_temporary_pub()

    pub = create_temporary_pub(sql_mode=True)

    cleanup_connection()

    sub_options = sub_options_class(True)

    pub.cfg['postgresql']['createdb-connection-params'] = {

        'user': pub.cfg['postgresql']['user'],

        'database': 'postgres'

    }

    delete_cmd.delete_tenant(pub, sub_options, [])

    pgconn = psycopg2.connect(**pub.cfg['postgresql']['createdb-connection-params'])

    cur = pgconn.cursor()

    cur.execute("""SELECT datname

                FROM pg_database

                WHERE datname = '%s'""" % pub.cfg['postgresql']['database'])

    assert not cur.fetchall()

    cur.close()

    pgconn.close()

    clean_temporary_pub()

    pub = create_temporary_pub(sql_mode=True)

    cleanup_connection()

    sub_options = sub_options_class(False)

    pub.cfg['postgresql']['createdb-connection-params'] = {

        'user': pub.cfg['postgresql']['user'],

        'database': 'postgres'

    }

    delete_cmd.delete_tenant(pub, sub_options, [])

    pgconn = psycopg2.connect(**pub.cfg['postgresql']['createdb-connection-params'])

    pgconn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

    cur = pgconn.cursor()

    cur.execute("""SELECT datname

                FROM pg_database

                WHERE datname like '%removed%'""")

    result = cur.fetchall()

    assert len(result) == 1

    #clean this db after test

    cur.execute("""DROP DATABASE %s""" % result[0][0])

    cur.execute("""SELECT datname

                FROM pg_database

                WHERE datname = '%s'""" % pub.cfg['postgresql']['database'])

    assert not cur.fetchall()

    cur.close()

    conn.close()

    clean_temporary_pub()

                cur.execute("""SELECT table_name

                            FROM information_schema.tables

                            WHERE table_schema = 'public' AND

                            table_type = 'BASE TABLE' AND

                            table_name = 'wcs_meta'""")

                if cur.fetchall():

                    new_database = False

#w.c.s. -  web application for online forms 

# Copyright (C) 2005-2014  Entr'ouvert 

# 

# 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 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 

# 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, see <http://www.gnu.org/licenses/>.

import os

import sys

import psycopg2

import psycopg2.errorcodes

from psycopg2 import OperationalError

from datetime import datetime

from shutil import rmtree

from qommon.ctl import Command, make_option

class CmdDeleteTenant(Command):

    name = 'delete_tenant'

    def __init__(self):

        Command.__init__(self, [

                make_option('--force-drop', action='store_true', default=False,

                           dest='force_drop'),

                ])

    def execute(self, base_options, sub_options, args):

        import publisher

        publisher.WcsPublisher.configure(self.config)

        pub = publisher.WcsPublisher.create_publisher(

                register_cron=False, register_tld_names=False)

        hostname = args[0]

        pub.app_dir = os.path.join(pub.app_dir, hostname)

        pub.set_config()

        self.delete_tenant(pub, sub_options, args)

    def delete_tenant(self, pub, options, args):

        postgresql_cfg = {}

        for k, v in pub.cfg['postgresql'].items():

            if v and isinstance(v, basestring):

                postgresql_cfg[k] = v

        createdb_cfg = pub.cfg['postgresql'].get('createdb-connection-params', {})

        createdb = True

        if not createdb_cfg:

            createdb_cfg = postgresql_cfg

            createdb = False

        try:

            pgconn = psycopg2.connect(**createdb_cfg)

        except psycopg2.Error as e:

            print >> sys.stderr, 'failed to connect to postgresql (%s)' % psycopg2.errorcodes.lookup(e.pgcode)

            return

        pgconn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

        cur = pgconn.cursor()

        try:

            dbname = pub.cfg['postgresql']['database']

            if createdb:

                if options.force_drop:

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

                else:

                    deletion_date = datetime.now().strftime('%Y%m%d_%H%M%S_%f')

                    cur.execute('ALTER DATABASE %s RENAME TO removed_%s_%s' % (dbname, deletion_date, dbname))

            else:

                cur.execute("""SELECT table_name

                            FROM information_schema.tables

                            WHERE table_schema = 'public' AND

                            table_type = 'BASE TABLE'""")

                tables_names = [x[0] for x in cur.fetchall()]

                if options.force_drop:

                    for table_name in tables_names:

                        cur.execute('DROP TABLE %s CASCADE' % table_name)

                else:

                    deletion_date = datetime.now().strftime('%Y%m%d_%H%M%S_%f')

                    schema_name = 'removed_%s_%s' % (deletion_date, dbname)

                    cur.execute("CREATE SCHEMA %s" % schema_name[:63])

                    for table_name in tables_names:

                        cur.execute('ALTER TABLE %s SET SCHEMA %s' %

                                    (table_name, schema_name[:63]))

            if options.force_drop:

                rmtree(pub.app_dir)

            else:

                os.rename(pub.app_dir, pub.app_dir + '_removed_%s.invalid' % deletion_date)

        except psycopg2.Error as e:

            if e.pgcode == psycopg2.errorcodes.DUPLICATE_TABLE:

                # when tables names are too long, postgresql truncates, which

                # can create some duplicates

                temp_date = datetime.now().strftime('%Y%m%d_%H%M%S_%f')

                cur.execute('ALTER TABLE %s RENAME TO removed_%s_%s' %

                            (table_name, temp_date, table_name))

            else:

                print >> sys.stderr, 'failed to alter database %s : (%s)' % (createdb_cfg['database'], psycopg2.errorcodes.lookup(e.pgcode))

                return

        cur.close()

CmdDeleteTenant.register()