equal
deleted
inserted
replaced
15 # |
15 # |
16 # You should have received a copy of the GNU Lesser General Public License along |
16 # You should have received a copy of the GNU Lesser General Public License along |
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
18 """Tests for cubicweb.server.utils module.""" |
18 """Tests for cubicweb.server.utils module.""" |
19 |
19 |
20 from unittest import TestCase |
20 from cubicweb.devtools import testlib |
21 |
|
22 from cubicweb.server import utils |
21 from cubicweb.server import utils |
23 |
22 |
24 |
23 |
25 class UtilsTC(TestCase): |
24 class UtilsTC(testlib.BaseTestCase): |
26 |
25 |
27 def test_crypt(self): |
26 def test_crypt(self): |
28 for hash in ( |
27 for hash in ( |
29 utils.crypt_password('xxx'), # default sha512 |
28 utils.crypt_password('xxx'), # default sha512 |
30 b'ab$5UsKFxRKKN.d8iBIFBnQ80', # custom md5 |
29 b'ab$5UsKFxRKKN.d8iBIFBnQ80', # custom md5 |
38 |
37 |
39 # accept any password for empty hashes (is it a good idea?) |
38 # accept any password for empty hashes (is it a good idea?) |
40 self.assertEqual(utils.crypt_password('xxx', ''), '') |
39 self.assertEqual(utils.crypt_password('xxx', ''), '') |
41 self.assertEqual(utils.crypt_password('yyy', ''), '') |
40 self.assertEqual(utils.crypt_password('yyy', ''), '') |
42 |
41 |
|
42 def test_schedule_periodic_task(self): |
|
43 scheduler = utils.scheduler() |
|
44 this = [] |
|
45 |
|
46 def fill_this(x): |
|
47 this.append(x) |
|
48 if len(this) > 2: |
|
49 raise SystemExit() |
|
50 elif len(this) > 1: |
|
51 raise RuntimeError() |
|
52 |
|
53 event = utils.schedule_periodic_task(scheduler, 0.01, fill_this, 1) |
|
54 self.assertEqual(event.action.__name__, 'fill_this') |
|
55 self.assertEqual(len(scheduler.queue), 1) |
|
56 |
|
57 with self.assertLogs('cubicweb.scheduler', level='ERROR') as cm: |
|
58 scheduler.run() |
|
59 self.assertEqual(this, [1] * 3) |
|
60 self.assertEqual(len(cm.output), 2) |
|
61 self.assertIn('Unhandled exception in periodic task "fill_this"', |
|
62 cm.output[0]) |
|
63 self.assertIn('"fill_this" not re-scheduled', cm.output[1]) |
|
64 |
43 |
65 |
44 if __name__ == '__main__': |
66 if __name__ == '__main__': |
45 import unittest |
67 import unittest |
46 unittest.main() |
68 unittest.main() |