<s> import os <EOL> import shutil <EOL> import stat <EOL> from threading import Lock , Thread <EOL> import time <EOL> import unittest <EOL> from pdagent . constants import ConsumeEvent , EnqueueWarnings <EOL> from pdagent . pdqueue import PDQEnqueuer , PDQueue , EmptyQueueError <EOL> from pdagent import pdqueue <EOL> _TEST_DIR = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> if _TEST_DIR . startswith ( "<STR_LIT>" ) : <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" , _TEST_DIR <EOL> _TEST_DIR = "<STR_LIT>" <EOL> print "<STR_LIT>" , _TEST_DIR <EOL> TEST_QUEUE_DIR = os . path . join ( _TEST_DIR , "<STR_LIT>" ) <EOL> TEST_DB_DIR = os . path . join ( _TEST_DIR , "<STR_LIT>" ) <EOL> BACKOFF_INTERVAL = <NUM_LIT:5> <EOL> ERROR_RETRY_LIMIT = <NUM_LIT:3> <EOL> class NoOpLock : <EOL> def __init__ ( self , lockfile ) : <EOL> pass <EOL> def acquire ( self ) : <EOL> pass <EOL> def release ( self ) : <EOL> pass <EOL> class MockDB : <EOL> def __init__ ( self ) : <EOL> self . _data = None <EOL> def get ( self ) : <EOL> return self . _data <EOL> def set ( self , json_data ) : <EOL> self . _data = json_data <EOL> class MockTime : <EOL> def __init__ ( self , time_sec = time . time ( ) ) : <EOL> self . _time_sec = time_sec <EOL> def time ( self ) : <EOL> return self . _time_sec <EOL> def sleep ( self , duration_sec ) : <EOL> self . _time_sec += duration_sec <EOL> def gmtime ( self ) : <EOL> return "<STR_LIT>" <EOL> def strftime ( self , fmt , t ) : <EOL> if fmt == "<STR_LIT>" : <EOL> return t <EOL> else : <EOL> raise Exception ( "<STR_LIT>" + fmt ) <EOL> class PDQueueTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> if os . path . exists ( TEST_QUEUE_DIR ) : <EOL> shutil . rmtree ( TEST_QUEUE_DIR ) <EOL> for t in pdqueue . QUEUE_SUBDIRS : <EOL> os . makedirs ( os . path . join ( TEST_QUEUE_DIR , t ) ) <EOL> if os . path . exists ( TEST_DB_DIR ) : <EOL> shutil . rmtree ( TEST_DB_DIR ) <EOL> os . makedirs ( TEST_DB_DIR ) <EOL> def new_queue ( self ) : <EOL> mock_time = MockTime ( ) <EOL> eq = PDQEnqueuer ( <EOL> queue_dir = TEST_QUEUE_DIR , <EOL> lock_class = NoOpLock , <EOL> time_calc = mock_time , <EOL> enqueue_file_mode = <NUM_LIT:0> <NUM_LIT> , <EOL> default_umask = <NUM_LIT:0> <NUM_LIT> <EOL> ) <EOL> q = PDQueue ( <EOL> queue_dir = TEST_QUEUE_DIR , <EOL> lock_class = NoOpLock , <EOL> time_calc = mock_time , <EOL> event_size_max_bytes = <NUM_LIT:10> , <EOL> backoff_interval = BACKOFF_INTERVAL , <EOL> retry_limit_for_possible_errors = ERROR_RETRY_LIMIT , <EOL> backoff_db = MockDB ( ) , <EOL> counter_db = MockDB ( ) <EOL> ) <EOL> return eq , q <EOL> def test__open_creat_excl ( self ) : <EOL> from pdagent . pdqueue import _open_creat_excl <EOL> eq , _ = self . new_queue ( ) <EOL> fname_abs = eq . _abspath ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> fd1 = _open_creat_excl ( fname_abs , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . assertNotEquals ( fd1 , None ) <EOL> fd2 = None <EOL> try : <EOL> fd2 = _open_creat_excl ( fname_abs , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . assertEquals ( fd2 , None ) <EOL> finally : <EOL> os . close ( fd1 ) <EOL> if fd2 : <EOL> os . close ( fd2 ) <EOL> def test__link ( self ) : <EOL> from pdagent . pdqueue import _link <EOL> eq , _ = self . new_queue ( ) <EOL> f1 = eq . _abspath ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> f2 = eq . _abspath ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertFalse ( os . path . exists ( f1 ) ) <EOL> self . assertFalse ( os . path . exists ( f2 ) ) <EOL> open ( f1 , "<STR_LIT:w>" ) . write ( "<STR_LIT:foo>" ) <EOL> self . assertEquals ( open ( f1 ) . read ( ) , "<STR_LIT:foo>" ) <EOL> self . assertTrue ( _link ( f1 , f2 ) ) <EOL> self . assertEquals ( open ( f1 ) . read ( ) , "<STR_LIT:foo>" ) <EOL> self . assertEquals ( open ( f2 ) . read ( ) , "<STR_LIT:foo>" ) <EOL> self . assertFalse ( _link ( f1 , f2 ) ) <EOL> open ( f1 , "<STR_LIT:w>" ) . write ( "<STR_LIT:bar>" ) <EOL> self . assertFalse ( _link ( f1 , f2 ) ) <EOL> self . assertEquals ( open ( f1 ) . read ( ) , "<STR_LIT:bar>" ) <EOL> self . assertEquals ( open ( f2 ) . read ( ) , "<STR_LIT:bar>" ) <EOL> os . unlink ( f1 ) <EOL> self . assertFalse ( os . path . exists ( f1 ) ) <EOL> self . assertEquals ( open ( f2 ) . read ( ) , "<STR_LIT:bar>" ) <EOL> os . unlink ( f2 ) <EOL> self . assertFalse ( os . path . exists ( f1 ) ) <EOL> self . assertFalse ( os . path . exists ( f2 ) ) <EOL> def test_enqueue_and_dequeue ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ ] ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo ] ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_foo ) ) . read ( ) , "<STR_LIT:foo>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> f_bar , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo , f_bar ] ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_foo ) ) . read ( ) , "<STR_LIT:foo>" ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_bar ) ) . read ( ) , "<STR_LIT:bar>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> f_baz , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo , f_bar , f_baz ] ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_foo ) ) . read ( ) , "<STR_LIT:foo>" ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_bar ) ) . read ( ) , "<STR_LIT:bar>" ) <EOL> self . assertEquals ( open ( q . _abspath ( "<STR_LIT>" , f_baz ) ) . read ( ) , "<STR_LIT>" ) <EOL> def verify_and_consume ( event , event_id ) : <EOL> def consume ( s , i ) : <EOL> self . assertEquals ( event , s ) <EOL> self . assertEquals ( event_id , i ) <EOL> return ConsumeEvent . CONSUMED <EOL> return consume <EOL> q . dequeue ( verify_and_consume ( "<STR_LIT:foo>" , f_foo ) ) <EOL> q . dequeue ( verify_and_consume ( "<STR_LIT:bar>" , f_bar ) ) <EOL> q . dequeue ( verify_and_consume ( "<STR_LIT>" , f_baz ) ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ ] ) <EOL> self . assertRaises ( <EOL> EmptyQueueError , q . dequeue , lambda s , i : ConsumeEvent . CONSUMED <EOL> ) <EOL> success_contents = [ ] <EOL> for f in q . _queued_files ( "<STR_LIT>" ) : <EOL> fd = open ( q . _abspath ( "<STR_LIT>" , f ) ) <EOL> success_contents . append ( fd . read ( ) ) <EOL> fd . close ( ) <EOL> self . assertEquals ( success_contents , [ "<STR_LIT:foo>" , "<STR_LIT:bar>" , "<STR_LIT>" ] ) <EOL> def test_bad_fnames ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> def make_bad_entry ( n ) : <EOL> badf = q . _abspath ( "<STR_LIT>" , n ) <EOL> open ( badf , "<STR_LIT:w>" ) . close ( ) <EOL> make_bad_entry ( "<STR_LIT>" ) <EOL> make_bad_entry ( "<STR_LIT>" ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> make_bad_entry ( "<STR_LIT>" ) <EOL> make_bad_entry ( "<STR_LIT>" ) <EOL> q . flush ( <EOL> lambda s , i : ConsumeEvent . CONSUMED , <EOL> lambda : False <EOL> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( <EOL> q . _queued_files ( "<STR_LIT>" ) , <EOL> [ <EOL> "<STR_LIT>" , <EOL> f_foo , <EOL> ] <EOL> ) <EOL> self . assertEquals ( <EOL> q . _queued_files ( "<STR_LIT>" ) , <EOL> [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> ) <EOL> def test_consume_error ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> def erroneous_consume_foo ( s , i ) : <EOL> self . assertEquals ( "<STR_LIT:foo>" , s ) <EOL> self . assertEquals ( f_foo , i ) <EOL> return ConsumeEvent . BAD_ENTRY <EOL> q . dequeue ( erroneous_consume_foo ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:1> ) <EOL> self . assertRaises ( <EOL> EmptyQueueError , q . dequeue , lambda s : ConsumeEvent . CONSUMED <EOL> ) <EOL> def test_dequeue_no_lock_on_no_work ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> trace = [ ] <EOL> f_delete_me = [ ] <EOL> class FileDeletingLockClass : <EOL> def __init__ ( self , lockfile ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> def acquire ( self ) : <EOL> if f_delete_me : <EOL> f = f_delete_me . pop ( ) <EOL> trace . append ( "<STR_LIT:D>" ) <EOL> trace . append ( f ) <EOL> os . remove ( f ) <EOL> trace . append ( "<STR_LIT>" ) <EOL> def release ( self ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> q . lock_class = FileDeletingLockClass <EOL> self . assertRaises ( EmptyQueueError , q . dequeue , None ) <EOL> self . assertEquals ( trace , [ ] ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> f_foo_abs = q . _abspath ( "<STR_LIT>" , f_foo ) <EOL> f_delete_me . append ( f_foo_abs ) <EOL> self . assertRaises ( EmptyQueueError , q . dequeue , None ) <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT:D>" , f_foo_abs , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_huge_event_not_processed ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> f , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f ] ) <EOL> def unexpected_consume ( s , i ) : <EOL> self . fail ( "<STR_LIT>" % s ) <EOL> q . dequeue ( unexpected_consume ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:1> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> def test_backoff_bad_event ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> e1_1 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> e1_2 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> e2_1 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> events_processed = [ ] <EOL> count = <NUM_LIT:0> <EOL> max_total_attempts = ERROR_RETRY_LIMIT + <NUM_LIT:1> <EOL> def consume_with_backoff ( s , i ) : <EOL> events_processed . append ( s ) <EOL> if count == <NUM_LIT:1> and s == "<STR_LIT>" and i == e2_1 : <EOL> return ConsumeEvent . CONSUMED <EOL> elif count <= max_total_attempts and s == "<STR_LIT:foo>" and i == e1_1 : <EOL> return ConsumeEvent . BACKOFF_SVCKEY_BAD_ENTRY <EOL> elif count == max_total_attempts and s == "<STR_LIT:bar>" and i == e1_2 : <EOL> return ConsumeEvent . CONSUMED <EOL> else : <EOL> self . fail ( <EOL> "<STR_LIT>" % ( s , count ) ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 , e2_1 ] ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:0> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( len ( events_processed ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:0> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> for i in range ( <NUM_LIT:2> , max_total_attempts ) : <EOL> q . time . sleep ( BACKOFF_INTERVAL ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , i , i - <NUM_LIT:1> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> q . time . sleep ( BACKOFF_INTERVAL ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( <EOL> q . _queued_files ( "<STR_LIT>" ) , <EOL> [ e1_1 ] <EOL> ) <EOL> self . _assertBackoffData ( q , None ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:2> , <NUM_LIT:1> ) ) <EOL> self . assertRaises ( <EOL> EmptyQueueError , q . dequeue , lambda s , i : ConsumeEvent . CONSUMED <EOL> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:2> , <NUM_LIT:1> ) ) <EOL> def test_backoff_not_consumed ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> e1_1 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> e1_2 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) <EOL> q . time . sleep ( <NUM_LIT> ) <EOL> e2_1 , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> events_processed = [ ] <EOL> count = <NUM_LIT:0> <EOL> max_total_attempts = ERROR_RETRY_LIMIT + <NUM_LIT:1> <EOL> def consume_with_backoff ( s , i ) : <EOL> events_processed . append ( s ) <EOL> if count == <NUM_LIT:1> and s == "<STR_LIT>" : <EOL> return ConsumeEvent . CONSUMED <EOL> elif count <= max_total_attempts + <NUM_LIT:1> and s == "<STR_LIT:foo>" and i == e1_1 : <EOL> return ConsumeEvent . BACKOFF_SVCKEY_NOT_CONSUMED <EOL> elif count == max_total_attempts + <NUM_LIT:2> and ( ( s == "<STR_LIT:foo>" and i == e1_1 ) or s == "<STR_LIT:bar>" and i == e1_2 ) : <EOL> return ConsumeEvent . CONSUMED <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( s , count ) ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 , e2_1 ] ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:0> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( len ( events_processed ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT:0> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> for i in range ( <NUM_LIT:2> , max_total_attempts ) : <EOL> q . time . sleep ( BACKOFF_INTERVAL ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , [ ( "<STR_LIT>" , i , i - <NUM_LIT:1> ) ] ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> for i in [ <NUM_LIT:0> , <NUM_LIT:1> ] : <EOL> q . time . sleep ( BACKOFF_INTERVAL ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ e1_1 , e1_2 ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( <EOL> q , <EOL> [ ( "<STR_LIT>" , max_total_attempts + i , - <NUM_LIT:1> ) ] <EOL> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> q . time . sleep ( BACKOFF_INTERVAL ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_backoff , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertBackoffData ( q , None ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:3> , <NUM_LIT:0> ) ) <EOL> self . assertRaises ( <EOL> EmptyQueueError , q . dequeue , lambda s : ConsumeEvent . CONSUMED <EOL> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:3> , <NUM_LIT:0> ) ) <EOL> def test_stop_processing ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> q . time . sleep ( <NUM_LIT:1> ) <EOL> eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) <EOL> q . time . sleep ( <NUM_LIT:1> ) <EOL> eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> events_processed = [ ] <EOL> count = <NUM_LIT:0> <EOL> def consume_with_stopall ( s , i ) : <EOL> events_processed . append ( s ) <EOL> if count == <NUM_LIT:1> and s == "<STR_LIT:foo>" and i == f_foo : <EOL> return ConsumeEvent . STOP_ALL <EOL> elif count == <NUM_LIT:2> : <EOL> return ConsumeEvent . CONSUMED <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( s , count ) ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:3> ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_stopall , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:3> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> count += <NUM_LIT:1> <EOL> events_processed = [ ] <EOL> q . flush ( consume_with_stopall , lambda : False ) <EOL> self . assertEquals ( events_processed , [ "<STR_LIT:foo>" , "<STR_LIT:bar>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertCounterData ( q , ( <NUM_LIT:3> , <NUM_LIT:0> ) ) <EOL> self . assertRaises ( <EOL> EmptyQueueError , q . dequeue , lambda s : ConsumeEvent . CONSUMED <EOL> ) <EOL> def test_enqueue_never_blocks ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> f_foo , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> trace = [ ] <EOL> class LockClass : <EOL> def __init__ ( self , lockfile ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> def acquire ( self ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> def release ( self ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> q . lock_class = LockClass <EOL> def dequeue ( ) : <EOL> try : <EOL> def consume ( * args ) : <EOL> trace . append ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> trace . append ( "<STR_LIT>" ) <EOL> return ConsumeEvent . CONSUMED <EOL> q . dequeue ( consume ) <EOL> except EmptyQueueError : <EOL> trace . append ( "<STR_LIT>" ) <EOL> thread_dequeue = Thread ( target = dequeue ) <EOL> thread_dequeue . start ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo ] ) <EOL> f_bar , _ = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo , f_bar ] ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_bar ] ) <EOL> def test_parallel_dequeue ( self ) : <EOL> eq1 , q1 = self . new_queue ( ) <EOL> _ , q2 = self . new_queue ( ) <EOL> eq1 . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> dequeue_lockfile = q1 . _dequeue_lockfile <EOL> trace = [ ] <EOL> lock = Lock ( ) <EOL> def make_lock_class ( name ) : <EOL> outer_self = self <EOL> class LockClass : <EOL> def __init__ ( self , lockfile ) : <EOL> outer_self . assertEquals ( dequeue_lockfile , lockfile ) <EOL> def acquire ( self ) : <EOL> trace . append ( name + "<STR_LIT>" ) <EOL> lock . acquire ( ) <EOL> trace . append ( name + "<STR_LIT>" ) <EOL> def release ( self ) : <EOL> trace . append ( name + "<STR_LIT>" ) <EOL> lock . release ( ) <EOL> return LockClass <EOL> q1 . lock_class = make_lock_class ( "<STR_LIT>" ) <EOL> q2 . lock_class = make_lock_class ( "<STR_LIT>" ) <EOL> def dequeue2 ( ) : <EOL> try : <EOL> def consume2 ( * args ) : <EOL> self . fail ( ) <EOL> q2 . dequeue ( consume2 ) <EOL> except EmptyQueueError : <EOL> trace . append ( "<STR_LIT>" ) <EOL> thread_dequeue2 = Thread ( target = dequeue2 ) <EOL> def consume1 ( s , i ) : <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> thread_dequeue2 . start ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEquals ( trace , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> trace . append ( "<STR_LIT>" + s ) <EOL> return ConsumeEvent . CONSUMED <EOL> q1 . dequeue ( consume1 ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEquals ( trace , [ <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] ) <EOL> def test_restrictive_umask ( self ) : <EOL> orig_umask = os . umask ( <NUM_LIT:0> <NUM_LIT> ) <EOL> try : <EOL> eq , q = self . new_queue ( ) <EOL> f_foo , problems = eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) <EOL> self . assertEquals ( problems , [ EnqueueWarnings . UMASK_TOO_RESTRICTIVE ] ) <EOL> self . assertEquals ( q . _queued_files ( ) , [ f_foo ] ) <EOL> f_stmode = os . stat ( q . _abspath ( "<STR_LIT>" , f_foo ) ) . st_mode <EOL> fmode = f_stmode & ( stat . S_IRWXU | stat . S_IRWXG | stat . S_IRWXO ) <EOL> self . assertEquals ( fmode , eq . enqueue_file_mode ) <EOL> finally : <EOL> os . umask ( orig_umask ) <EOL> def test_resurrect ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> fnames = [ ] <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" , "<STR_LIT:foo>" ) [ <NUM_LIT:0> ] ) <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" , "<STR_LIT:bar>" ) [ <NUM_LIT:0> ] ) <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) [ <NUM_LIT:0> ] ) <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) [ <NUM_LIT:0> ] ) <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" , "<STR_LIT>" ) [ <NUM_LIT:0> ] ) <EOL> for i in [ <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:4> ] : <EOL> q . _unsafe_change_event_type ( fnames [ i ] , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:3> ) <EOL> q . resurrect ( "<STR_LIT>" ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:3> ) <EOL> errfiles = q . _queued_files ( "<STR_LIT>" ) <EOL> self . assertEquals ( len ( errfiles ) , <NUM_LIT:2> ) <EOL> for errname in errfiles : <EOL> self . assertEquals ( errname . find ( "<STR_LIT>" ) , - <NUM_LIT:1> ) <EOL> self . assertTrue ( <EOL> errname . find ( "<STR_LIT>" ) != - <NUM_LIT:1> or <EOL> errname . find ( "<STR_LIT>" ) != - <NUM_LIT:1> <EOL> ) <EOL> q . resurrect ( "<STR_LIT>" ) <EOL> q . resurrect ( ) <EOL> self . assertEquals ( len ( q . _queued_files ( ) ) , <NUM_LIT:5> ) <EOL> self . assertEquals ( len ( q . _queued_files ( "<STR_LIT>" ) ) , <NUM_LIT:0> ) <EOL> self . _assertCounterData ( q , None ) <EOL> def test_stats ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> events = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> fnames = [ ] <EOL> for e in events : <EOL> k = e [ <NUM_LIT:1> ] <EOL> fnames . append ( eq . enqueue ( "<STR_LIT>" % k , e ) [ <NUM_LIT:0> ] ) <EOL> eq . time . sleep ( <NUM_LIT:5> ) <EOL> for i in [ <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:4> ] : <EOL> q . _unsafe_change_event_type ( fnames [ i ] , "<STR_LIT>" , "<STR_LIT>" ) <EOL> for i in [ <NUM_LIT:2> , <NUM_LIT:7> , <NUM_LIT:8> ] : <EOL> q . _unsafe_change_event_type ( fnames [ i ] , "<STR_LIT>" , "<STR_LIT>" ) <EOL> q . backoff_info . increment ( "<STR_LIT>" ) <EOL> for _ in range ( <NUM_LIT:20> ) : <EOL> q . counter_info . increment_success ( ) <EOL> for _ in range ( <NUM_LIT:2> ) : <EOL> q . counter_info . increment_failure ( ) <EOL> def stats_by_state ( data ) : <EOL> stats = dict ( ) <EOL> for d in data : <EOL> ( service_type , count , oldest_age , newest_age , <EOL> service_key_count ) = d <EOL> if count : <EOL> stats [ service_type + "<STR_LIT>" ] = { <EOL> "<STR_LIT:count>" : count , <EOL> "<STR_LIT>" : oldest_age , <EOL> "<STR_LIT>" : newest_age , <EOL> "<STR_LIT>" : service_key_count <EOL> } <EOL> return stats <EOL> snapshot_stats = stats_by_state ( [ <EOL> ( "<STR_LIT>" , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT:2> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT:5> , <NUM_LIT:2> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:3> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> ] ) <EOL> snapshot_stats [ "<STR_LIT>" ] = <NUM_LIT:1> <EOL> expected_stats = { <EOL> "<STR_LIT>" : snapshot_stats , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:20> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> self . assertEqual ( q . get_stats ( detailed_snapshot = True ) , expected_stats ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> self . assertEqual ( q . get_stats ( ) , expected_stats ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> snapshot_stats [ "<STR_LIT>" ] = stats_by_state ( [ <EOL> ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) , <EOL> ] ) <EOL> snapshot_stats [ "<STR_LIT>" ] = stats_by_state ( [ <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:2> , <NUM_LIT:30> , <NUM_LIT> , <NUM_LIT:1> ) , <EOL> ] ) <EOL> snapshot_stats [ "<STR_LIT>" ] = stats_by_state ( [ <EOL> ( "<STR_LIT>" , <NUM_LIT:2> , <NUM_LIT:20> , <NUM_LIT:15> , <NUM_LIT:1> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ] ) <EOL> snapshot_stats [ "<STR_LIT>" ] = stats_by_state ( [ <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:2> , <NUM_LIT:10> , <NUM_LIT:5> , <NUM_LIT:1> ) , <EOL> ( "<STR_LIT>" , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> ] ) <EOL> self . assertEqual ( <EOL> q . get_stats ( detailed_snapshot = True , per_service_key_snapshot = True ) , <EOL> expected_stats <EOL> ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> snapshot_stats . pop ( "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> q . get_stats ( <EOL> detailed_snapshot = True , <EOL> per_service_key_snapshot = True , <EOL> service_key = "<STR_LIT>" <EOL> ) , <EOL> expected_stats <EOL> ) <EOL> def test_cleanup ( self ) : <EOL> eq , q = self . new_queue ( ) <EOL> def enqueue_before ( sec , prefix = "<STR_LIT>" ) : <EOL> enqueue_time_us = ( int ( time . time ( ) ) - sec ) * ( <NUM_LIT:1000> * <NUM_LIT:1000> ) <EOL> fname = "<STR_LIT>" % ( <EOL> enqueue_time_us , <EOL> "<STR_LIT>" % ( enqueue_time_us % <NUM_LIT:10> ) <EOL> ) <EOL> fpath = os . path . join ( q . queue_dir , prefix , fname ) <EOL> os . close ( os . open ( fpath , os . O_CREAT ) ) <EOL> return fname <EOL> q1 = enqueue_before ( <NUM_LIT> ) <EOL> t1 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> e1 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> s1_1 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> s1_2 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> q2 = enqueue_before ( <NUM_LIT:1000> ) <EOL> t2 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> s2 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> e2_1 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> e2_2 = enqueue_before ( <NUM_LIT> , prefix = "<STR_LIT>" ) <EOL> q . cleanup ( <NUM_LIT> ) <EOL> expected_unremoved = [ q1 , q2 , t2 , e2_2 , e2_1 , s2 ] <EOL> actual_unremoved = q . _queued_files ( ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> self . assertEquals ( expected_unremoved , actual_unremoved ) <EOL> invalid = "<STR_LIT>" <EOL> os . close ( os . open ( os . path . join ( q . queue_dir , "<STR_LIT>" , invalid ) , os . O_CREAT ) ) <EOL> q . cleanup ( <NUM_LIT:100> ) <EOL> expected_unremoved = [ q1 , q2 , invalid ] <EOL> actual_unremoved = q . _queued_files ( ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> actual_unremoved . extend ( q . _queued_files ( "<STR_LIT>" ) ) <EOL> self . assertEquals ( expected_unremoved , actual_unremoved ) <EOL> self . _assertCounterData ( q , None ) <EOL> def _assertBackoffData ( self , q , data ) : <EOL> backup_data = q . backoff_info . _db . get ( ) <EOL> attempts = { } <EOL> retries = { } <EOL> if data : <EOL> for ( svc_key , count , backoff_index ) in data : <EOL> attempts [ svc_key ] = count <EOL> retries [ svc_key ] = int ( q . time . time ( ) + BACKOFF_INTERVAL ) <EOL> self . assertEqual ( backup_data , { <EOL> "<STR_LIT>" : attempts , <EOL> "<STR_LIT>" : retries <EOL> } ) <EOL> def _assertCounterData ( self , q , data ) : <EOL> counter_db = q . counter_info . _db <EOL> counter_data = counter_db . get ( ) <EOL> try : <EOL> counter_db . cached_started_on <EOL> except AttributeError : <EOL> self . assertNotEquals ( counter_data [ "<STR_LIT>" ] , None ) <EOL> counter_db . cached_started_on = counter_data [ "<STR_LIT>" ] <EOL> expected = { <EOL> "<STR_LIT>" : counter_db . cached_started_on <EOL> } <EOL> if data : <EOL> success , failure = data <EOL> if success != <NUM_LIT:0> : <EOL> expected [ "<STR_LIT>" ] = success <EOL> if failure != <NUM_LIT:0> : <EOL> expected [ "<STR_LIT>" ] = failure <EOL> self . assertEqual ( counter_data , expected ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
