Gecko [ aodai.devttt.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	475 B	-rw-r--r--
cert.pem.no_trailing_newline	1.38 KB	-rw-r--r--
crash_test_dummy.py	549 B	-rw-r--r--
ignore_test_failure.py	30.72 KB	-rw-r--r--
iosim.py	17.74 KB	-rw-r--r--
key.pem.no_trailing_newline	1.67 KB	-rw-r--r--
mock_win32process.py	1.27 KB	-rw-r--r--
myrebuilder1.py	151 B	-rw-r--r--
myrebuilder2.py	151 B	-rw-r--r--
plugin_basic.py	901 B	-rw-r--r--
plugin_extra1.py	392 B	-rw-r--r--
plugin_extra2.py	550 B	-rw-r--r--
process_cmdline.py	123 B	-rw-r--r--
process_echoer.py	214 B	-rw-r--r--
process_fds.py	984 B	-rw-r--r--
process_getargv.py	234 B	-rw-r--r--
process_getenv.py	268 B	-rw-r--r--
process_linger.py	297 B	-rw-r--r--
process_reader.py	178 B	-rw-r--r--
process_signal.py	220 B	-rw-r--r--
process_stdinreader.py	739 B	-rw-r--r--
process_tester.py	779 B	-rw-r--r--
process_tty.py	130 B	-rw-r--r--
process_twisted.py	1.15 KB	-rw-r--r--
proto_helpers.py	1.34 KB	-rw-r--r--
reflect_helper_IE.py	60 B	-rw-r--r--
reflect_helper_VE.py	81 B	-rw-r--r--
reflect_helper_ZDE.py	48 B	-rw-r--r--
server.pem	4.34 KB	-rw-r--r--
ssl_helpers.py	1.61 KB	-rw-r--r--
stdio_test_consumer.py	1.14 KB	-rw-r--r--
stdio_test_halfclose.py	2 KB	-rw-r--r--
stdio_test_hostpeer.py	1.06 KB	-rw-r--r--
stdio_test_lastwrite.py	1.13 KB	-rw-r--r--
stdio_test_loseconn.py	1.55 KB	-rw-r--r--
stdio_test_producer.py	1.45 KB	-rw-r--r--
stdio_test_write.py	902 B	-rw-r--r--
stdio_test_writeseq.py	894 B	-rw-r--r--
test_abstract.py	3.35 KB	-rw-r--r--
test_adbapi.py	25.56 KB	-rw-r--r--
test_amp.py	107.92 KB	-rw-r--r--
test_application.py	33.34 KB	-rw-r--r--
test_compat.py	17.75 KB	-rw-r--r--
test_context.py	1.43 KB	-rw-r--r--
test_cooperator.py	20.84 KB	-rw-r--r--
test_defer.py	121.86 KB	-rw-r--r--
test_defgen.py	10.38 KB	-rw-r--r--
test_dict.py	1.49 KB	-rw-r--r--
test_dirdbm.py	6.69 KB	-rw-r--r--
test_error.py	8.72 KB	-rw-r--r--
test_factories.py	4.46 KB	-rw-r--r--
test_fdesc.py	7.28 KB	-rw-r--r--
test_finger.py	1.85 KB	-rw-r--r--
test_formmethod.py	3.87 KB	-rw-r--r--
test_ftp.py	126.96 KB	-rw-r--r--
test_ftp_options.py	2.63 KB	-rw-r--r--
test_htb.py	3.08 KB	-rw-r--r--
test_ident.py	6.56 KB	-rw-r--r--
test_internet.py	45.38 KB	-rw-r--r--
test_iosim.py	8.7 KB	-rw-r--r--
test_iutils.py	13.31 KB	-rw-r--r--
test_lockfile.py	14.98 KB	-rw-r--r--
test_log.py	35.23 KB	-rw-r--r--
test_logfile.py	17.47 KB	-rw-r--r--
test_loopback.py	13.99 KB	-rw-r--r--
test_main.py	2.12 KB	-rw-r--r--
test_memcache.py	24.69 KB	-rw-r--r--
test_modules.py	17.02 KB	-rw-r--r--
test_monkey.py	5.39 KB	-rw-r--r--
test_paths.py	69.31 KB	-rw-r--r--
test_pcp.py	12.23 KB	-rw-r--r--
test_persisted.py	13.09 KB	-rw-r--r--
test_plugin.py	24.79 KB	-rw-r--r--
test_policies.py	32.29 KB	-rw-r--r--
test_postfix.py	4.32 KB	-rw-r--r--
test_process.py	85.16 KB	-rw-r--r--
test_protocols.py	7.16 KB	-rw-r--r--
test_randbytes.py	3.19 KB	-rw-r--r--
test_rebuild.py	7.15 KB	-rw-r--r--
test_reflect.py	23.9 KB	-rw-r--r--
test_roots.py	1.6 KB	-rw-r--r--
test_shortcut.py	1.92 KB	-rw-r--r--
test_sip.py	24.9 KB	-rw-r--r--
test_sob.py	5.53 KB	-rw-r--r--
test_socks.py	17.09 KB	-rw-r--r--
test_ssl.py	22.75 KB	-rw-r--r--
test_sslverify.py	113.89 KB	-rw-r--r--
test_stateful.py	1.97 KB	-rw-r--r--
test_stdio.py	12.43 KB	-rw-r--r--
test_strerror.py	5.12 KB	-rw-r--r--
test_strports.py	1.67 KB	-rw-r--r--
test_task.py	45.26 KB	-rw-r--r--
test_tcp.py	64.27 KB	-rw-r--r--
test_tcp_internals.py	12.73 KB	-rw-r--r--
test_text.py	6.22 KB	-rw-r--r--
test_threadable.py	3.26 KB	-rw-r--r--
test_threadpool.py	21.44 KB	-rw-r--r--
test_threads.py	12.9 KB	-rw-r--r--
test_tpfile.py	1.51 KB	-rw-r--r--
test_twistd.py	72.04 KB	-rw-r--r--
test_twisted.py	6.13 KB	-rw-r--r--
test_udp.py	24.43 KB	-rw-r--r--
test_unix.py	13.26 KB	-rw-r--r--
test_usage.py	22.76 KB	-rw-r--r--
testutils.py	5.06 KB	-rw-r--r--

Code Editor : test_lockfile.py

# Copyright (c) 2005 Divmod, Inc.
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.python.lockfile}.
"""

import errno
import os
from unittest import skipIf, skipUnless

from twisted.python import lockfile
from twisted.python.reflect import requireModule
from twisted.python.runtime import platform
from twisted.trial.unittest import TestCase

skipKill = False
skipKillReason = ""
if platform.isWindows():
    if (
        requireModule("win32api.OpenProcess") is None
        and requireModule("pywintypes") is None
    ):
        skipKill = True
        skipKillReason = (
            "On windows, lockfile.kill is not implemented "
            "in the absence of win32api and/or pywintypes."
        )

class UtilTests(TestCase):
    """
    Tests for the helper functions used to implement L{FilesystemLock}.
    """

def test_symlinkEEXIST(self):
        """
        L{lockfile.symlink} raises L{OSError} with C{errno} set to L{EEXIST}
        when an attempt is made to create a symlink which already exists.
        """
        name = self.mktemp()
        lockfile.symlink("foo", name)
        exc = self.assertRaises(OSError, lockfile.symlink, "foo", name)
        self.assertEqual(exc.errno, errno.EEXIST)

@skipUnless(
        platform.isWindows(),
        "special rename EIO handling only necessary and correct on " "Windows.",
    )
    def test_symlinkEIOWindows(self):
        """
        L{lockfile.symlink} raises L{OSError} with C{errno} set to L{EIO} when
        the underlying L{rename} call fails with L{EIO}.

