<s> import os <EOL> import tempfile <EOL> import threading <EOL> import unittest2 as unittest <EOL> from cactus . listener import PollingListener <EOL> from cactus . tests . compat import has_symlink <EOL> try : <EOL> from cactus . listener import FSEventsListener <EOL> except ImportError : <EOL> FSEventsListener = None <EOL> def sleep ( s ) : <EOL> event = threading . Event ( ) <EOL> event . wait ( timeout = s ) <EOL> class PollingListenerTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _callbacks = [ ] <EOL> self . _callback_count = <NUM_LIT:0> <EOL> self . lock = threading . Lock ( ) <EOL> def _callback ( self , event ) : <EOL> with self . lock : <EOL> self . _callbacks . append ( event ) <EOL> @ property <EOL> def callbacks ( self ) : <EOL> with self . lock : <EOL> return self . _callbacks <EOL> def wait ( self ) : <EOL> time_total = <NUM_LIT:0> <EOL> time_increment = <NUM_LIT> <EOL> time_max = <NUM_LIT> <EOL> while True : <EOL> self . assertEqual ( time_total < time_max , True , "<STR_LIT>" ) <EOL> if len ( self . callbacks ) == self . _callback_count + <NUM_LIT:1> : <EOL> self . _callback_count += <NUM_LIT:1> <EOL> return <EOL> sleep ( time_increment ) <EOL> time_total += time_increment <EOL> def create_listener ( self , path ) : <EOL> return PollingListener ( path , self . _callback ) <EOL> @ unittest . skipUnless ( has_symlink , "<STR_LIT>" ) <EOL> def testSymlinkFolder ( self ) : <EOL> path1 = os . path . realpath ( os . path . join ( tempfile . mkdtemp ( ) , "<STR_LIT>" ) ) <EOL> path2 = os . path . realpath ( os . path . join ( tempfile . mkdtemp ( ) , "<STR_LIT>" ) ) <EOL> os . mkdir ( path1 ) <EOL> os . mkdir ( path2 ) <EOL> path_link = os . path . join ( path1 , "<STR_LIT>" ) <EOL> path_link_file = os . path . join ( path2 , "<STR_LIT>" ) <EOL> path_link_file_original = os . path . join ( path_link , "<STR_LIT>" ) <EOL> os . symlink ( path2 , path_link ) <EOL> with open ( path_link_file , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> self . listener = self . create_listener ( path1 ) <EOL> self . listener . run ( ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> with open ( path_link_file , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> self . wait ( ) <EOL> self . assertEqual ( len ( self . callbacks ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . callbacks [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , [ path_link_file_original ] ) <EOL> def testSimpleFile ( self ) : <EOL> path_watch = os . path . realpath ( os . path . join ( tempfile . mkdtemp ( ) , "<STR_LIT>" ) ) <EOL> path_file = os . path . join ( path_watch , "<STR_LIT>" ) <EOL> os . mkdir ( path_watch ) <EOL> self . listener = self . create_listener ( path_watch ) <EOL> self . listener . run ( ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> with open ( path_file , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> self . wait ( ) <EOL> self . assertEqual ( len ( self . callbacks ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . callbacks [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , [ path_file ] ) <EOL> os . remove ( path_file ) <EOL> self . wait ( ) <EOL> self . assertEqual ( len ( self . callbacks ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( self . callbacks [ <NUM_LIT:1> ] [ "<STR_LIT>" ] , [ path_file ] ) <EOL> @ unittest . skipUnless ( has_symlink , "<STR_LIT>" ) <EOL> def testSymlinkFile ( self ) : <EOL> path_watch = os . path . realpath ( os . path . join ( tempfile . mkdtemp ( ) , "<STR_LIT>" ) ) <EOL> path_linked = os . path . realpath ( os . path . join ( tempfile . mkdtemp ( ) , "<STR_LIT>" ) ) <EOL> os . mkdir ( path_watch ) <EOL> os . mkdir ( path_linked ) <EOL> file_actual1 = os . path . join ( path_linked , "<STR_LIT>" ) <EOL> file_actual2 = os . path . join ( path_linked , "<STR_LIT>" ) <EOL> file_link = os . path . join ( path_watch , "<STR_LIT>" ) <EOL> with open ( file_actual1 , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> with open ( file_actual2 , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> os . symlink ( file_actual1 , file_link ) <EOL> self . listener = self . create_listener ( path_watch ) <EOL> self . listener . run ( ) <EOL> if hasattr ( self . listener , "<STR_LIT>" ) : <EOL> self . assertEqual ( len ( self . listener . streamed_folders ) , <NUM_LIT:2> ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> with open ( file_actual1 , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> self . wait ( ) <EOL> self . assertEqual ( len ( self . callbacks ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . callbacks [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , [ file_link ] ) <EOL> with open ( file_actual2 , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> with open ( file_actual1 , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> self . wait ( ) <EOL> self . assertEqual ( len ( self . callbacks ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( self . callbacks [ <NUM_LIT:0> ] [ "<STR_LIT>" ] , [ file_link ] ) <EOL> if FSEventsListener : <EOL> class FSEventsListenerTest ( PollingListener ) : <EOL> def create_listener ( self , path ) : <EOL> return FSEventsListener ( path , self . _callback ) </s>
