<s> import time <EOL> from datetime import timedelta , datetime <EOL> from django . test import TestCase , TransactionTestCase <EOL> from django . utils . timezone import utc , now <EOL> from dateutil . relativedelta import relativedelta <EOL> from nose2 . tools import params <EOL> from pq . job import Job <EOL> from pq import Queue <EOL> from . fixtures import some_calculation , say_hello , Calculator , do_nothing <EOL> class TestJobCreation ( TestCase ) : <EOL> def test_job_create ( self ) : <EOL> job = Job . create ( func = some_calculation , args = ( <NUM_LIT:3> , <NUM_LIT:4> ) , kwargs = dict ( z = <NUM_LIT:2> ) ) <EOL> self . assertIsNotNone ( job . created_at ) <EOL> self . assertIsNotNone ( job . description ) <EOL> self . assertIsNone ( job . instance ) <EOL> self . assertEquals ( job . func , some_calculation ) <EOL> self . assertEquals ( job . args , ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> self . assertEquals ( job . kwargs , { '<STR_LIT:z>' : <NUM_LIT:2> } ) <EOL> self . assertIsNone ( job . origin ) <EOL> self . assertIsNone ( job . enqueued_at ) <EOL> def test_create_instance_method_job ( self ) : <EOL> """<STR_LIT>""" <EOL> c = Calculator ( <NUM_LIT:2> ) <EOL> job = Job . create ( func = c . calculate , args = ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> self . assertEquals ( job . func , c . calculate ) <EOL> self . assertEquals ( job . instance , c ) <EOL> self . assertEquals ( job . args , ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> def test_create_job_from_string_function ( self ) : <EOL> """<STR_LIT>""" <EOL> job = Job . create ( func = '<STR_LIT>' , args = ( '<STR_LIT>' , ) ) <EOL> self . assertEquals ( job . func , say_hello ) <EOL> self . assertIsNone ( job . instance ) <EOL> self . assertEquals ( job . args , ( '<STR_LIT>' , ) ) <EOL> class TestScheduledJobCreation ( TestCase ) : <EOL> def test_scheduled_job_create ( self ) : <EOL> """<STR_LIT>""" <EOL> dt = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , tzinfo = utc ) <EOL> rd = relativedelta ( months = <NUM_LIT:1> , days = - <NUM_LIT:1> ) <EOL> job = Job . create ( func = some_calculation , <EOL> args = ( <NUM_LIT:3> , <NUM_LIT:4> ) , kwargs = dict ( z = <NUM_LIT:2> ) , <EOL> scheduled_for = dt , <EOL> repeat = - <NUM_LIT:1> , <EOL> interval = rd , <EOL> between = '<STR_LIT>' <EOL> ) <EOL> job . save ( ) <EOL> job = Job . objects . get ( pk = job . id ) <EOL> self . assertEqual ( rd , job . interval ) <EOL> self . assertEqual ( dt , job . scheduled_for ) <EOL> class TestJobSave ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . job = Job . create ( func = some_calculation , args = ( <NUM_LIT:3> , <NUM_LIT:4> ) , kwargs = dict ( z = <NUM_LIT:2> ) ) <EOL> def test_job_save ( self ) : <EOL> """<STR_LIT>""" <EOL> self . job . save ( ) <EOL> self . assertIsNotNone ( self . job . id ) <EOL> class Test_get_job_or_promise ( TransactionTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . q = Queue ( scheduled = True ) <EOL> self . timeout = <NUM_LIT> <EOL> future = now ( ) + timedelta ( seconds = self . timeout / <NUM_LIT:2> ) <EOL> self . job = self . q . schedule_call ( future , do_nothing ) <EOL> def test_get_job_or_promise ( self ) : <EOL> """<STR_LIT>""" <EOL> job , promise , timeout = Job . _get_job_or_promise ( <EOL> self . q . connection , self . q , self . timeout ) <EOL> self . assertLessEqual ( timeout , self . timeout ) <EOL> self . assertIsNone ( job ) <EOL> self . assertEqual ( promise , self . q . name ) <EOL> def test_get_no_job_no_promise ( self ) : <EOL> """<STR_LIT>""" <EOL> job , promise , timeout = Job . _get_job_or_promise ( <EOL> self . q . connection , self . q , <NUM_LIT:1> ) <EOL> self . assertEqual ( timeout , <NUM_LIT:1> ) <EOL> self . assertIsNone ( job ) <EOL> self . assertIsNone ( promise ) <EOL> def test_get_earlier_job_no_promise ( self ) : <EOL> """<STR_LIT>""" <EOL> now_job = self . q . enqueue ( do_nothing ) <EOL> job , promise , timeout = Job . _get_job_or_promise ( <EOL> self . q . connection , self . q , <NUM_LIT> ) <EOL> self . assertEqual ( timeout , <NUM_LIT> ) <EOL> self . assertEqual ( now_job . id , job . id ) <EOL> self . assertIsNone ( promise ) <EOL> class Test_get_job_no_promise ( TransactionTestCase ) : <EOL> def setUp ( self ) : <EOL> self . q = Queue ( scheduled = True ) <EOL> self . timeout = <NUM_LIT> <EOL> future = now ( ) + timedelta ( seconds = <NUM_LIT:1> ) <EOL> self . job = self . q . schedule_call ( future , do_nothing ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> def test_get_job_no_promise ( self ) : <EOL> """<STR_LIT>""" <EOL> job , promise , timeout = Job . _get_job_or_promise ( <EOL> self . q . connection , self . q , self . timeout ) <EOL> self . assertEqual ( timeout , self . timeout ) <EOL> self . assertEquals ( job . id , self . job . id ) <EOL> self . assertIsNone ( promise ) <EOL> class TestJobSchedule ( TestCase ) : <EOL> def test_job_get_schedule_options ( self ) : <EOL> """<STR_LIT>""" <EOL> j = Job . create ( <EOL> do_nothing , <EOL> interval = <NUM_LIT> , <EOL> between = '<STR_LIT>' , <EOL> repeat = <NUM_LIT:10> , <EOL> weekdays = ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> ) <EOL> self . assertIsNotNone ( j . get_schedule_options ( ) ) </s>
