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.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_reference_ids(cls):

        '''Retrieve ids of users reference in some carddata or formdata.'''

        from wcs.carddef import CardDef

        from wcs.formdef import FormDef

        referenced_ids = set()

        conn, cur = get_connection_and_cursor()

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

            data_class = objectdef.data_class()

            # referenced in form/card data.user_id

            sql_statement = (

                'SELECT CAST(data.user_id AS INTEGER) FROM %(table)s AS data WHERE data.user_id IS NOT NULL'

                % {

                    'table': data_class._table_name,

                }

            )

            cur.execute(sql_statement)

            referenced_ids.update(user_id for user_id, in cur.fetchall())

            conn.commit()

            # referenced in form/card data_evolution.who

            sql_statement = 'SELECT CAST(evolution.who AS INTEGER) FROM %(table)s AS evolution' % {

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

            }

            cur.execute(sql_statement)

            referenced_ids.update(user_id for user_id, in cur.fetchall())

            conn.commit()

            # referenced in form/card data.workflow_roles_array

            sql_statement = '''SELECT CAST(SUBSTRING(workflow_role.workflow_role FROM 7) AS INTEGER)

                                 FROM %(table)s AS data, UNNEST(data.workflow_roles_array) AS workflow_role

                                 WHERE SUBSTRING(workflow_role.workflow_role FROM 1 FOR 6) = '_user:' ''' % {

                'table': data_class._table_name,

            }

            cur.execute(sql_statement)

            referenced_ids.update(user_id for user_id, in cur.fetchall())

            conn.commit()

        return referenced_ids

    @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.'''

        # 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)

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

        conn.commit()

        cur.close()

        to_delete_ids = deleted_ids.difference(cls.get_reference_ids())

        return to_delete_ids