Produits Entr'ouvert » w.c.s.

import datetime

import time

from quixote import cleanup, get_publisher

def test_clean_deleted_users():

    from wcs.carddef import CardDef

    from wcs.formdef import FormDef

    from wcs.workflows import Evolution

    User = pub.user_class

    User.wipe()

    FormDef.wipe()

    CardDef.wipe()

    formdef = FormDef()

    formdef.name = 'foobar'

    formdef.url_name = 'foobar'

    formdef.fields = []

    formdef.store()

    data_class = formdef.data_class()

    carddef = CardDef()

    carddef.name = 'barfoo'

    carddef.url_name = 'barfoo'

    carddef.fields = []

    carddef.store()

    card_data_class = carddef.data_class()

    user1 = User()

    user1.name = 'Pierre'

    user1.deleted_timestamp = datetime.datetime.now()

    user1.store()

    user2 = User()

    user2.name = 'Jean'

    user2.deleted_timestamp = datetime.datetime.now()

    user2.store()

    user3 = User()

    user3.name = 'Michel'

    user3.deleted_timestamp = datetime.datetime.now()

    user3.store()

    user4 = User()

    user4.name = 'Martin'

    user4.deleted_timestamp = datetime.datetime.now()

    user4.store()

    user5 = User()

    user5.name = 'Alain'

    user5.deleted_timestamp = datetime.datetime.now()

    user5.store()

    formdata1 = data_class()

    formdata1.user_id = user1.id

    evo = Evolution()

    evo.time = time.localtime()

    evo.who = user4.id

    formdata1.evolution = [evo]

    formdata1.workflow_roles = {'_received': '_user:%s' % user5.id}

    formdata1.store()

    carddata1 = card_data_class()

    carddata1.user_id = user3.id

    carddata1.store()

    assert User.count() == 5

    get_publisher().clean_deleted_users()

    assert User.count() == 4

    assert {user.name for user in User.select()} == {'Pierre', 'Michel', 'Martin', 'Alain'}

    data_class.wipe()

    card_data_class.wipe()

    get_publisher().clean_deleted_users()

    assert User.count() == 0

        # once a day delete users without any formdata

        cls.register_cronjob(CronJob(cls.clean_deleted_users, name='clean_deleted_users', minutes=[0]))

    def clean_deleted_users(self):

        for user_id in self.user_class.get_to_delete_ids():

            self.user_class.remove_object(user_id)

    @classmethod

    def get_to_delete_ids(cls):

        '''Retrieve ids of users which are deleted on the IdP and are no more referenced by any form or card.'''

        from wcs.carddef import CardDef

        from wcs.formdef import FormDef

        # fetch marked as deleted users

        conn, cur = get_connection_and_cursor()

        sql_statement = 'SELECT users.id FROM users WHERE users.deleted_timestamp IS NOT NULL'

        cur.execute(sql_statement)

        to_delete_ids = {user_id for user_id, in cur.fetchall()}

        conn.commit()

        # iteratively reduce to_delete_ids by retaining only unreferenced users

        for carddef in CardDef.select() + FormDef.select():

            # keep only user.id not present in form/card_data.user_id

            data_class = carddef.data_class()

            sql_statement = '''SELECT users.id

                                 FROM users LEFT JOIN %(table)s ON users.id = CAST(%(table)s.user_id AS INTEGER)

                                WHERE users.deleted_timestamp IS NOT NULL

                                AND %(table)s.id IS NULL

                                AND users.id IN %%(to_delete_ids)s

                                ''' % {

                'table': data_class._table_name

            }

            cur.execute(sql_statement, {'to_delete_ids': tuple(to_delete_ids)})

            to_delete_ids = {user_id for user_id, in cur.fetchall()}

            # keep only user.id not present in form/card_evolutions.who columns

            sql_statement = '''SELECT users.id

                                 FROM users LEFT JOIN %(table)s ON users.id = CAST(%(table)s.who AS INTEGER)

                                WHERE users.deleted_timestamp IS NOT NULL

                                AND users.id IN %%(to_delete_ids)s

                                AND %(table)s.id IS NULL''' % {

                'table': '%s_evolutions' % data_class._table_name,

            }

            cur.execute(sql_statement, {'to_delete_ids': tuple(to_delete_ids)})

            to_delete_ids = {user_id for user_id, in cur.fetchall()}

            # keep only user.id not present in form/card_data.workflow_roles_array

            sql_statement = '''SELECT users.id

                                 FROM %(table)s AS data

                                 JOIN UNNEST(data.workflow_roles_array) AS workflow_role ON 'true'

                                 RIGHT JOIN users ON workflow_role.workflow_role = '_user:' || users.id

                                 WHERE users.id IN %%(to_delete_ids)s

                                 AND users.deleted_timestamp IS NOT NULL AND data.id IS NULL''' % {

                'table': data_class._table_name,

            }

            cur.execute(sql_statement, {'to_delete_ids': tuple(to_delete_ids)})

            to_delete_ids = {user_id for user_id, in cur.fetchall()}

            conn.commit()

        cur.close()

        return to_delete_ids