<s> from djcelery_transactions import task <EOL> from celery . registry import tasks <EOL> from django . db import transaction <EOL> from django . test import TransactionTestCase <EOL> my_global = [ ] <EOL> marker = object ( ) <EOL> @ task <EOL> def my_task ( ) : <EOL> my_global . append ( marker ) <EOL> tasks . register ( my_task ) <EOL> class SpecificException ( Exception ) : <EOL> pass <EOL> class DjangoCeleryTestCase ( TransactionTestCase ) : <EOL> """<STR_LIT>""" <EOL> def tearDown ( self ) : <EOL> my_global [ : ] = [ ] <EOL> def test_commited_transaction_fire_task ( self ) : <EOL> """<STR_LIT>""" <EOL> @ transaction . commit_on_success <EOL> def do_something ( ) : <EOL> my_task . delay ( ) <EOL> do_something ( ) <EOL> self . assertTrue ( my_global [ <NUM_LIT:0> ] is marker ) <EOL> def test_rollbacked_transaction_discard_task ( self ) : <EOL> """<STR_LIT>""" <EOL> @ transaction . commit_on_success <EOL> def do_something ( ) : <EOL> my_task . delay ( ) <EOL> raise SpecificException <EOL> try : <EOL> do_something ( ) <EOL> except SpecificException : <EOL> self . assertFalse ( my_global ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) </s>
