如何在Python中测试文件锁定

Question

所以我想写一些可能被其他进程锁定/阻止写入/删除的文件，并喜欢提前测试。

据我所知：os.access(path, os.W_OK)只查找权限，尽管当前无法写入文件，但会返回true。所以我有一个小功能：

def write_test(path):
    try:
        fobj = open(path, 'a')
        fobj.close()
        return True
    except IOError:
    return False

当我用程序手动打开一个文件时，它实际上运行得很好。但作为一个想成为优秀开发人员的人，我想把它放在一个测试中，自动看看它是否按预期工作。

问题是：如果我只是open(path, 'a')文件，我仍然可以open()它再次没有问题！甚至来自另一个Python实例。尽管Explorer实际上会告诉我该文件当前已在Python中打开！

我在这里查找了其他帖子&关于锁定。大多数人建议安装软件包。你可能会明白，我不想这样做来测试几行代码。所以我挖出包裹，看看最终锁定的实际位置。。。

fcntl？我没有。win32con？也不要。。。现在在文件锁中有这样的：

self.fd = os.open(self.lockfile, os.O_CREAT|os.O_EXCL|os.O_RDWR)

当我对文件执行此操作时，它会抱怨文件存在！！嗯。。。是的！这就是主意！但即使我在一条不存在的道路上做这件事。我仍然可以open(path, 'a')它！甚至从另一个python实例。。。

我开始觉得我没有理解这里一些非常基本的东西。我找错东西了吗？有人能给我指正确的方向吗？谢谢

Answer 1

您正试图仅使用系统调用open（）来实现文件锁定问题。类Unix系统默认使用咨询文件锁定。这意味着协作进程可以使用锁来协调它们之间对文件的访问，但不协作进程也可以自由忽略锁，并以他们选择的任何方式访问文件。换句话说，文件锁只锁定其他文件锁，而不是I/O。请参阅维基百科。

正如系统调用open（）引用中所述，使用锁定文件执行原子文件锁定的解决方案是在同一文件系统上创建一个唯一的文件（例如，合并主机名和pid），使用link（2）创建到锁定文件的链接。如果link（）返回0，则锁定成功。否则，请对唯一文件使用stat（2）来检查其链接数是否已增加到2，在这种情况下锁定也成功。

这就是为什么在filelock中，他们还使用函数fcntl.flock（），并将所有这些东西放在模块中。

Answer 2

好吧！多亏了那些家伙，我现在真的有东西了！这就是我的功能：

def lock_test(path):
    """
    Checks if a file can, aside from it's permissions, be changed right now (True)
    or is already locked by another process (False).
    :param str path: file to be checked
    :rtype: bool
    """
    import msvcrt
    try:
        fd = os.open(path, os.O_APPEND | os.O_EXCL | os.O_RDWR)
    except OSError:
        return False
    try:
        msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
        msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
        os.close(fd)
        return True
    except (OSError, IOError):
        os.close(fd)
        return False

单元测试可能看起来像这样：

class Test(unittest.TestCase):
    def test_lock_test(self):
        testfile = 'some_test_name4142351345.xyz'
        testcontent = 'some random blaaa'
        with open(testfile, 'w') as fob:
            fob.write(testcontent)
        # test successful locking and unlocking
        self.assertTrue(lock_test(testfile))
        os.remove(testfile)
        self.assertFalse(os.path.exists(testfile))
        # make file again, lock and test False locking
        with open(testfile, 'w') as fob:
            fob.write(testcontent)
        fd = os.open(testfile, os.O_APPEND | os.O_RDWR)
        msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
        self.assertFalse(lock_test(testfile))
        msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
        self.assertTrue(lock_test(testfile))
        os.close(fd)
        with open(testfile) as fob:
            content = fob.read()
            self.assertTrue(content == testcontent)
        os.remove(testfile)

工作。缺点是：

• 这是在自我测试
• 因此，最初的OSError捕获甚至没有测试，只是用msvcrt再次锁定

但我不知道现在该如何做得更好。