Renaming a file on Windows may fail if the target of the rename is in
        the process of being deleted (directory deletion appears not to be
        atomic).
        """
        name = self.mktemp()

def fakeRename(src, dst):
            raise OSError(errno.EIO, None)

self.patch(lockfile, "rename", fakeRename)
        exc = self.assertRaises(IOError, lockfile.symlink, name, "foo")
        self.assertEqual(exc.errno, errno.EIO)

def test_readlinkENOENT(self):
        """
        L{lockfile.readlink} raises L{OSError} with C{errno} set to L{ENOENT}
        when an attempt is made to read a symlink which does not exist.
        """
        name = self.mktemp()
        exc = self.assertRaises(OSError, lockfile.readlink, name)
        self.assertEqual(exc.errno, errno.ENOENT)

@skipUnless(
        platform.isWindows(),
        "special readlink EACCES handling only necessary and " "correct on Windows.",
    )
    def test_readlinkEACCESWindows(self):
        """
        L{lockfile.readlink} raises L{OSError} with C{errno} set to L{EACCES}
        on Windows when the underlying file open attempt fails with C{EACCES}.

Opening a file on Windows may fail if the path is inside a directory
        which is in the process of being deleted (directory deletion appears
        not to be atomic).
        """
        name = self.mktemp()

def fakeOpen(path, mode):
            raise OSError(errno.EACCES, None)

self.patch(lockfile, "_open", fakeOpen)
        exc = self.assertRaises(IOError, lockfile.readlink, name)
        self.assertEqual(exc.errno, errno.EACCES)

@skipIf(skipKill, skipKillReason)
    def test_kill(self):
        """
        L{lockfile.kill} returns without error if passed the PID of a
        process which exists and signal C{0}.
        """
        lockfile.kill(os.getpid(), 0)

@skipIf(skipKill, skipKillReason)
    def test_killESRCH(self):
        """
        L{lockfile.kill} raises L{OSError} with errno of L{ESRCH} if
        passed a PID which does not correspond to any process.
        """
        # Hopefully there is no process with PID 2 ** 31 - 1
        exc = self.assertRaises(OSError, lockfile.kill, 2 ** 31 - 1, 0)
        self.assertEqual(exc.errno, errno.ESRCH)

def test_noKillCall(self):
        """
        Verify that when L{lockfile.kill} does end up as None (e.g. on Windows
        without pywin32), it doesn't end up being called and raising a
        L{TypeError}.
        """
        self.patch(lockfile, "kill", None)
        fl = lockfile.FilesystemLock(self.mktemp())
        fl.lock()
        self.assertFalse(fl.lock())

class LockingTests(TestCase):
    def _symlinkErrorTest(self, errno):
        def fakeSymlink(source, dest):
            raise OSError(errno, None)

self.patch(lockfile, "symlink", fakeSymlink)

lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        exc = self.assertRaises(OSError, lock.lock)
        self.assertEqual(exc.errno, errno)

def test_symlinkError(self):
        """
        An exception raised by C{symlink} other than C{EEXIST} is passed up to
        the caller of L{FilesystemLock.lock}.
        """
        self._symlinkErrorTest(errno.ENOSYS)

@skipIf(
        platform.isWindows(),
        "POSIX-specific error propagation not expected on Windows.",
    )
    def test_symlinkErrorPOSIX(self):
        """
        An L{OSError} raised by C{symlink} on a POSIX platform with an errno of
        C{EACCES} or C{EIO} is passed to the caller of L{FilesystemLock.lock}.

