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

Benjamin Dauvergne, 19 décembre 2018 16:20

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

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

 tests/test_users.py | 41 +++++++++++++++++++++++++++++++++++-
 wcs/publisher.py    |  6 ++++++
 wcs/sql.py          | 12 +++++++++++
 wcs/users.py        | 51 +++++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 109 insertions(+), 1 deletion(-)

     import pytest
     from quixote import cleanup
     from quixote import cleanup, get_publisher
     from wcs import publisher
     from wcs import fields
-...
         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
         get_publisher().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

             super(WcsPublisher, cls).register_cronjobs()
             # every hour: check for global action timeouts
             cls.register_cronjob(CronJob(cls.apply_global_action_timeouts, hourly=True))
             cls.register_cronjob(CronJob(cls.clean_deleted_users, hourly=True))
         def is_using_postgresql(self):
             return bool(self.has_site_option('postgresql') and self.cfg.get('postgresql', {}))
-...
                 import sql
                 sql.cleanup_connection()
         def clean_deleted_users(self):
             for user_id in self.user_class.get_to_delete_ids():
                 self.user_class.remove_object(user_id)
     set_publisher_class(WcsPublisher)
     WcsPublisher.register_extra_dir(os.path.join(os.path.dirname(__file__), 'extra'))

             return objects
         @classmethod
         def get_to_delete_ids(cls):
             from .formdef import FormDef
             deleted_ids = set(str(user.id) for user in cls.select([Equal('deleted', True)]))
             for formdef in FormDef.select():
                 for formdata in formdef.data_class().select([Contains('user_id', deleted_ids)], iterator=True):
                     user_id = str(formdata.user_id)
                     if user_id in deleted_ids:
                         deleted_ids.remove(str(formdata.user_id))
             return deleted_ids
     class Session(SqlMixin, wcs.sessions.BasicSession):
         _table_name = 'sessions'

     # 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 get_to_delete_ids(cls):
             from .formdef import FormDef
             deleted_ids = set(str(user.id) for user in cls.select([st.Equal('deleted', True)]))
             for formdef in FormDef.select():
                 data_class = formdef.data_class()
                 formdatas = data_class.select(
                     [st.Contains('user_id', deleted_ids)],
                     iterator=True)
                 for formdata in formdatas:
                     user_id = str(formdata.user_id)
                     if user_id in deleted_ids:
                         deleted_ids.remove(str(formdata.user_id))
             return deleted_ids
         @classmethod
         def clean_deleted_users_sql(cls):
             from wcs.sql import AnyFormData
             from wcs.formdef import FormDef
             deleted = cls.select([st.Equal('deleted', True)])
             deleted_ids = set(str(u.id) for u in deleted)
             active_ids = set()
             if FormDef.count():
                 # without any FormDef global views are not created.
                 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(str(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(str(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'))
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

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