1from __future__ import absolute_import, unicode_literals 2 3from case import Mock 4 5from celery.bin.purge import purge 6from celery.five import WhateverIO 7 8 9class test_purge: 10 11 def test_run(self): 12 out = WhateverIO() 13 a = purge(app=self.app, stdout=out) 14 a._purge = Mock(name='_purge') 15 a._purge.return_value = 0 16 a.run(force=True) 17 assert 'No messages purged' in out.getvalue() 18 19 a._purge.return_value = 100 20 a.run(force=True) 21 assert '100 messages' in out.getvalue() 22 23 a.out = Mock(name='out') 24 a.ask = Mock(name='ask') 25 a.run(force=False) 26 a.ask.assert_called_with(a.warn_prompt, ('yes', 'no'), 'no') 27 a.ask.return_value = 'yes' 28 a.run(force=False) 29