On POSIX, unlike on Windows, these are unexpected errors which cannot
        be handled by L{FilesystemLock}.
        """
        self._symlinkErrorTest(errno.EACCES)
        self._symlinkErrorTest(errno.EIO)

def test_cleanlyAcquire(self):
        """
        If the lock has never been held, it can be acquired and the C{clean}
        and C{locked} attributes are set to C{True}.
        """
        lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        self.assertTrue(lock.lock())
        self.assertTrue(lock.clean)
        self.assertTrue(lock.locked)

def test_cleanlyRelease(self):
        """
        If a lock is released cleanly, it can be re-acquired and the C{clean}
        and C{locked} attributes are set to C{True}.
        """
        lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        self.assertTrue(lock.lock())
        lock.unlock()
        self.assertFalse(lock.locked)

lock = lockfile.FilesystemLock(lockf)
        self.assertTrue(lock.lock())
        self.assertTrue(lock.clean)
        self.assertTrue(lock.locked)

def test_cannotLockLocked(self):
        """
        If a lock is currently locked, it cannot be locked again.
        """
        lockf = self.mktemp()
        firstLock = lockfile.FilesystemLock(lockf)
        self.assertTrue(firstLock.lock())

secondLock = lockfile.FilesystemLock(lockf)
        self.assertFalse(secondLock.lock())
        self.assertFalse(secondLock.locked)

def test_uncleanlyAcquire(self):
        """
        If a lock was held by a process which no longer exists, it can be
        acquired, the C{clean} attribute is set to C{False}, and the
        C{locked} attribute is set to C{True}.
        """
        owner = 12345

def fakeKill(pid, signal):
            if signal != 0:
                raise OSError(errno.EPERM, None)
            if pid == owner:
                raise OSError(errno.ESRCH, None)

lockf = self.mktemp()
        self.patch(lockfile, "kill", fakeKill)
        lockfile.symlink(str(owner), lockf)

lock = lockfile.FilesystemLock(lockf)
        self.assertTrue(lock.lock())
        self.assertFalse(lock.clean)
        self.assertTrue(lock.locked)

self.assertEqual(lockfile.readlink(lockf), str(os.getpid()))

def test_lockReleasedBeforeCheck(self):
        """
        If the lock is initially held but then released before it can be
        examined to determine if the process which held it still exists, it is
        acquired and the C{clean} and C{locked} attributes are set to C{True}.
        """

def fakeReadlink(name):
            # Pretend to be another process releasing the lock.
            lockfile.rmlink(lockf)
            # Fall back to the real implementation of readlink.
            readlinkPatch.restore()
            return lockfile.readlink(name)

readlinkPatch = self.patch(lockfile, "readlink", fakeReadlink)

def fakeKill(pid, signal):
            if signal != 0:
                raise OSError(errno.EPERM, None)
            if pid == 43125:
                raise OSError(errno.ESRCH, None)

self.patch(lockfile, "kill", fakeKill)

lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        lockfile.symlink(str(43125), lockf)
        self.assertTrue(lock.lock())
        self.assertTrue(lock.clean)
        self.assertTrue(lock.locked)

@skipUnless(
        platform.isWindows(),
        "special rename EIO handling only necessary and correct on " "Windows.",
    )
    def test_lockReleasedDuringAcquireSymlink(self):
        """
        If the lock is released while an attempt is made to acquire
        it, the lock attempt fails and C{FilesystemLock.lock} returns
        C{False}.  This can happen on Windows when L{lockfile.symlink}
        fails with L{IOError} of C{EIO} because another process is in
        the middle of a call to L{os.rmdir} (implemented in terms of
        RemoveDirectory) which is not atomic.
        """

def fakeSymlink(src, dst):
            # While another process id doing os.rmdir which the Windows
            # implementation of rmlink does, a rename call will fail with EIO.
            raise OSError(errno.EIO, None)

self.patch(lockfile, "symlink", fakeSymlink)

lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        self.assertFalse(lock.lock())
        self.assertFalse(lock.locked)

@skipUnless(
        platform.isWindows(),
        "special readlink EACCES handling only necessary and " "correct on Windows.",
    )
    def test_lockReleasedDuringAcquireReadlink(self):
        """
        If the lock is initially held but is released while an attempt
        is made to acquire it, the lock attempt fails and
        L{FilesystemLock.lock} returns C{False}.
        """

def fakeReadlink(name):
            # While another process is doing os.rmdir which the
            # Windows implementation of rmlink does, a readlink call
            # will fail with EACCES.
            raise OSError(errno.EACCES, None)

self.patch(lockfile, "readlink", fakeReadlink)

lockf = self.mktemp()
        lock = lockfile.FilesystemLock(lockf)
        lockfile.symlink(str(43125), lockf)
        self.assertFalse(lock.lock())
        self.assertFalse(lock.locked)

def _readlinkErrorTest(self, exceptionType, errno):
        def fakeReadlink(name):
            raise exceptionType(errno, None)

self.patch(lockfile, "readlink", fakeReadlink)

lockf = self.mktemp()

# Make it appear locked so it has to use readlink
        lockfile.symlink(str(43125), lockf)

lock = lockfile.FilesystemLock(lockf)
        exc = self.assertRaises(exceptionType, lock.lock)
        self.assertEqual(exc.errno, errno)
        self.assertFalse(lock.locked)

def test_readlinkError(self):
        """
        An exception raised by C{readlink} other than C{ENOENT} is passed up to
        the caller of L{FilesystemLock.lock}.
        """
        self._readlinkErrorTest(OSError, errno.ENOSYS)
        self._readlinkErrorTest(IOError, errno.ENOSYS)

@skipIf(
        platform.isWindows(),
        "POSIX-specific error propagation not expected on Windows.",
    )
    def test_readlinkErrorPOSIX(self):
        """
        Any L{IOError} raised by C{readlink} on a POSIX platform passed to the
        caller of L{FilesystemLock.lock}.

