如何在功能测试中模拟 INotify 失败？

Posted 2023-02-17

【中文标题】如何在功能测试中模拟 INotify 失败？

【英文标题】：How to simulate INotify failure in functional test?

【发布时间】：2018-03-27 03:23:33

【问题描述】：

我有一个使用 inotify 跟踪文件系统更改的 Linux 应用程序。我想为它编写一个功能测试套件，从最终用户的角度测试应用程序，作为其中的一部分，我想测试文件系统失败的情况，特别是我想测试 inotify 失败。 具体来说，我想让 inotify_init()、inotify_add_watch()、inotify_rm_watch() 调用和对 inotify 文件描述符的调用 read() 在测试中需要它时返回错误。

但问题是我找不到模拟inotify失败的方法。我想知道是否有人已经遇到过这样的问题并且知道一些解决方案。

【问题讨论】：

你在这里使用什么单元测试框架？

@TarunLalwani 我的应用程序是用 Python 编写的，但我没有为功能测试选择框架。功能测试套件可以用任何语言和任何框架编写，因为它不直接使用应用程序的任何内部组件。

你使用什么模块来与 Python 中的 inotify 交互？

@cryptoplex inotify_simple: pypi.python.org/pypi/inotify_simple

【参考方案1】：

如果您想避免任何嘲笑，最好的办法就是直接达到操作系统限制来引发错误。例如，inotify_init 可能会失败并显示EMFILE errno，如果调用进程已达到其打开文件描述符数量的限制。要以 100% 的精度达到此类条件，您可以使用两个技巧：

通过changing values in procfs动态操作运行进程的限制 将您的应用进程分配给专用 cgroup 并通过 cgroups API 为其提供约 0% 的 CPU 时间来“暂停”它（这是 android 限制后台应用程序并实现其节能“打盹”模式的方式）。

所有可能的 inotify 错误情况都记录在 inotify、inotify_init 和 inotify_add_watch 的手册页中（我认为 inotify_rm_watch 不会失败，除非您的代码中出现纯粹的编程错误）。

除了普通的错误（例如超过/proc/sys/fs/inotify/max_user_watches）之外，inotify 有几种故障模式（队列空间耗尽、手表 ID 重用），但这些都不是严格意义上的“故障”。

当有人执行文件系统更改的速度快于您的反应速度时，就会发生队列耗尽。很容易重现：使用 cgroups 在程序打开 inotify 描述符时暂停程序（因此事件队列不会耗尽），并通过修改观察到的文件/目录快速生成 lots 通知。一旦您拥有/proc/sys/fs/inotify/max_queued_events 的未处理事件，并取消暂停您的程序，它将收到IN_Q_OVERFLOW（并可能错过一些不适合队列的事件）。

Watch ID 重用很难重现，因为现代内核从类似于文件描述符的行为切换到类似于 PID 的 watch-ID 行为。您应该使用与测试 PID 重用时相同的方法——创建和销毁 lots 个 inotify 监视，直到整数监视 ID 环绕。

Inotify 也有一些棘手的极端情况，在正常操作期间很少发生（例如，我知道的所有 Java 绑定，包括 Android 和 OpenJDK，都不能正确处理所有这些）：same-inode 问题和处理IN_UNMOUNT.

inotify 文档中很好地解释了同一个 inode 问题：

对 inotify_add_watch() 的成功调用会返回此 inotify 实例的唯一监视描述符，用于对应于路径名的文件系统对象（inode）。如果文件系统对象以前没有被这个 inotify 实例监视，那么监视描述符是新分配的。如果文件系统对象已被监视（可能通过指向同一对象的不同链接），则返回现有监视的描述符。

简单来说：如果您观看两个指向同一个文件的硬链接，它们的数字观看 ID 将是相同的。如果您将 watch 存储在 hashmap 之类的东西中，这种行为很容易导致丢失对第二个 inotify watch 的跟踪，并以整数 watch ID 为键。

