为什么twisted.internet.inotify.INotify()事件在基于twisted.trial的测试中发生在测试之外？

Question

我刚开始使用

twisted.trial

创建测试。我试图了解

twisted.internet.inotify.INotify

类如何与我的

twisted.trial

单元测试代码交互。

测试代码：

import sys
import os
from twisted.trial import unittest
from twisted.internet import inotify
from twisted.python import filepath

STORAGE_PATH = '/tmp/_fake_worker'
CONVERSION_FNAME = "cloud_stream_conv.txt"
TEST_PATH = os.path.join(STORAGE_PATH, CONVERSION_FNAME)


class FileWatcher(object):

    def __init__(self, test_path=TEST_PATH):
        self.test_path = test_path
        self.notifier = inotify.INotify()

    def start(self):
        self.log("Starting FileWatcher")
        self.notifier.startReading()
        self.notifier.watch(
            filepath.FilePath(self.test_path),
            callbacks=[self.notify]
        )

    def stop(self):
        self.log("Stopping FileWatcher")
        self.notifier.loseConnection()

    def log(self, format_spec, *args):
        sys.stderr.write(format_spec % args)
        sys.stderr.write("\n")

    def notify(self, ignored, filepath, mask):
        try:
            self.log(
                "event %s on %s",
                ', '.join(inotify.humanReadableMask(mask)),
                filepath
            )
        except Exception:
            self.log("\nUncaught exception: %s\n", sys.exc_info())


class TestFileWatcher(unittest.TestCase):

    def setUp(self):
        os.makedirs(STORAGE_PATH, exist_ok=True)
        self.test_file = filepath.FilePath(TEST_PATH)
        self.test_file.touch()
        self.fw = FileWatcher()
        self.log_entries = []

    def tearDown(self):
        print("Started tearDown for test %s\n" % self.id)
        self.test_file.remove()
        print("Finished tearDown for test %s\n" % self.id)

    def fake_log(self, format_spec, *args):
        self.log_entries.append(
            format_spec % args
        )

    def test_activity(self):
        self.patch(self.fw, "log", self.fake_log)
        self.fw.start()
        self.assertTrue(self.log_entries)
        print(
            "\nlog_entries (%d) = %s\n" % (
                len(self.log_entries),
                self.log_entries,
            )
        )

当我运行测试时，我看到这个：

evertz@4979b35e5956:~/src$ venv/bin/python -m twisted.trial -e tests.component.test_simple_inotify
tests.component.test_simple_inotify
  TestFileWatcher
    test_activity ... 
log_entries (1) = ['Starting FileWatcher']

Started tearDown for test <bound method TestCase.id of <tests.component.test_simple_inotify.TestFileWatcher testMethod=test_activity>>

Finished tearDown for test <bound method TestCase.id of <tests.component.test_simple_inotify.TestFileWatcher testMethod=test_activity>>

