test_phi_accrual_failure_detector.py

import time
from typing import Iterable

import pytest

from phi_accrual_failure_detector import PhiAccrualFailureDetector


class TestPhiAccrualFailureDetector:

    @classmethod
    def get_time_mocked(cls, times: Iterable[float]):
        def mocked_func():
            current_time = next(times)
            print(f"Current time: {current_time}")
            return current_time

        return mocked_func
      
    def test_failure_if_heartbeat_missed(self):
        def mock_time() -> iter:
            yield 0
            yield 1000
            yield 1100
            yield 1200
            yield 5200
            yield 8200

        failure_detector = PhiAccrualFailureDetector(
            threshold=3,
            max_sample_size=1000,
            min_std_deviation_ms=10,
            acceptable_heartbeat_pause_ms=0,
            first_heartbeat_estimate_ms=1000
        )

        failure_detector._get_time = self.get_time_mocked(mock_time())

        failure_detector.heartbeat() # 0
        failure_detector.heartbeat() # 1000
        failure_detector.heartbeat() # 1100

        assert failure_detector.is_available() is True # 1200
        failure_detector._get_time() # 5200 (missed)
        assert failure_detector.is_available() is False # 8200