如何使用pytest模拟完全重启

Question

如何测试我的程序对意外关闭的鲁棒性？

我的 python 代码将运行在意外关闭的微控制器上。我想测试意外重启的代码的每个部分，并验证它是否正确处理了这个问题。

尝试：我尝试将代码放入其自己的进程中，然后提前终止它，但这不起作用，因为 MyClass 从命令行调用 7zip，即使在进程结束后仍继续：

import multiprocessing
import os

def MyClass(multiprocessing.Process):
   ...
   def run():
      os.system("7z a myfile.7z myfile")


process = MyClass()
process.start()
time.sleep(4)
print("terminating early")
process.terminate()
print("done")

我想要的：

class TestMyClass(unittest.TestCase):
    def test_MyClass_continuity(self):
        myclass = MyClass().start()
        myclass.kill_everything()
        myclass = MyClass().start()
        self.assert_everything_worked_as_expected()

有没有简单的方法来做到这一点？如果不是，您如何设计可以在任何时候终止的健壮代码（例如测试状态机）？

类似问题（截至 2021 年 10 月 26 日尚未回答）：Simulating abnormal termination in pytest

非常感谢！

Answer 1

您的逻辑启动了一个包含在 MyClass 对象中的进程，该对象本身通过 os.system 调用生成了一个新进程。

当您终止 MyClass 进程时，您杀死了父进程，但将 7zip 进程运行保留为孤儿。

此外，process.terminate方法向子进程发送了一个SIGTERM信号。子进程可以拦截所述信号并在终止之前执行一些清理例程。如果您想模拟没有机会清理（断电）的情况，这并不理想。您很可能想发送 SIGKILL 信号（在 Linux 上）。

杀死父子进程，需要解决整个进程组。

import os
import time
import signal
import multiprocessing


class MyClass(multiprocessing.Process):
    def run(self):
        # Ping localhost for a limited amount of time
        os.system("ping -c 12 127.0.0.1")


process = MyClass()
process.start()

time.sleep(4)

print("terminating early")

# Send SIGKILL signal to the entire process group
group_id = os.getpgid(process.pid)
os.killpg(group_id, signal.SIGKILL)

print("done")

以上仅适用于 Unix 操作系统，不适用于 Windows 操作系统。

对于Windows，您需要使用psutil模块。

import os
import time
import multiprocessing

import psutil


class MyClass(multiprocessing.Process):
    def run(self):
        # Ping localhost for a limited amount of time
        os.system("ping -c 12 127.0.0.1")


def kill_process_group(pid):
    process = psutil.Process(pid)
    children = process.children(recursive=True)

    # First terminate all children
    for child in children:
        child.kill()
    psutil.wait_procs(children)

    # Then terminate the parent process
    process.kill()
    process.wait()


process = MyClass()
process.start()

time.sleep(4)

print("terminating early")

kill_process_group(process.pid)

print("done")

Answer 2

我认为这是数据持久化和一致性的问题。您需要确保所有持久的数据（即写入磁盘）也是一致的。

想象一下将某种数据写入状态文件。意外终止后应用程序将读取什么？新状态的一半和以前的状态的一半？一半的新状态，其余的都是 0x00?

所以你的问题的答案 “您如何设计可以随时终止的健壮代码？” 是在处理持久数据时使用原子操作。大多数 数据库 都在这方面提供了一些保证。对于处理本地文件，我个人通常使用 重命名文件 。这样我就可以写入一个临时文件而不用担心一致性，只有当它完成时（一定要刷新缓冲区！）因此保持一致我使用重命名的原子操作使临时文件成为新的单点的真理，因此也坚持不懈。如果在过程中的任何一点应用程序意外终止，持久数据将始终保持一致。它将是以前的状态（以及临时文件中的一些垃圾）或新状态，但两者之间没有任何东西。

无论您的选择是什么，请务必阅读有关原子性的文档以了解可能发生的情况。 IE。在正确的时间点中断的文件重命名可能看起来像创建硬 link.

请注意，仅终止进程与切断电源不同，因为 OS 保留运行并关闭文件、刷新缓冲区等。例如，在使用 SQLite 时，我很少看到“ journal files" 当我刚刚终止应用程序时，但我在切断电源时经常看到它们。

如何使用pytest模拟完全重启

How to use pytest to simulate full reboot

python

error-handling

multiprocessing

pytest

microcontroller