0001-users-add-cronjob-to-delete-users-24430.patch
tests/test_users.py | ||
---|---|---|
1 |
import datetime |
|
1 | 2 |
import shutil |
2 | 3 | |
3 | 4 |
import pytest |
4 |
from quixote import cleanup |
|
5 |
from quixote import cleanup, get_publisher
|
|
5 | 6 | |
6 | 7 |
from wcs import fields |
7 | 8 |
from wcs.variables import LazyUser |
... | ... | |
92 | 93 |
with pytest.raises(AttributeError): |
93 | 94 |
# noqa pylint: disable=pointless-statement |
94 | 95 |
user.xxx |
96 | ||
97 | ||
98 |
def test_clean_deleted_users(): |
|
99 |
from wcs.formdef import FormDef |
|
100 |
from wcs.qommon import storage as st |
|
101 | ||
102 |
User = pub.user_class |
|
103 | ||
104 |
User.wipe() |
|
105 |
FormDef.wipe() |
|
106 | ||
107 |
formdef = FormDef() |
|
108 |
formdef.name = 'foobar' |
|
109 |
formdef.url_name = 'foobar' |
|
110 |
formdef.fields = [] |
|
111 |
formdef.store() |
|
112 |
data_class = formdef.data_class() |
|
113 | ||
114 |
user1 = User() |
|
115 |
user1.name = 'Pierre' |
|
116 |
user1.deleted_timestamp = datetime.datetime.now() |
|
117 |
user1.store() |
|
118 | ||
119 |
user2 = User() |
|
120 |
user2.name = 'Jean' |
|
121 |
user2.deleted_timestamp = datetime.datetime.now() |
|
122 |
user2.store() |
|
123 | ||
124 |
formdata1 = data_class() |
|
125 |
formdata1.user_id = user1.id |
|
126 |
formdata1.store() |
|
127 | ||
128 |
assert User.count() == 2 |
|
129 | ||
130 |
get_publisher().clean_deleted_users() |
|
131 | ||
132 |
assert User.count() == 1 |
|
133 | ||
134 |
assert len(User.select([st.Equal('name', 'Pierre')])) == 1 |
|
135 |
assert len(User.select([st.Equal('name', 'Jean')])) == 0 |
|
136 | ||
137 |
data_class.wipe() |
|
138 | ||
139 |
get_publisher().clean_deleted_users() |
|
140 | ||
141 |
assert User.count() == 0 |
wcs/publisher.py | ||
---|---|---|
123 | 123 |
cls.register_cronjob( |
124 | 124 |
CronJob(cls.update_deprecations_report, name='update_deprecations_report', hours=[2], minutes=[0]) |
125 | 125 |
) |
126 |
# once a day delete users without any formdata |
|
127 |
cls.register_cronjob(CronJob(cls.clean_deleted_users, name='clean_deleted_users', minutes=[0])) |
|
126 | 128 |
# other jobs |
127 | 129 |
data_sources.register_cronjob() |
128 | 130 |
formdef.register_cronjobs() |
... | ... | |
508 | 510 |
return value_.get_value() |
509 | 511 |
return value_ |
510 | 512 | |
513 |
def clean_deleted_users(self): |
|
514 |
for user_id in self.user_class.get_to_delete_ids(): |
|
515 |
self.user_class.remove_object(user_id) |
|
516 | ||
511 | 517 | |
512 | 518 |
set_publisher_class(WcsPublisher) |
513 | 519 |
WcsPublisher.register_extra_dir(os.path.join(os.path.dirname(__file__), 'extra')) |
wcs/sql.py | ||
---|---|---|
3248 | 3248 | |
3249 | 3249 |
return objects |
3250 | 3250 | |
3251 |
@classmethod |
|
3252 |
def get_to_delete_ids(cls): |
|
3253 |
conn, cur = get_connection_and_cursor() |
|
3254 |
sql_statement = '''SELECT users.id |
|
3255 |
FROM users LEFT JOIN wcs_all_forms ON users.id = CAST(wcs_all_forms.user_id AS INTEGER) |
|
3256 |
WHERE users.deleted_timestamp IS NOT NULL |
|
3257 |
AND wcs_all_forms.id IS NULL''' |
|
3258 |
cur.execute(sql_statement) |
|
3259 |
deleted_ids = [user_id for user_id, in cur.fetchall()] |
|
3260 |
conn.commit() |
|
3261 |
cur.close() |
|
3262 |
return deleted_ids |
|
3263 | ||
3251 | 3264 | |
3252 | 3265 |
class Role(SqlMixin, wcs.roles.Role): |
3253 | 3266 |
_table_name = 'roles' |
wcs/users.py | ||
---|---|---|
320 | 320 |
def get_full_name(self): |
321 | 321 |
return self.display_name |
322 | 322 | |
323 |
@classmethod |
|
324 |
def get_to_delete_ids(cls): |
|
325 |
from .formdef import FormDef |
|
326 | ||
327 |
deleted_ids = {user.id for user in cls.select([st.NotNull('deleted_timestamp')])} |
|
328 | ||
329 |
active_ids = set() |
|
330 |
for formdef in FormDef.select(): |
|
331 |
data_class = formdef.data_class() |
|
332 |
active_ids.update( |
|
333 |
formdata.user_id for formdata in data_class.select(iterator=True) if formdata.user_id |
|
334 |
) |
|
335 | ||
336 |
return deleted_ids - active_ids |
|
337 | ||
323 | 338 | |
324 | 339 |
Substitutions.register( |
325 | 340 |
'session_user_display_name', |
326 |
- |