On POSIX, unlike on Windows, these are unexpected errors which cannot
        be handled by L{FilesystemLock}.
        """
        self._readlinkErrorTest(IOError, errno.ENOSYS)
        self._readlinkErrorTest(IOError, errno.EACCES)

def test_lockCleanedUpConcurrently(self):
        """
        If a second process cleans up the lock after a first one checks the
        lock and finds that no process is holding it, the first process does
        not fail when it tries to clean up the lock.
        """

def fakeRmlink(name):
            rmlinkPatch.restore()
            # Pretend to be another process cleaning up the lock.
            lockfile.rmlink(lockf)
            # Fall back to the real implementation of rmlink.
            return lockfile.rmlink(name)

rmlinkPatch = self.patch(lockfile, "rmlink", fakeRmlink)

def fakeKill(pid, signal):
            if signal != 0:
                raise OSError(errno.EPERM, None)
            if pid == 43125:
                raise OSError(errno.ESRCH, None)

self.patch(lockfile, "kill", fakeKill)

def test_rmlinkError(self):
        """
        An exception raised by L{rmlink} other than C{ENOENT} is passed up
        to the caller of L{FilesystemLock.lock}.
        """

def fakeRmlink(name):
            raise OSError(errno.ENOSYS, None)

self.patch(lockfile, "rmlink", fakeRmlink)

def fakeKill(pid, signal):
            if signal != 0:
                raise OSError(errno.EPERM, None)
            if pid == 43125:
                raise OSError(errno.ESRCH, None)

self.patch(lockfile, "kill", fakeKill)

lockf = self.mktemp()

# Make it appear locked so it has to use readlink
        lockfile.symlink(str(43125), lockf)

lock = lockfile.FilesystemLock(lockf)
        exc = self.assertRaises(OSError, lock.lock)
        self.assertEqual(exc.errno, errno.ENOSYS)
        self.assertFalse(lock.locked)

def test_killError(self):
        """
        If L{kill} raises an exception other than L{OSError} with errno set to
        C{ESRCH}, the exception is passed up to the caller of
        L{FilesystemLock.lock}.
        """

def fakeKill(pid, signal):
            raise OSError(errno.EPERM, None)

self.patch(lockfile, "kill", fakeKill)

lockf = self.mktemp()

# Make it appear locked so it has to use readlink
        lockfile.symlink(str(43125), lockf)

lock = lockfile.FilesystemLock(lockf)
        exc = self.assertRaises(OSError, lock.lock)
        self.assertEqual(exc.errno, errno.EPERM)
        self.assertFalse(lock.locked)

def test_unlockOther(self):
        """
        L{FilesystemLock.unlock} raises L{ValueError} if called for a lock
        which is held by a different process.
        """
        lockf = self.mktemp()
        lockfile.symlink(str(os.getpid() + 1), lockf)
        lock = lockfile.FilesystemLock(lockf)
        self.assertRaises(ValueError, lock.unlock)

def test_isLocked(self):
        """
        L{isLocked} returns C{True} if the named lock is currently locked,
        C{False} otherwise.
        """
        lockf = self.mktemp()
        self.assertFalse(lockfile.isLocked(lockf))
        lock = lockfile.FilesystemLock(lockf)
        self.assertTrue(lock.lock())
        self.assertTrue(lockfile.isLocked(lockf))
        lock.unlock()
        self.assertFalse(lockfile.isLocked(lockf))

${this.title}

Code Editor : test_lockfile.py