0003-users-add-cronjob-to-delete-users-24430.patch

Benjamin Dauvergne, 04 octobre 2018 23:44

Voir les différences: en ligne côte à côte

Subject: [PATCH 3/4] users: add cronjob to delete users (#24430)

 tests/test_users.py | 39 +++++++++++++++++++++++++++++++++++++++
 wcs/users.py        | 41 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 80 insertions(+)

         with pytest.raises(AttributeError):
             user.xxx
     def test_clean_deleted_users():
         from wcs.formdef import FormDef
         from wcs.qommon import storage as st
         User = pub.user_class
         User.wipe()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'foobar'
         formdef.url_name = 'foobar'
         formdef.fields = []
         formdef.store()
         data_class = formdef.data_class()
         user1 = User()
         user1.name = 'Pierre'
         user1.deleted = True
         user1.store()
         user2 = User()
         user2.name = 'Jean'
         user2.deleted = True
         user2.store()
         formdata1 = data_class()
         formdata1.user_id = user1.id
         formdata1.store()
         assert User.count() == 2
         User.clean_deleted_users()
         assert User.count() == 1
         assert len(User.select([st.Equal('name', 'Pierre')])) == 1
         assert len(User.select([st.Equal('name', 'Jean')])) == 0

     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     from quixote import get_publisher
     from qommon import _
     from qommon.misc import simplify
     from qommon.storage import StorableObject
     from qommon import get_cfg
     from qommon.cron import CronJob
     import wcs.qommon.storage as st
     from qommon.substitution import Substitutions, invalidate_substitution_cache
     from qommon.publisher import get_publisher_class
     class User(StorableObject):
         _names = 'users'
-...
         def get_full_name(self):
             return self.display_name
         @classmethod
         def clean_deleted_users(cls):
             # really delete users without any form
             if get_publisher().is_using_postgresql():
                 cls.clean_deleted_users_sql()
             else:
                 cls.clean_deleted_users_pickle()
         @classmethod
         def clean_deleted_users_sql(cls):
             from wcs.sql import AnyFormData
             deleted = cls.select([st.Equal('deleted', True)])
             deleted_ids = set(u.id for u in deleted)
             active_ids = set(f.user_id for f in AnyFormData.select([st.Contains('user_id', deleted_ids)], iterator=True))
             to_delete_ids = deleted_ids - active_ids
             for user_id in to_delete_ids:
                 cls.remove_object(user_id)
         @classmethod
         def clean_deleted_users_pickle(cls):
             from wcs.formdef import FormDef
             deleted = cls.select([st.Equal('deleted', True)])
             deleted_ids = set(u.id for u in deleted)
             active_ids = set()
             for formdef in FormDef.select():
                 for formdata in formdef.data_class().select([st.Contains('user_id', deleted_ids)], iterator=True):
                     active_ids.add(formdata.user_id)
             to_delete_ids = deleted_ids - active_ids
             for user_id in to_delete_ids:
                 cls.remove_object(user_id)
     Substitutions.register('session_user_display_name', category=N_('User'), comment=N_('Session User Display Name'))
     Substitutions.register('session_user_email', category=N_('User'), comment=N_('Session User Email'))
     Substitutions.register_dynamic_source(User)
     if get_publisher_class():
         # clean deleted users hourly
         get_publisher_class().register_cronjob(CronJob(User.clean_deleted_users, minutes=[20], hours=[6]))
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0003-users-add-cronjob-to-delete-users-24430.patch