How to Mock a Liveness Check and Restart a Process in Python

Simulate a failing process and restart it after a liveness check fails, using a mock class and a liveness loop.

Medium Python 3.9+ Aug 9, 2026 Reliability & rate limiting 14 views 0 copies

Python code

48 lines
Python 3.9+
import subprocess
import sys
import time
import os

class ProcessMock:
    def __init__(self, name, fail_after_seconds=3):
        self.name = name
        self.fail_after = fail_after_seconds
        self.start_time = None
        self.is_running = False

    def start(self):
        self.start_time = time.time()
        self.is_running = True
        print(f"[{self.name}] started")

    def is_alive(self):
        if not self.is_running:
            return False
        return time.time() - self.start_time < self.fail_after

    def stop(self):
        self.is_running = False
        print(f"[{self.name}] stopped")

    def restart(self):
        print(f"[{self.name}] restarting...")
        self.stop()
        self.start()

def run_liveness_loop(process, check_interval=1.0):
    attempts = 0
    max_attempts = 3
    while attempts < max_attempts:
        time.sleep(check_interval)
        if process.is_alive():
            print(f"[{process.name}] healthy")
        else:
            print(f"[{process.name}] FAILED liveness check")
            process.restart()
            attempts += 1
    print(f"[{process.name}] giving up after {max_attempts} restart attempts")

if __name__ == "__main__":
    process = ProcessMock(name="worker", fail_after_seconds=2)
    process.start()
    run_liveness_loop(process)

Output

stdout
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] healthy
[worker] healthy
[worker] FAILED liveness check
[worker] restarting...
[worker] stopped
[worker] started
[worker] giving up after 3 restart attempts

How it works

The ProcessMock class simulates a long-running process with a hard-coded lifetime (fail_after_seconds). The is_alive() method checks whether the current time minus the start time is less than the failure threshold, mimicking a health probe. The run_liveness_loop polls with time.sleep, and on failure it restarts the process and increments an attempt counter. After a maximum number of restarts, it gives up, which is a common pattern in production supervisors. The loop uses simple polling rather than threads or asyncio to keep the mock self-contained and easy to test.

Common mistakes

  • Forgetting to sleep before the first liveness check, causing immediate false failures.
  • Not resetting a start time on restart, so the process never appears alive again.
  • Infinite restart loops without a max attempt count, leading to resource exhaustion.
  • Using `is_alive()` with a fixed timestamp instead of tracking start time per start call.

Variations

  1. Use `threading.Event.wait(timeout)` instead of `time.sleep` for a responsive check.
  2. Implement liveness as a context manager to automatically stop/restart in a `with` block.

Real-world use cases

  • Testing orchestrator logic that restarts a background worker after a health check fails.
  • Simulating a crash in a bulk job to validate that retry counters and give-up behavior work end-to-end.
  • Writing unit tests for a container orchestrator's restart policy before deploy to production.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Reliability & rate limiting

Related tutorials and quizzes for this topic.