<s> from __future__ import absolute_import <EOL> from datetime import datetime , timedelta <EOL> from itertools import count <EOL> from celery . five import monotonic <EOL> from celery . schedules import schedule , crontab <EOL> from djcelery import schedulers <EOL> from djcelery import celery <EOL> from djcelery . app import app <EOL> from djcelery . models import PeriodicTask , IntervalSchedule , CrontabSchedule <EOL> from djcelery . models import PeriodicTasks <EOL> from djcelery . tests . utils import unittest <EOL> from djcelery . compat import unicode <EOL> def create_model_interval ( schedule , ** kwargs ) : <EOL> return create_model ( interval = IntervalSchedule . from_schedule ( schedule ) , <EOL> ** kwargs ) <EOL> def create_model_crontab ( schedule , ** kwargs ) : <EOL> return create_model ( crontab = CrontabSchedule . from_schedule ( schedule ) , <EOL> ** kwargs ) <EOL> _next_id_get = count ( <NUM_LIT:0> ) <EOL> _next_id = lambda : next ( _next_id_get ) <EOL> def create_model ( Model = PeriodicTask , ** kwargs ) : <EOL> entry = dict ( name = '<STR_LIT>' . format ( _next_id ( ) ) , <EOL> task = '<STR_LIT>' . format ( _next_id ( ) ) , <EOL> args = '<STR_LIT>' , <EOL> kwargs = '<STR_LIT>' , <EOL> queue = '<STR_LIT>' , <EOL> routing_key = '<STR_LIT>' , <EOL> exchange = '<STR_LIT:foo>' ) <EOL> return Model ( ** dict ( entry , ** kwargs ) ) <EOL> class EntryTrackSave ( schedulers . ModelEntry ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . saved = <NUM_LIT:0> <EOL> super ( EntryTrackSave , self ) . __init__ ( * args , ** kwargs ) <EOL> def save ( self ) : <EOL> self . saved += <NUM_LIT:1> <EOL> super ( EntryTrackSave , self ) . save ( ) <EOL> class EntrySaveRaises ( schedulers . ModelEntry ) : <EOL> def save ( self ) : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> class TrackingScheduler ( schedulers . DatabaseScheduler ) : <EOL> Entry = EntryTrackSave <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . flushed = <NUM_LIT:0> <EOL> schedulers . DatabaseScheduler . __init__ ( self , * args , ** kwargs ) <EOL> def sync ( self ) : <EOL> self . flushed += <NUM_LIT:1> <EOL> schedulers . DatabaseScheduler . sync ( self ) <EOL> class test_ModelEntry ( unittest . TestCase ) : <EOL> Entry = EntryTrackSave <EOL> def tearDown ( self ) : <EOL> PeriodicTask . objects . all ( ) . delete ( ) <EOL> def test_entry ( self ) : <EOL> m = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) ) <EOL> e = self . Entry ( m ) <EOL> self . assertListEqual ( e . args , [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . assertDictEqual ( e . kwargs , { '<STR_LIT>' : '<STR_LIT:foo>' } ) <EOL> self . assertTrue ( e . schedule ) <EOL> self . assertEqual ( e . total_run_count , <NUM_LIT:0> ) <EOL> self . assertIsInstance ( e . last_run_at , datetime ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , e . options ) <EOL> right_now = celery . now ( ) <EOL> m2 = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) , <EOL> last_run_at = right_now ) <EOL> self . assertTrue ( m2 . last_run_at ) <EOL> e2 = self . Entry ( m2 ) <EOL> self . assertIs ( e2 . last_run_at , right_now ) <EOL> e3 = e2 . next ( ) <EOL> self . assertGreater ( e3 . last_run_at , e2 . last_run_at ) <EOL> self . assertEqual ( e3 . total_run_count , <NUM_LIT:1> ) <EOL> class test_DatabaseScheduler ( unittest . TestCase ) : <EOL> Scheduler = TrackingScheduler <EOL> def setUp ( self ) : <EOL> PeriodicTask . objects . all ( ) . delete ( ) <EOL> self . prev_schedule = celery . conf . CELERYBEAT_SCHEDULE <EOL> celery . conf . CELERYBEAT_SCHEDULE = { } <EOL> m1 = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) ) <EOL> m2 = create_model_interval ( schedule ( timedelta ( minutes = <NUM_LIT:20> ) ) ) <EOL> m3 = create_model_crontab ( crontab ( minute = '<STR_LIT>' ) ) <EOL> for obj in m1 , m2 , m3 : <EOL> obj . save ( ) <EOL> self . s = self . Scheduler ( app = app ) <EOL> self . m1 = PeriodicTask . objects . get ( name = m1 . name ) <EOL> self . m2 = PeriodicTask . objects . get ( name = m2 . name ) <EOL> self . m3 = PeriodicTask . objects . get ( name = m3 . name ) <EOL> def tearDown ( self ) : <EOL> celery . conf . CELERYBEAT_SCHEDULE = self . prev_schedule <EOL> PeriodicTask . objects . all ( ) . delete ( ) <EOL> def test_constructor ( self ) : <EOL> self . assertIsInstance ( self . s . _dirty , set ) <EOL> self . assertIsNone ( self . s . _last_sync ) <EOL> self . assertTrue ( self . s . sync_every ) <EOL> def test_all_as_schedule ( self ) : <EOL> sched = self . s . schedule <EOL> self . assertTrue ( sched ) <EOL> self . assertEqual ( len ( sched ) , <NUM_LIT:4> ) <EOL> self . assertIn ( '<STR_LIT>' , sched ) <EOL> for n , e in sched . items ( ) : <EOL> self . assertIsInstance ( e , self . s . Entry ) <EOL> def test_schedule_changed ( self ) : <EOL> self . m2 . args = '<STR_LIT>' <EOL> self . m2 . save ( ) <EOL> e2 = self . s . schedule [ self . m2 . name ] <EOL> self . assertListEqual ( e2 . args , [ <NUM_LIT:16> , <NUM_LIT:16> ] ) <EOL> self . m1 . args = '<STR_LIT>' <EOL> self . m1 . save ( ) <EOL> e1 = self . s . schedule [ self . m1 . name ] <EOL> self . assertListEqual ( e1 . args , [ <NUM_LIT:32> , <NUM_LIT:32> ] ) <EOL> e1 = self . s . schedule [ self . m1 . name ] <EOL> self . assertListEqual ( e1 . args , [ <NUM_LIT:32> , <NUM_LIT:32> ] ) <EOL> self . m3 . delete ( ) <EOL> self . assertRaises ( KeyError , self . s . schedule . __getitem__ , self . m3 . name ) <EOL> def test_should_sync ( self ) : <EOL> self . assertTrue ( self . s . should_sync ( ) ) <EOL> self . s . _last_sync = monotonic ( ) <EOL> self . assertFalse ( self . s . should_sync ( ) ) <EOL> self . s . _last_sync -= self . s . sync_every <EOL> self . assertTrue ( self . s . should_sync ( ) ) <EOL> def test_reserve ( self ) : <EOL> e1 = self . s . schedule [ self . m1 . name ] <EOL> self . s . schedule [ self . m1 . name ] = self . s . reserve ( e1 ) <EOL> self . assertEqual ( self . s . flushed , <NUM_LIT:1> ) <EOL> e2 = self . s . schedule [ self . m2 . name ] <EOL> self . s . schedule [ self . m2 . name ] = self . s . reserve ( e2 ) <EOL> self . assertEqual ( self . s . flushed , <NUM_LIT:1> ) <EOL> self . assertIn ( self . m2 . name , self . s . _dirty ) <EOL> def test_sync_saves_last_run_at ( self ) : <EOL> e1 = self . s . schedule [ self . m2 . name ] <EOL> last_run = e1 . last_run_at <EOL> last_run2 = last_run - timedelta ( days = <NUM_LIT:1> ) <EOL> e1 . model . last_run_at = last_run2 <EOL> self . s . _dirty . add ( self . m2 . name ) <EOL> self . s . sync ( ) <EOL> e2 = self . s . schedule [ self . m2 . name ] <EOL> self . assertEqual ( e2 . last_run_at , last_run2 ) <EOL> def test_sync_syncs_before_save ( self ) : <EOL> e1 = self . s . schedule [ self . m2 . name ] <EOL> self . s . _last_sync = monotonic ( ) <EOL> e2 = self . s . schedule [ e1 . name ] = self . s . reserve ( e1 ) <EOL> self . assertEqual ( self . s . flushed , <NUM_LIT:1> ) <EOL> m2 = PeriodicTask . objects . get ( pk = self . m2 . pk ) <EOL> m2 . args = '<STR_LIT>' <EOL> m2 . save ( ) <EOL> e3 = self . s . schedule [ self . m2 . name ] <EOL> self . assertEqual ( self . s . flushed , <NUM_LIT:2> ) <EOL> self . assertEqual ( e3 . last_run_at , e2 . last_run_at ) <EOL> self . assertListEqual ( e3 . args , [ <NUM_LIT:16> , <NUM_LIT:16> ] ) <EOL> def test_sync_not_dirty ( self ) : <EOL> self . s . _dirty . clear ( ) <EOL> self . s . sync ( ) <EOL> def test_sync_object_gone ( self ) : <EOL> self . s . _dirty . add ( '<STR_LIT>' ) <EOL> self . s . sync ( ) <EOL> def test_sync_rollback_on_save_error ( self ) : <EOL> self . s . schedule [ self . m1 . name ] = EntrySaveRaises ( self . m1 ) <EOL> self . s . _dirty . add ( self . m1 . name ) <EOL> self . assertRaises ( RuntimeError , self . s . sync ) <EOL> class test_models ( unittest . TestCase ) : <EOL> def test_IntervalSchedule_unicode ( self ) : <EOL> self . assertEqual ( unicode ( IntervalSchedule ( every = <NUM_LIT:1> , period = '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( unicode ( IntervalSchedule ( every = <NUM_LIT:10> , period = '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' ) <EOL> def test_CrontabSchedule_unicode ( self ) : <EOL> self . assertEqual ( unicode ( CrontabSchedule ( minute = <NUM_LIT:3> , <EOL> hour = <NUM_LIT:3> , <EOL> day_of_week = None ) ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( unicode ( CrontabSchedule ( minute = <NUM_LIT:3> , <EOL> hour = <NUM_LIT:3> , <EOL> day_of_week = '<STR_LIT>' , <EOL> day_of_month = '<STR_LIT>' , <EOL> month_of_year = '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' ) <EOL> def test_PeriodicTask_unicode_interval ( self ) : <EOL> p = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) ) <EOL> self . assertEqual ( unicode ( p ) , <EOL> '<STR_LIT>' . format ( p . name ) ) <EOL> def test_PeriodicTask_unicode_crontab ( self ) : <EOL> p = create_model_crontab ( crontab ( hour = '<STR_LIT>' , day_of_week = '<STR_LIT>' ) ) <EOL> self . assertEqual ( unicode ( p ) , <EOL> '<STR_LIT>' . format ( p . name ) ) <EOL> def test_PeriodicTask_schedule_property ( self ) : <EOL> p1 = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) ) <EOL> s1 = p1 . schedule <EOL> self . assertEqual ( s1 . run_every . total_seconds ( ) , <NUM_LIT:10> ) <EOL> p2 = create_model_crontab ( crontab ( hour = '<STR_LIT>' , <EOL> minute = '<STR_LIT>' , <EOL> day_of_month = '<STR_LIT>' , <EOL> month_of_year = '<STR_LIT>' ) ) <EOL> s2 = p2 . schedule <EOL> self . assertSetEqual ( s2 . hour , set ( [ <NUM_LIT:4> , <NUM_LIT:5> ] ) ) <EOL> self . assertSetEqual ( s2 . minute , set ( [ <NUM_LIT:10> , <NUM_LIT:20> , <NUM_LIT:30> ] ) ) <EOL> self . assertSetEqual ( s2 . day_of_week , set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) ) <EOL> self . assertSetEqual ( s2 . day_of_month , set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ] ) ) <EOL> self . assertSetEqual ( s2 . month_of_year , set ( [ <NUM_LIT:1> , <NUM_LIT:4> , <NUM_LIT:7> , <NUM_LIT:10> ] ) ) <EOL> def test_PeriodicTask_unicode_no_schedule ( self ) : <EOL> p = create_model ( ) <EOL> self . assertEqual ( unicode ( p ) , '<STR_LIT>' . format ( p . name ) ) <EOL> def test_CrontabSchedule_schedule ( self ) : <EOL> s = CrontabSchedule ( minute = '<STR_LIT>' , hour = '<STR_LIT>' , day_of_week = '<STR_LIT:*>' , <EOL> day_of_month = '<STR_LIT>' , month_of_year = '<STR_LIT>' ) <EOL> self . assertEqual ( s . schedule . minute , set ( [ <NUM_LIT:3> , <NUM_LIT:7> ] ) ) <EOL> self . assertEqual ( s . schedule . hour , set ( [ <NUM_LIT:3> , <NUM_LIT:4> ] ) ) <EOL> self . assertEqual ( s . schedule . day_of_week , set ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] ) ) <EOL> self . assertEqual ( s . schedule . day_of_month , set ( [ <NUM_LIT:1> , <NUM_LIT:16> ] ) ) <EOL> self . assertEqual ( s . schedule . month_of_year , set ( [ <NUM_LIT:1> , <NUM_LIT:7> ] ) ) <EOL> class test_model_PeriodicTasks ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> PeriodicTasks . objects . all ( ) . delete ( ) <EOL> def test_track_changes ( self ) : <EOL> self . assertIsNone ( PeriodicTasks . last_change ( ) ) <EOL> m1 = create_model_interval ( schedule ( timedelta ( seconds = <NUM_LIT:10> ) ) ) <EOL> m1 . save ( ) <EOL> x = PeriodicTasks . last_change ( ) <EOL> self . assertTrue ( x ) <EOL> m1 . args = '<STR_LIT>' <EOL> m1 . save ( ) <EOL> y = PeriodicTasks . last_change ( ) <EOL> self . assertTrue ( y ) <EOL> self . assertGreater ( y , x ) </s>