event attrib on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')
event delete_self on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')
                                                     [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.015s

PASSED (successes=1)

我也尝试过这个变体进行测试：

    def test_activity(self):
        self.patch(self.fw, "log", self.fake_log)

        def delete_file(result):
            self.test_file.remove()

        def report_activity(result):
            self.assertTrue(self.log_entries)
            print(
                "\nlog_entries (%d) = %s\n" % (
                    len(self.log_entries),
                    self.log_entries,
                )
            )

        self.fw.start()
        d = defer.Deferred()
        d.addCallback(delete_file)
        d.addCallback(report_activity)
        reactor.callLater(2, d.callback, None)
        return d

但仍然得到类似的东西：

evertz@4979b35e5956:~/src$ venv/bin/python -m twisted.trial -e tests.component.test_simple_inotify
tests.component.test_simple_inotify
  TestFileWatcher
    test_activity ... 
log_entries (1) = ['Starting FileWatcher']

Started tearDown for test <bound method TestCase.id of <tests.component.test_simple_inotify.TestFileWatcher testMethod=test_activity>>

Finished tearDown for test <bound method TestCase.id of <tests.component.test_simple_inotify.TestFileWatcher testMethod=test_activity>>

event attrib on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')
event delete_self on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')
                                                     [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 2.006s

PASSED (successes=1)

我预计以

event [...]

开头的消息将位于我的测试代码中的

self.log_entries

变量内，并且事件实际上会在测试之前/期间发生，而不是在

tearDown()

之后发生。如何编写一个测试，使

INotify()

实例触发其回调，以便我可以正确测试它？

基于 Jean-Paul Calderone 的建议的建议的解决方案：

import sys
import os
import time
from twisted.trial import unittest
from twisted.internet import inotify
from twisted.internet import reactor
from twisted.internet import defer
from twisted.python import filepath

STORAGE_PATH = '/tmp/_fake_worker'
CONVERSION_FNAME = "cloud_stream_conv.txt"
TEST_PATH = os.path.join(STORAGE_PATH, CONVERSION_FNAME)


class FileWatcher(object):

    def __init__(self, parent, test_path=TEST_PATH):
        self.parent = parent
        self.test_path = test_path
        self.notifier = inotify.INotify()
        self.doneCalled = False

    def start(self):
        self.log("Starting FileWatcher")
        self.notifier.startReading()
        self.notifier.watch(
            filepath.FilePath(self.test_path),
            callbacks=[self.notify]
        )

    def stop(self):
        self.log("Stopping FileWatcher")
        self.notifier.loseConnection()

    def log(self, format_spec, *args):
        sys.stderr.write(format_spec % args)
        sys.stderr.write("\n")

    def notify(self, ignored, filepath, mask):
        try:
            self.log(
                "event %s on %s",
                ', '.join(inotify.humanReadableMask(mask)),
                filepath
            )
            if not self.doneCalled:
                reactor.callLater(2, self.parent.done.callback, None)
                self.doneCalled = True
        except Exception:
            self.log("\nUncaught exception: %s\n", sys.exc_info())

class TestFileWatcher(unittest.TestCase):

    def setUp(self):
        os.makedirs(STORAGE_PATH, exist_ok=True)
        self.test_file = filepath.FilePath(TEST_PATH)
        self.test_file.touch()
        self.fw = FileWatcher(self)
        self.log_entries = []

    def tearDown(self):
        try:
            self.test_file.remove()
        except FileNotFoundError:
            pass

    def fake_log(self, format_spec, *args):
        self.log_entries.append(
            format_spec % args
        )

    def test_activity(self):
        self.patch(self.fw, "log", self.fake_log)
        self.done = defer.Deferred()
        self.fw.start()

        self.test_file.remove()

        def report_activity(result):
            self.assertTrue(self.log_entries)
            print(
                "\nlog_entries (%d) = %s\n" % (
                    len(self.log_entries),
                    self.log_entries,
                )
            )

        self.done.addCallback(report_activity)
        return self.done

当我运行这个时，我得到了我想要的预期结果：

evertz@f24cc5aacff8:~/src$ venv/bin/python -m twisted.trial -e tests.component.test_simple_inotify.TestFileWatcher.test_activity
tests.component.test_simple_inotify
  TestFileWatcher
    test_activity ... 
log_entries (3) = ['Starting FileWatcher', "event attrib on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')", "event delete_self on FilePath(b'/tmp/_fake_worker/cloud_stream_conv.txt')"]

                                                     [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 2.005s

PASSED (successes=1)

我仍然在这里使用可变状态变量，但我将其放在

inotify

活动所在的位置，并使用

reactor.callLater()

为其提供额外的时间，以便捕获所有状态更改。这不是最优雅的解决方案，但这需要经验。

Answer 1

首先，简单介绍一下试用。

twisted.trial.unittest.TestCase

支持“异步”测试。也就是说，返回 Deferred 的测试（或“协程”，如果您有足够新的 Twisted 版本）。如果测试方法返回这些类型之一的值，则该测试将继续运行，直到异步结果可用，然后异步结果将被解释为同步结果。当这种情况发生时，全球反应堆正在运行。

此外，在任何

twisted.trial.unittest.TestCase

测试方法完成后，试验都会让全局反应器运行很短的一段时间，然后才考虑测试完成。

最后，回想一下 Twisted 是一个协作式多任务系统。这意味着在通常情况下，在任何给定时间只有程序的一部分正在运行，并且只有当您允许这种情况发生时，正在运行的部分才会发生变化。

现在，关于 Twisted 对 inotify 的支持。

twisted.internet.inotify.INotify

使用反应器从 Linux inotify 系统接收信息。它需要一个正在运行的反应堆，否则无法正常运行。

您的第一个测试方法存在问题，即它不允许反应器运行直到完成。您观察到了这样做的结果 - setUp 运行，然后是您的测试方法，然后tearDown，然后试运行反应器一点，这会导致您的 inotify 代码运行 - 似乎在测试结束后。

你的第二种测试方法也有类似的问题。它确实允许反应堆运行（通过返回但通过使其返回值成为

Deferred

来表明它尚未完成）。但是，它不会引发任何 inotify 活动 (

delete_file

)，直到它“立即”发出信号表示已完成：

它使用
```
callLater
```
延迟
```
d
```
的发射 2 秒
2 秒后，
```
delete_file
```
运行。
紧接着，反应堆没有机会运行，
```
report_activity
```
运行。
紧接着，由于反应堆没有机会运行，Trial 看到
```
d
```
有结果，这导致 Trial 认为测试已完成。
再试运行一下反应堆，inotify 的事情就会发生。

因此，要解决您的问题，您需要确保 inotify 片段可以在您向 Trial 发出测试完成信号之前发生。

这可能看起来像：

    ...

    def fake_log(self, format_spec, *args):
        self.log_entries.append(
            format_spec % args
        )
        self.done.callback(None)

    ...

    def test_activity(self):
        self.patch(self.fw, "log", self.fake_log)
        self.done = defer.Deferred()
        self.fw.start()

        self.test_file.remove()

        def report_activity(result):
            self.assertTrue(self.log_entries)
            print(
                "\nlog_entries (%d) = %s\n" % (
                    len(self.log_entries),
                    self.log_entries,
                )
            )

        self.done.addCallback(report_activity)
        return self.done

请注意，您要观察的 inotify 活动现在如何控制表示测试完成的 Deferred 活动。

另请注意，我不

喜欢我在这里如何管理self.done

：这是一个远距离可变状态和操作的示例，这使得代码难以理解和维护；它还可能会引入多次调用 inotify watch 回调的问题（您只能回调 Deferred 一次）。我这样做是因为它最容易接近您的示例代码，所以我认为它比重新安排所有内容更好地回答您的具体问题。

为什么twisted.internet.inotify.INotify()事件在基于twisted.trial的测试中发生在测试之外？

问题描述投票：0回答：1

1个回答

最新问题

为什么twisted.internet.inotify.INotify()事件在基于twisted.trial的测试中发生在测试之外？

问题描述 投票：0回答：1

1个回答

最新问题

问题描述投票：0回答：1