<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import unittest <EOL> from pants . util . timeout import Timeout , TimeoutReached <EOL> class FakeTimer ( object ) : <EOL> def __init__ ( self , seconds , handler ) : <EOL> self . _seconds = seconds <EOL> self . _handler = handler <EOL> self . _clock = <NUM_LIT:0> <EOL> self . _started = False <EOL> def cancel ( self ) : <EOL> pass <EOL> def start ( self ) : <EOL> self . _started = True <EOL> def move_clock_forward ( self , seconds ) : <EOL> if not self . _started : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> self . _clock += seconds <EOL> if self . _clock > self . _seconds : <EOL> self . _handler ( ) <EOL> class TestTimeout ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestTimeout , self ) . setUp ( ) <EOL> self . _fake_timer = None <EOL> self . _aborted = False <EOL> def _abort_handler ( self ) : <EOL> self . _aborted = True <EOL> def _make_fake_timer ( self , seconds , handler ) : <EOL> self . _fake_timer = FakeTimer ( seconds , handler ) <EOL> return self . _fake_timer <EOL> def _move_clock_forward ( self , seconds ) : <EOL> if self . _fake_timer is not None : <EOL> self . _fake_timer . move_clock_forward ( seconds ) <EOL> def test_timeout_success ( self ) : <EOL> with Timeout ( <NUM_LIT:2> , threading_timer = self . _make_fake_timer , abort_handler = self . _abort_handler ) : <EOL> self . _move_clock_forward ( <NUM_LIT:1> ) <EOL> self . assertFalse ( self . _aborted ) <EOL> def test_timeout_failure ( self ) : <EOL> with self . assertRaises ( TimeoutReached ) : <EOL> with Timeout ( <NUM_LIT:2> , threading_timer = self . _make_fake_timer , abort_handler = self . _abort_handler ) : <EOL> self . _move_clock_forward ( <NUM_LIT:3> ) <EOL> self . assertTrue ( self . _aborted ) <EOL> def test_timeout_none ( self ) : <EOL> with Timeout ( None , threading_timer = self . _make_fake_timer , abort_handler = self . _abort_handler ) : <EOL> self . _move_clock_forward ( <NUM_LIT:3> ) <EOL> self . assertFalse ( self . _aborted ) <EOL> def test_timeout_zero ( self ) : <EOL> with Timeout ( <NUM_LIT:0> , threading_timer = self . _make_fake_timer , abort_handler = self . _abort_handler ) : <EOL> self . _move_clock_forward ( <NUM_LIT:3> ) <EOL> self . assertFalse ( self . _aborted ) </s>
