How to use pytest to simulate full reboot

Question

How do I test that my program is robust to unexpected shut-downs?

My python code will run on a microcontroller that shuts off unexpectedly. I would like to test each part of the code rebooting unexpectedly and verify that it handles this correctly.

Attempt: I tried putting code into its own process, then terminating it early, but this doesn't work because MyClass calls 7zip from the command line which continues even after process dies:

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")

What I want:

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

Is there an easy way to do this? If not, how do you design robust code that could terminate at any point (e.g. testing state machines)?

Similar question (unanswered as of 26/10/21): Simulating abnormal termination in pytest

Thanks a lot!

Answer 1

Your logic starts a process wrapped within the MyClass object which itself spawns a new process via the os.system call.

When you terminate the MyClass process, you kill the parent process but you leave the 7zip process running as orphan.

Moreover, the process.terminate method sends a SIGTERM signal to the child process. The child process can intercept said signal and perform some cleanup routines before terminating. This is not ideal if you want to simulate a situation where there is no chance to clean up (a power loss). You most likely want to send a SIGKILL signal instead (on Linux).

To kill the parent and child process, you need to address the entire process group.

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")

The above works only on Unix OSes and not on Windows ones.

For Windows, you need to use the psutil module.

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")