<s> """<STR_LIT>""" <EOL> import os , errno <EOL> from twisted . trial import unittest <EOL> from twisted . python import lockfile <EOL> from twisted . python . runtime import platform <EOL> skipKill = None <EOL> if platform . isWindows ( ) : <EOL> try : <EOL> from win32api import OpenProcess <EOL> import pywintypes <EOL> except ImportError : <EOL> skipKill = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class UtilTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_symlinkEEXIST ( self ) : <EOL> """<STR_LIT>""" <EOL> name = self . mktemp ( ) <EOL> lockfile . symlink ( '<STR_LIT:foo>' , name ) <EOL> exc = self . assertRaises ( OSError , lockfile . symlink , '<STR_LIT:foo>' , name ) <EOL> self . assertEqual ( exc . errno , errno . EEXIST ) <EOL> def test_symlinkEIOWindows ( self ) : <EOL> """<STR_LIT>""" <EOL> name = self . mktemp ( ) <EOL> def fakeRename ( src , dst ) : <EOL> raise IOError ( errno . EIO , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeRename ) <EOL> exc = self . assertRaises ( IOError , lockfile . symlink , name , "<STR_LIT:foo>" ) <EOL> self . assertEqual ( exc . errno , errno . EIO ) <EOL> if not platform . isWindows ( ) : <EOL> test_symlinkEIOWindows . skip = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_readlinkENOENT ( self ) : <EOL> """<STR_LIT>""" <EOL> name = self . mktemp ( ) <EOL> exc = self . assertRaises ( OSError , lockfile . readlink , name ) <EOL> self . assertEqual ( exc . errno , errno . ENOENT ) <EOL> def test_readlinkEACCESWindows ( self ) : <EOL> """<STR_LIT>""" <EOL> name = self . mktemp ( ) <EOL> def fakeOpen ( path , mode ) : <EOL> raise IOError ( errno . EACCES , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeOpen ) <EOL> exc = self . assertRaises ( IOError , lockfile . readlink , name ) <EOL> self . assertEqual ( exc . errno , errno . EACCES ) <EOL> if not platform . isWindows ( ) : <EOL> test_readlinkEACCESWindows . skip = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_kill ( self ) : <EOL> """<STR_LIT>""" <EOL> lockfile . kill ( os . getpid ( ) , <NUM_LIT:0> ) <EOL> test_kill . skip = skipKill <EOL> def test_killESRCH ( self ) : <EOL> """<STR_LIT>""" <EOL> exc = self . assertRaises ( OSError , lockfile . kill , <NUM_LIT:2> ** <NUM_LIT> - <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . assertEqual ( exc . errno , errno . ESRCH ) <EOL> test_killESRCH . skip = skipKill <EOL> def test_noKillCall ( self ) : <EOL> """<STR_LIT>""" <EOL> self . patch ( lockfile , "<STR_LIT>" , None ) <EOL> fl = lockfile . FilesystemLock ( self . mktemp ( ) ) <EOL> fl . lock ( ) <EOL> self . assertFalse ( fl . lock ( ) ) <EOL> class LockingTestCase ( unittest . TestCase ) : <EOL> def _symlinkErrorTest ( self , errno ) : <EOL> def fakeSymlink ( source , dest ) : <EOL> raise OSError ( errno , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeSymlink ) <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> exc = self . assertRaises ( OSError , lock . lock ) <EOL> self . assertEqual ( exc . errno , errno ) <EOL> def test_symlinkError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _symlinkErrorTest ( errno . ENOSYS ) <EOL> def test_symlinkErrorPOSIX ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _symlinkErrorTest ( errno . EACCES ) <EOL> self . _symlinkErrorTest ( errno . EIO ) <EOL> if platform . isWindows ( ) : <EOL> test_symlinkErrorPOSIX . skip = ( <EOL> "<STR_LIT>" ) <EOL> def test_cleanlyAcquire ( self ) : <EOL> """<STR_LIT>""" <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertTrue ( lock . clean ) <EOL> self . assertTrue ( lock . locked ) <EOL> def test_cleanlyRelease ( self ) : <EOL> """<STR_LIT>""" <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> lock . unlock ( ) <EOL> self . assertFalse ( lock . locked ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertTrue ( lock . clean ) <EOL> self . assertTrue ( lock . locked ) <EOL> def test_cannotLockLocked ( self ) : <EOL> """<STR_LIT>""" <EOL> lockf = self . mktemp ( ) <EOL> firstLock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( firstLock . lock ( ) ) <EOL> secondLock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertFalse ( secondLock . lock ( ) ) <EOL> self . assertFalse ( secondLock . locked ) <EOL> def test_uncleanlyAcquire ( self ) : <EOL> """<STR_LIT>""" <EOL> owner = <NUM_LIT> <EOL> def fakeKill ( pid , signal ) : <EOL> if signal != <NUM_LIT:0> : <EOL> raise OSError ( errno . EPERM , None ) <EOL> if pid == owner : <EOL> raise OSError ( errno . ESRCH , None ) <EOL> lockf = self . mktemp ( ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeKill ) <EOL> lockfile . symlink ( str ( owner ) , lockf ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertFalse ( lock . clean ) <EOL> self . assertTrue ( lock . locked ) <EOL> self . assertEqual ( lockfile . readlink ( lockf ) , str ( os . getpid ( ) ) ) <EOL> def test_lockReleasedBeforeCheck ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeReadlink ( name ) : <EOL> lockfile . rmlink ( lockf ) <EOL> readlinkPatch . restore ( ) <EOL> return lockfile . readlink ( name ) <EOL> readlinkPatch = self . patch ( lockfile , '<STR_LIT>' , fakeReadlink ) <EOL> def fakeKill ( pid , signal ) : <EOL> if signal != <NUM_LIT:0> : <EOL> raise OSError ( errno . EPERM , None ) <EOL> if pid == <NUM_LIT> : <EOL> raise OSError ( errno . ESRCH , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeKill ) <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertTrue ( lock . clean ) <EOL> self . assertTrue ( lock . locked ) <EOL> def test_lockReleasedDuringAcquireSymlink ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeSymlink ( src , dst ) : <EOL> raise OSError ( errno . EIO , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeSymlink ) <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertFalse ( lock . lock ( ) ) <EOL> self . assertFalse ( lock . locked ) <EOL> if not platform . isWindows ( ) : <EOL> test_lockReleasedDuringAcquireSymlink . skip = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_lockReleasedDuringAcquireReadlink ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeReadlink ( name ) : <EOL> raise IOError ( errno . EACCES , None ) <EOL> readlinkPatch = self . patch ( lockfile , '<STR_LIT>' , fakeReadlink ) <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> self . assertFalse ( lock . lock ( ) ) <EOL> self . assertFalse ( lock . locked ) <EOL> if not platform . isWindows ( ) : <EOL> test_lockReleasedDuringAcquireReadlink . skip = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def _readlinkErrorTest ( self , exceptionType , errno ) : <EOL> def fakeReadlink ( name ) : <EOL> raise exceptionType ( errno , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeReadlink ) <EOL> lockf = self . mktemp ( ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> exc = self . assertRaises ( exceptionType , lock . lock ) <EOL> self . assertEqual ( exc . errno , errno ) <EOL> self . assertFalse ( lock . locked ) <EOL> def test_readlinkError ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _readlinkErrorTest ( OSError , errno . ENOSYS ) <EOL> self . _readlinkErrorTest ( IOError , errno . ENOSYS ) <EOL> def test_readlinkErrorPOSIX ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _readlinkErrorTest ( IOError , errno . ENOSYS ) <EOL> self . _readlinkErrorTest ( IOError , errno . EACCES ) <EOL> if platform . isWindows ( ) : <EOL> test_readlinkErrorPOSIX . skip = ( <EOL> "<STR_LIT>" ) <EOL> def test_lockCleanedUpConcurrently ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeRmlink ( name ) : <EOL> rmlinkPatch . restore ( ) <EOL> lockfile . rmlink ( lockf ) <EOL> return lockfile . rmlink ( name ) <EOL> rmlinkPatch = self . patch ( lockfile , '<STR_LIT>' , fakeRmlink ) <EOL> def fakeKill ( pid , signal ) : <EOL> if signal != <NUM_LIT:0> : <EOL> raise OSError ( errno . EPERM , None ) <EOL> if pid == <NUM_LIT> : <EOL> raise OSError ( errno . ESRCH , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeKill ) <EOL> lockf = self . mktemp ( ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertTrue ( lock . clean ) <EOL> self . assertTrue ( lock . locked ) <EOL> def test_rmlinkError ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeRmlink ( name ) : <EOL> raise OSError ( errno . ENOSYS , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeRmlink ) <EOL> def fakeKill ( pid , signal ) : <EOL> if signal != <NUM_LIT:0> : <EOL> raise OSError ( errno . EPERM , None ) <EOL> if pid == <NUM_LIT> : <EOL> raise OSError ( errno . ESRCH , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeKill ) <EOL> lockf = self . mktemp ( ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> exc = self . assertRaises ( OSError , lock . lock ) <EOL> self . assertEqual ( exc . errno , errno . ENOSYS ) <EOL> self . assertFalse ( lock . locked ) <EOL> def test_killError ( self ) : <EOL> """<STR_LIT>""" <EOL> def fakeKill ( pid , signal ) : <EOL> raise OSError ( errno . EPERM , None ) <EOL> self . patch ( lockfile , '<STR_LIT>' , fakeKill ) <EOL> lockf = self . mktemp ( ) <EOL> lockfile . symlink ( str ( <NUM_LIT> ) , lockf ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> exc = self . assertRaises ( OSError , lock . lock ) <EOL> self . assertEqual ( exc . errno , errno . EPERM ) <EOL> self . assertFalse ( lock . locked ) <EOL> def test_unlockOther ( self ) : <EOL> """<STR_LIT>""" <EOL> lockf = self . mktemp ( ) <EOL> lockfile . symlink ( str ( os . getpid ( ) + <NUM_LIT:1> ) , lockf ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertRaises ( ValueError , lock . unlock ) <EOL> def test_isLocked ( self ) : <EOL> """<STR_LIT>""" <EOL> lockf = self . mktemp ( ) <EOL> self . assertFalse ( lockfile . isLocked ( lockf ) ) <EOL> lock = lockfile . FilesystemLock ( lockf ) <EOL> self . assertTrue ( lock . lock ( ) ) <EOL> self . assertTrue ( lockfile . isLocked ( lockf ) ) <EOL> lock . unlock ( ) <EOL> self . assertFalse ( lockfile . isLocked ( lockf ) ) </s>
