python - 如何使用pytest模拟完全重启

Question

如何测试我的程序对意外关闭的鲁棒性？
我的 Python 代码将在意外关闭的微 Controller 上运行。我想测试意外重启的代码的每个部分，并验证它是否正确处理了这个问题。
试图:
我尝试将代码放入它自己的进程中，然后提前终止它，但这不起作用，因为 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()

是否有捷径可寻？如果不是，您如何设计可以随时终止的健壮代码(例如测试状态机)？
类似问题(截至 26/10/21 未答复): 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")