<s> import os <EOL> import time <EOL> import times <EOL> from datetime import datetime <EOL> from django . test import TransactionTestCase , TestCase <EOL> from django . utils . timezone import utc <EOL> from nose2 . tools import params <EOL> from pq import Queue <EOL> from pq . queue import get_failed_queue <EOL> from pq . worker import Worker <EOL> from pq . job import Job <EOL> from . fixtures import say_hello , div_by_zero , create_file_after_timeout <EOL> class TestWorker ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . fooq , self . barq = Queue ( '<STR_LIT:foo>' ) , Queue ( '<STR_LIT:bar>' ) <EOL> def test_create_worker ( self ) : <EOL> """<STR_LIT>""" <EOL> w = Worker . create ( [ self . fooq , self . barq ] ) <EOL> self . assertEquals ( w . queues , [ self . fooq , self . barq ] ) <EOL> class TestWorkNoJobs ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . fooq , self . barq = Queue ( '<STR_LIT:foo>' ) , Queue ( '<STR_LIT:bar>' ) <EOL> self . w = Worker . create ( [ self . fooq , self . barq ] ) <EOL> def test_work_no_jobs ( self ) : <EOL> self . assertEquals ( self . w . work ( burst = True ) , False , <EOL> '<STR_LIT>' ) <EOL> class TestWorkerWithJobs ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . fooq , self . barq = Queue ( '<STR_LIT:foo>' ) , Queue ( '<STR_LIT:bar>' ) <EOL> self . w = Worker . create ( [ self . fooq , self . barq ] ) <EOL> self . fooq . enqueue ( say_hello , name = '<STR_LIT>' ) <EOL> def test_worker_with_jobs ( self ) : <EOL> self . assertEquals ( self . w . work ( burst = True ) , True , <EOL> '<STR_LIT>' ) <EOL> class TestWorkViaStringArg ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( '<STR_LIT:foo>' ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> self . job = self . q . enqueue ( '<STR_LIT>' , name = '<STR_LIT>' ) <EOL> def test_work_via_string_argument ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEquals ( self . w . work ( burst = True ) , True , <EOL> '<STR_LIT>' ) <EOL> job = Job . objects . get ( id = self . job . id ) <EOL> self . assertEquals ( job . result , '<STR_LIT>' ) <EOL> class TestWorkIsUnreadable ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . q . save ( ) <EOL> self . fq = get_failed_queue ( ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> def test_work_is_unreadable ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEquals ( self . fq . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . q . count , <NUM_LIT:0> ) <EOL> job = Job . create ( func = div_by_zero , args = ( <NUM_LIT:3> , ) ) <EOL> job . save ( ) <EOL> job . instance = '<STR_LIT>' <EOL> job . queue = self . q <EOL> job . save ( ) <EOL> self . assertEquals ( self . q . count , <NUM_LIT:1> ) <EOL> self . w . work ( burst = True ) <EOL> self . assertEquals ( self . q . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . fq . count , <NUM_LIT:1> ) <EOL> class TestWorkFails ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . fq = get_failed_queue ( ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> self . job = self . q . enqueue ( div_by_zero ) <EOL> self . enqueued_at = self . job . enqueued_at <EOL> def test_work_fails ( self ) : <EOL> """<STR_LIT>""" <EOL> self . w . work ( burst = True ) <EOL> self . assertEquals ( self . q . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . fq . count , <NUM_LIT:1> ) <EOL> job = Job . objects . get ( id = self . job . id ) <EOL> self . assertEquals ( job . origin , self . q . name ) <EOL> self . assertEquals ( job . enqueued_at , self . enqueued_at ) <EOL> self . assertIsNotNone ( job . exc_info ) <EOL> class TestWorkerCustomExcHandling ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . fq = get_failed_queue ( ) <EOL> def black_hole ( job , * exc_info ) : <EOL> return False <EOL> self . black_hole = black_hole <EOL> self . job = self . q . enqueue ( div_by_zero ) <EOL> def test_custom_exc_handling ( self ) : <EOL> """<STR_LIT>""" <EOL> w = Worker . create ( [ self . q ] , exc_handler = self . black_hole ) <EOL> w . work ( burst = True ) <EOL> self . assertEquals ( self . q . count , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . fq . count , <NUM_LIT:0> ) <EOL> job = Job . objects . get ( id = self . job . id ) <EOL> self . assertEquals ( job . status , Job . FAILED ) <EOL> class TestWorkerTimeouts ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . sentinel_file = '<STR_LIT>' <EOL> self . q = Queue ( ) <EOL> self . fq = get_failed_queue ( ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> def test_timeouts ( self ) : <EOL> """<STR_LIT>""" <EOL> jobr = self . q . enqueue ( <EOL> create_file_after_timeout , <EOL> args = ( self . sentinel_file , <NUM_LIT:4> ) , <EOL> timeout = <NUM_LIT:1> ) <EOL> self . assertEquals ( os . path . exists ( self . sentinel_file ) , False ) <EOL> self . w . work ( burst = True ) <EOL> self . assertEquals ( os . path . exists ( self . sentinel_file ) , False ) <EOL> job = Job . objects . get ( id = jobr . id ) <EOL> self . assertIn ( '<STR_LIT>' , job . exc_info ) <EOL> def tearDown ( self ) : <EOL> try : <EOL> os . unlink ( self . sentinel_file ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:2> : <EOL> pass <EOL> class TestWorkerSetsResultTTL ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> @ params ( ( <NUM_LIT:10> , <NUM_LIT:10> ) , ( - <NUM_LIT:1> , - <NUM_LIT:1> ) , ( <NUM_LIT:0> , None ) ) <EOL> def test_worker_sets_result_ttl ( self , ttl , outcome ) : <EOL> """<STR_LIT>""" <EOL> job = self . q . enqueue ( say_hello , args = ( '<STR_LIT>' , ) , result_ttl = ttl ) <EOL> self . w . work ( burst = True ) <EOL> try : <EOL> rjob = Job . objects . get ( id = job . id ) <EOL> result_ttl = rjob . result_ttl <EOL> except Job . DoesNotExist : <EOL> result_ttl = None <EOL> self . assertEqual ( result_ttl , outcome ) <EOL> class TestWorkerDeletesExpiredTTL ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . w = Worker . create ( [ self . q ] ) <EOL> self . job = self . q . enqueue ( say_hello , args = ( '<STR_LIT>' , ) , result_ttl = <NUM_LIT:1> ) <EOL> self . w . work ( burst = True ) <EOL> def test_worker_deletes_expired_ttl ( self ) : <EOL> """<STR_LIT>""" <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . w . work ( burst = True ) <EOL> with self . assertRaises ( Job . DoesNotExist ) as exc : <EOL> rjob = Job . objects . get ( id = self . job . id ) <EOL> class TestWorkerDequeueTimeout ( TransactionTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . w = Worker . create ( [ self . q ] , <EOL> expires_after = <NUM_LIT:1> , <EOL> default_worker_ttl = <NUM_LIT:1> ) <EOL> def test_worker_dequeue_timeout ( self ) : <EOL> self . w . work ( ) <EOL> self . assertEqual ( self . w . _expires_after , - <NUM_LIT:1> ) <EOL> class TestRegisterHeartbeat ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( ) <EOL> self . w = Worker . create ( [ self . q ] , name = '<STR_LIT>' ) <EOL> self . w . heartbeat = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , tzinfo = utc ) <EOL> def test_worker_register_heartbeat ( self ) : <EOL> self . w . register_heartbeat ( timeout = <NUM_LIT:0> ) </s>
