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

    formdata1 = data_class()

    formdata1.user_id = user1.id

    evo = Evolution()

    evo.time = time.localtime()

    evo.who = user4.id

    formdata1.evolution = [evo]

    formdata1.store()

    carddata1 = card_data_class()

    carddata1.user_id = user3.id

    carddata1.store()

    assert User.count() == 4

    get_publisher().clean_deleted_users()

    assert User.count() == 3

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

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

        from wcs.carddef import CardDef

        from wcs.formdef import FormDef

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

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

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

            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''' % {

                'table': data_class._table_name

            }

            cur.execute(sql_statement)

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

            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 %(table)s.id IS NULL''' % {

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

            }

            cur.execute(sql_statement)

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

            conn.commit()

        cur.close()

        return deleted_ids

    @classmethod

    def get_to_delete_ids(cls):

        from .formdef import FormDef

        deleted_ids = {user.id for user in cls.select([st.NotNull('deleted_timestamp')])}

        active_ids = set()

        for formdef in FormDef.select():

            data_class = formdef.data_class()

            active_ids.update(

                formdata.user_id for formdata in data_class.select(iterator=True) if formdata.user_id

            )

        return deleted_ids - active_ids