<s> """<STR_LIT>""" <EOL> from twisted . internet import defer , reactor <EOL> from twisted . python import filepath , runtime <EOL> from twisted . python . reflect import requireModule <EOL> from twisted . trial import unittest <EOL> if requireModule ( '<STR_LIT>' ) is not None : <EOL> from twisted . internet import inotify <EOL> else : <EOL> inotify = None <EOL> class INotifyTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> if not runtime . platform . supportsINotify ( ) : <EOL> skip = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> self . dirname = filepath . FilePath ( self . mktemp ( ) ) <EOL> self . dirname . createDirectory ( ) <EOL> self . inotify = inotify . INotify ( ) <EOL> self . inotify . startReading ( ) <EOL> self . addCleanup ( self . inotify . loseConnection ) <EOL> def test_initializationErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> class FakeINotify : <EOL> def init ( self ) : <EOL> raise inotify . INotifyError ( ) <EOL> self . patch ( inotify . INotify , '<STR_LIT>' , FakeINotify ( ) ) <EOL> self . assertRaises ( inotify . INotifyError , inotify . INotify ) <EOL> def _notificationTest ( self , mask , operation , expectedPath = None ) : <EOL> """<STR_LIT>""" <EOL> if expectedPath is None : <EOL> expectedPath = self . dirname . child ( "<STR_LIT>" ) <EOL> notified = defer . Deferred ( ) <EOL> def cbNotified ( ( watch , filename , events ) ) : <EOL> self . assertEqual ( filename , expectedPath ) <EOL> self . assertTrue ( events & mask ) <EOL> notified . addCallback ( cbNotified ) <EOL> self . inotify . watch ( <EOL> self . dirname , mask = mask , <EOL> callbacks = [ lambda * args : notified . callback ( args ) ] ) <EOL> operation ( expectedPath ) <EOL> return notified <EOL> def test_access ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . setContent ( "<STR_LIT:foo>" ) <EOL> path . getContent ( ) <EOL> return self . _notificationTest ( inotify . IN_ACCESS , operation ) <EOL> def test_modify ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> fObj = path . open ( "<STR_LIT:w>" ) <EOL> fObj . write ( '<STR_LIT:foo>' ) <EOL> fObj . close ( ) <EOL> return self . _notificationTest ( inotify . IN_MODIFY , operation ) <EOL> def test_attrib ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . touch ( ) <EOL> path . touch ( ) <EOL> return self . _notificationTest ( inotify . IN_ATTRIB , operation ) <EOL> def test_closeWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> fObj = path . open ( "<STR_LIT:w>" ) <EOL> fObj . close ( ) <EOL> return self . _notificationTest ( inotify . IN_CLOSE_WRITE , operation ) <EOL> def test_closeNoWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . touch ( ) <EOL> fObj = path . open ( "<STR_LIT:r>" ) <EOL> fObj . close ( ) <EOL> return self . _notificationTest ( inotify . IN_CLOSE_NOWRITE , operation ) <EOL> def test_open ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> fObj = path . open ( "<STR_LIT:w>" ) <EOL> fObj . close ( ) <EOL> return self . _notificationTest ( inotify . IN_OPEN , operation ) <EOL> def test_movedFrom ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> fObj = path . open ( "<STR_LIT:w>" ) <EOL> fObj . close ( ) <EOL> path . moveTo ( filepath . FilePath ( self . mktemp ( ) ) ) <EOL> return self . _notificationTest ( inotify . IN_MOVED_FROM , operation ) <EOL> def test_movedTo ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> p = filepath . FilePath ( self . mktemp ( ) ) <EOL> p . touch ( ) <EOL> p . moveTo ( path ) <EOL> return self . _notificationTest ( inotify . IN_MOVED_TO , operation ) <EOL> def test_create ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> fObj = path . open ( "<STR_LIT:w>" ) <EOL> fObj . close ( ) <EOL> return self . _notificationTest ( inotify . IN_CREATE , operation ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . touch ( ) <EOL> path . remove ( ) <EOL> return self . _notificationTest ( inotify . IN_DELETE , operation ) <EOL> def test_deleteSelf ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . remove ( ) <EOL> return self . _notificationTest ( <EOL> inotify . IN_DELETE_SELF , operation , expectedPath = self . dirname ) <EOL> def test_moveSelf ( self ) : <EOL> """<STR_LIT>""" <EOL> def operation ( path ) : <EOL> path . moveTo ( filepath . FilePath ( self . mktemp ( ) ) ) <EOL> return self . _notificationTest ( <EOL> inotify . IN_MOVE_SELF , operation , expectedPath = self . dirname ) <EOL> def test_simpleSubdirectoryAutoAdd ( self ) : <EOL> """<STR_LIT>""" <EOL> def _callback ( wp , filename , mask ) : <EOL> def _ ( ) : <EOL> try : <EOL> self . assertTrue ( self . inotify . _isWatched ( subdir ) ) <EOL> d . callback ( None ) <EOL> except Exception : <EOL> d . errback ( ) <EOL> reactor . callLater ( <NUM_LIT:0> , _ ) <EOL> checkMask = inotify . IN_ISDIR | inotify . IN_CREATE <EOL> self . inotify . watch ( <EOL> self . dirname , mask = checkMask , autoAdd = True , <EOL> callbacks = [ _callback ] ) <EOL> subdir = self . dirname . child ( '<STR_LIT:test>' ) <EOL> d = defer . Deferred ( ) <EOL> subdir . createDirectory ( ) <EOL> return d <EOL> def test_simpleDeleteDirectory ( self ) : <EOL> """<STR_LIT>""" <EOL> calls = [ ] <EOL> def _callback ( wp , filename , mask ) : <EOL> def _ ( ) : <EOL> try : <EOL> self . assertTrue ( self . inotify . _isWatched ( subdir ) ) <EOL> subdir . remove ( ) <EOL> except Exception : <EOL> d . errback ( ) <EOL> def _eb ( ) : <EOL> try : <EOL> self . assertTrue ( not self . inotify . _isWatched ( subdir ) ) <EOL> d . callback ( None ) <EOL> except Exception : <EOL> d . errback ( ) <EOL> if not calls : <EOL> calls . append ( filename ) <EOL> reactor . callLater ( <NUM_LIT:0> , _ ) <EOL> else : <EOL> reactor . callLater ( <NUM_LIT:0> , _eb ) <EOL> checkMask = inotify . IN_ISDIR | inotify . IN_CREATE <EOL> self . inotify . watch ( <EOL> self . dirname , mask = checkMask , autoAdd = True , <EOL> callbacks = [ _callback ] ) <EOL> subdir = self . dirname . child ( '<STR_LIT:test>' ) <EOL> d = defer . Deferred ( ) <EOL> subdir . createDirectory ( ) <EOL> return d <EOL> def test_ignoreDirectory ( self ) : <EOL> """<STR_LIT>""" <EOL> self . inotify . watch ( self . dirname , autoAdd = True ) <EOL> self . assertTrue ( self . inotify . _isWatched ( self . dirname ) ) <EOL> self . inotify . ignore ( self . dirname ) <EOL> self . assertFalse ( self . inotify . _isWatched ( self . dirname ) ) <EOL> def test_humanReadableMask ( self ) : <EOL> """<STR_LIT>""" <EOL> for mask , value in inotify . _FLAG_TO_HUMAN : <EOL> self . assertEqual ( inotify . humanReadableMask ( mask ) [ <NUM_LIT:0> ] , value ) <EOL> checkMask = ( <EOL> inotify . IN_CLOSE_WRITE | inotify . IN_ACCESS | inotify . IN_OPEN ) <EOL> self . assertEqual ( <EOL> set ( inotify . humanReadableMask ( checkMask ) ) , <EOL> set ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> def test_recursiveWatch ( self ) : <EOL> """<STR_LIT>""" <EOL> subdir = self . dirname . child ( '<STR_LIT:test>' ) <EOL> subdir2 = subdir . child ( '<STR_LIT>' ) <EOL> subdir3 = subdir2 . child ( '<STR_LIT>' ) <EOL> subdir3 . makedirs ( ) <EOL> dirs = [ subdir , subdir2 , subdir3 ] <EOL> self . inotify . watch ( self . dirname , recursive = True ) <EOL> self . inotify . watch ( self . dirname , recursive = True ) <EOL> for d in dirs : <EOL> self . assertTrue ( self . inotify . _isWatched ( d ) ) <EOL> def test_connectionLostError ( self ) : <EOL> """<STR_LIT>""" <EOL> import os <EOL> in_ = inotify . INotify ( ) <EOL> os . close ( in_ . _fd ) <EOL> in_ . loseConnection ( ) <EOL> self . flushLoggedErrors ( ) <EOL> def test_noAutoAddSubdirectory ( self ) : <EOL> """<STR_LIT>""" <EOL> def _callback ( wp , fp , mask ) : <EOL> def _ ( ) : <EOL> try : <EOL> self . assertFalse ( self . inotify . _isWatched ( subdir . path ) ) <EOL> d . callback ( None ) <EOL> except Exception : <EOL> d . errback ( ) <EOL> reactor . callLater ( <NUM_LIT:0> , _ ) <EOL> checkMask = inotify . IN_ISDIR | inotify . IN_CREATE <EOL> self . inotify . watch ( <EOL> self . dirname , mask = checkMask , autoAdd = False , <EOL> callbacks = [ _callback ] ) <EOL> subdir = self . dirname . child ( '<STR_LIT:test>' ) <EOL> d = defer . Deferred ( ) <EOL> subdir . createDirectory ( ) <EOL> return d <EOL> def test_seriesOfWatchAndIgnore ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedPath = self . dirname . child ( "<STR_LIT>" ) <EOL> expectedPath . touch ( ) <EOL> notified = defer . Deferred ( ) <EOL> def cbNotified ( ( ignored , filename , events ) ) : <EOL> self . assertEqual ( filename , expectedPath ) <EOL> self . assertTrue ( events & inotify . IN_DELETE_SELF ) <EOL> def callIt ( * args ) : <EOL> notified . callback ( args ) <EOL> self . assertTrue ( self . inotify . watch ( expectedPath , callbacks = [ callIt ] ) ) <EOL> self . inotify . ignore ( expectedPath ) <EOL> self . assertTrue ( <EOL> self . inotify . watch ( <EOL> expectedPath , mask = inotify . IN_DELETE_SELF , callbacks = [ callIt ] ) ) <EOL> notified . addCallback ( cbNotified ) <EOL> expectedPath . remove ( ) <EOL> return notified <EOL> def test_ignoreFilePath ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedPath = self . dirname . child ( "<STR_LIT>" ) <EOL> expectedPath . touch ( ) <EOL> expectedPath2 = self . dirname . child ( "<STR_LIT>" ) <EOL> expectedPath2 . touch ( ) <EOL> notified = defer . Deferred ( ) <EOL> def cbNotified ( ( ignored , filename , events ) ) : <EOL> self . assertEqual ( filename , expectedPath2 ) <EOL> self . assertTrue ( events & inotify . IN_DELETE_SELF ) <EOL> def callIt ( * args ) : <EOL> notified . callback ( args ) <EOL> self . assertTrue ( <EOL> self . inotify . watch ( <EOL> expectedPath , inotify . IN_DELETE_SELF , callbacks = [ callIt ] ) ) <EOL> notified . addCallback ( cbNotified ) <EOL> self . assertTrue ( <EOL> self . inotify . watch ( <EOL> expectedPath2 , inotify . IN_DELETE_SELF , callbacks = [ callIt ] ) ) <EOL> self . inotify . ignore ( expectedPath ) <EOL> expectedPath . remove ( ) <EOL> expectedPath2 . remove ( ) <EOL> return notified <EOL> def test_ignoreNonWatchedFile ( self ) : <EOL> """<STR_LIT>""" <EOL> expectedPath = self . dirname . child ( "<STR_LIT>" ) <EOL> expectedPath . touch ( ) <EOL> self . assertRaises ( KeyError , self . inotify . ignore , expectedPath ) <EOL> def test_complexSubdirectoryAutoAdd ( self ) : <EOL> """<STR_LIT>""" <EOL> calls = set ( ) <EOL> def _callback ( wp , filename , mask ) : <EOL> calls . add ( filename ) <EOL> if len ( calls ) == <NUM_LIT:6> : <EOL> try : <EOL> self . assertTrue ( self . inotify . _isWatched ( subdir ) ) <EOL> self . assertTrue ( self . inotify . _isWatched ( subdir2 ) ) <EOL> self . assertTrue ( self . inotify . _isWatched ( subdir3 ) ) <EOL> created = someFiles + [ subdir , subdir2 , subdir3 ] <EOL> self . assertEqual ( len ( calls ) , len ( created ) ) <EOL> self . assertEqual ( calls , set ( created ) ) <EOL> except Exception : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> checkMask = inotify . IN_ISDIR | inotify . IN_CREATE <EOL> self . inotify . watch ( <EOL> self . dirname , mask = checkMask , autoAdd = True , <EOL> callbacks = [ _callback ] ) <EOL> subdir = self . dirname . child ( '<STR_LIT:test>' ) <EOL> subdir2 = subdir . child ( '<STR_LIT>' ) <EOL> subdir3 = subdir2 . child ( '<STR_LIT>' ) <EOL> d = defer . Deferred ( ) <EOL> subdir3 . makedirs ( ) <EOL> someFiles = [ subdir . child ( '<STR_LIT>' ) , <EOL> subdir2 . child ( '<STR_LIT>' ) , <EOL> subdir3 . child ( '<STR_LIT>' ) ] <EOL> for i , filename in enumerate ( someFiles ) : <EOL> filename . setContent ( filename . path ) <EOL> return d </s>
