Вот этот код:
import asyncio
import time
import statistics


class EventLoopMonitor:
    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.blocking_durations = []
        self.threshold = 0.001
        self.last_call_time = time.perf_counter()
        self._is_running = False

    def _callback(self):
        current_time = time.perf_counter()
        delta = current_time - self.last_call_time
        if delta > self.threshold:
            self.blocking_durations.append(delta)
        self.last_call_time = current_time
        if self._is_running:
            self.loop.call_soon(self._callback)

    async def monitor_callback(self) -> None:
        self._is_running = True
        self._callback()
        try:
            while True:
                await asyncio.sleep(3600)
        except asyncio.CancelledError:
            self._is_running = False
            pass

    def get_statistics(self) -> dict:
        if not self.blocking_durations:
            return {
                "average": 0.0,
                "median": 0.0,
                "min": 0.0,
                "max": 0.0,
                "count": 0
            }

        average = sum(self.blocking_durations) / len(self.blocking_durations)
        median = statistics.median(self.blocking_durations)
        maximum = max(self.blocking_durations)
        minimum = min(self.blocking_durations)
        count = len(self.blocking_durations)

        return {
            "average": average,
            "median": median,
            "min": minimum,
            "max": maximum,
            "count": count
        }

Должен решать задачу:

Предпосылки:

В асинхронном программировании блокировка цикла событий может серьезно повлиять на производительность приложения. Данная задача исследует влияние смешивания синхронных блокирующих операций (time.sleep) с асинхронным кодом (asyncio.sleep) и измеряет результирующее время блокировки цикла событий.

Описание Задачи

В тестах реализован сервис, который генерирует блокирующие и неблокирующие операции и непрерывно нагружает ими цикл событий (event loop).

Ваша задача:

Создать механизм для измерения времени блокировки цикла событий (блокировкой считается задержка более 0.001 секунды)
Вычислять и отображать среднюю продолжительность блокировок, медиану и максимальную длительность блокировки
Рассчитывать количество блокировок
Формат вывода:

{
    "average": 0.001,
    "median": 0.001,
    "max": 0.001,
    "num_blocking": 100
}
Формат класса:

import asyncio

class EventLoopMonitor:
    def __init__(self):
        pass
        
    async def monitor_callback(self) -> None:
        """
        Callback to measure time between event loop iterations.
        Runs asynchronously as a daemon in the event loop.
        """
        pass
            
    def get_statistics(self) -> dict:
        """
        Calculate statistics about blocking times.
        TODO: Replace with actual implementation
        """
            
        return {
            "count": 0,
            "average": 0,
            "max": 0,
            "min": 0,
            "median": 0
        }

Но у него ошибка в тесте:

=================================== FAILURES ===================================
___________ test_no_blocking_detection[0.1-10-5-1e-10-1e-07-0.5-0.1] ___________
[gw0] linux -- Python 3.12.3 /home/muzaffarsad/myenv/bin/python

blocking_chance = 0.1, num_tasks = 10, duration = 5, max_blocking_time = 1e-10
min_blocking_time = 1e-07, max_non_blocking_time = 0.5
min_non_blocking_time = 0.1

    @pytest.mark.parametrize(
        "blocking_chance,num_tasks,duration,max_blocking_time,min_blocking_time,max_non_blocking_time,min_non_blocking_time",
        [
            (0.1, 10, 5, 1e-10, 1e-7, 1e-10, 1e-5),
            (0.1, 10, 5, 1e-10, 1e-7, 0.5, 0.1)
        ],
    )
    @pytest.mark.asyncio
    async def test_no_blocking_detection(
            blocking_chance: float,
            num_tasks: int,
            duration: int,
            max_blocking_time: float,
            min_blocking_time: float,
            max_non_blocking_time: float,
            min_non_blocking_time: float
    ):
        """Test if multiple blocking operations are detected."""
        stats = await main(duration=duration, num_tasks=num_tasks, blocking_chance=blocking_chance, max_blocking_time=max_blocking_time, min_blocking_time=min_blocking_time, max_non_blocking_time=max_non_blocking_time, min_non_blocking_time=min_non_blocking_time)
>       assert stats["count"] == 0, "There should be no blocking operations"
E       AssertionError: There should be no blocking operations
E       assert 1 == 0

event_loop_monitor/test_private.py:157: AssertionError
=========================== short test summary info ============================
FAILED event_loop_monitor/test_private.py::test_no_blocking_detection[0.1-10-5-1e-10-1e-07-0.5-0.1] - AssertionError: There should be no blocking operations
assert 1 == 0
========================= 1 failed, 4 passed in 23.01s =========================