第二个问题更难观察，因此即使不是错误模式也很少得到适当的支持：卸载分区，目前通过 inotify 观察到。棘手的部分是：Linux 文件系统不允许您在文件描述符打开它们时自行卸载，但通过 inotify 观察文件不会阻止文件系统卸载。如果您的应用在单独的文件系统上观察文件，并且用户卸载了该文件系统，则您必须准备好处理由此产生的 IN_UNMOUNT 事件。

以上所有测试都应该可以在 tmpfs 文件系统上执行。

【讨论】：

“你想测试什么“失败”？”我想拨打inotify_init()、inotify_add_watch()、inotify_rm_watch()和@987654337电话@ 用于 inotify 文件描述符在测试需要时返回错误。

@GillBates 如果您想避免任何嘲笑，这种方法是您唯一的选择。但是，如果您可以接受一些低级“模拟”（在不修改程序代码的情况下安装操作系统级系统调用过滤器），也可以考虑my other answer。

【参考方案2】：

经过一番思考，我想出了另一个解决方案。您可以使用 Linux 的“seccomp”工具来“模拟”各个与 inotify 相关的系统调用的结果。这种方法的优点是简单、健壮且完全非侵入性。您可以有条件地调整系统调用的行为，同时在其他情况下仍使用原始操作系统行为。从技术上讲，这仍然算作模拟，但模拟层放置得很深，位于内核代码和用户空间系统调用接口之间。

您不需要修改程序代码，只需编写一个包装器，在exec-ing 您的应用程序之前安装一个合适的 seccomp 过滤器（下面的代码使用libseccomp）：

 // pass control to kernel syscall code by default
 scmp_filter_ctx ctx = seccomp_init(SCMP_ACT_ALLOW);
 if (!ctx) exit(1);

 // modify behavior of specific system call to return `EMFILE` error
 seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EMFILE), __NR_inotify_init, 0));

 execve(...

Seccomp 本质上是一个有限的解释器，运行 BPF 字节码的扩展版本，因此它的功能非常广泛。 libseccomp 允许您安装有限的条件过滤器（例如将系统调用的整数参数与常量值进行比较）。如果您想实现更令人印象深刻的条件行为（例如将传递给inotify_add_watch的文件路径与预定义值进行比较），您可以将 seccomp() 系统调用的直接使用与kernel bpf() facility 结合起来，用eBPF方言编写复杂的过滤程序。

编写系统调用过滤器可能很乏味，而且程序在 seccomp 影响下的行为实际上并不依赖于内核实现（seccomp 过滤器在将控制权传递给内核系统调用处理程序之前由内核调用）。因此，您可能希望将 seccomp 的稀疏使用与my other answer 中概述的更有机的方法结合起来。

【参考方案3】：

可能不像您希望的那样非侵入性，但来自inotify_simple 的INotify 类很小。你可以完全包装它，委托所有方法，并注入错误。

代码如下所示：

from inotify_simple.inotify_simple import INotify

class WrapINotify(object):

    init_error_list      = []
    add_watch_error_list = []
    rm_watch_error_list  = []
    read_error_list      = []

    def raise_if_error(self, error_list):

        if not error_list:
            return

        # Simulate INotify raising an exception
        exception = error_list.pop(0)

        raise exception

    def __init__(self):

        self.raise_if_error(WrapINotify.init_error_list)
        self.inotify = INotify()

    def add_watch(self, path, mask):

        self.raise_if_error(WrapINotify.add_watch_error_list)
        self.inotify.add_watch(path, mask)

    def rm_watch(self, wd):

        self.raise_if_error(WrapINotify.rm_watch_error_list)
        return self.inotify.rm_watch(wd)

    def read(self, timeout=None, read_delay=None):

        self.raise_if_error(WrapINotify.read_error_list)
        return self.inotify.read(timeout, read_delay)

    def close(self):

        self.inotify.close()

    def __enter__(self):

        return self.inotify.__enter__()

    def __exit__(self, exc_type, exc_value, traceback):

        self.inotify.__exit__(exc_type, exc_value, traceback)

使用此代码，您可以在其他地方执行以下操作：

WrapINotify.add_watch_error_list.append(OSError(28, 'No space left on disk'))

注入错误。当然，您可以在包装类中添加更多代码来实现不同的错误注入方案。