Function test_get_cpu_sample

tests/unit/_autoscaling/test_snapshotter.py:159–193

(
    snapshotter: Snapshotter, event_manager: LocalEventManager, default_memory_info: MemoryInfo
)

@pytest.mark.run_alone
async def test_get_cpu_sample(
    snapshotter: Snapshotter, event_manager: LocalEventManager, default_memory_info: MemoryInfo
) -> None:
    now = datetime.now(timezone.utc)
    snapshotter._SNAPSHOT_HISTORY = timedelta(hours=10)  # Extend history for testing

    events_data = [
        EventSystemInfoData(
            cpu_info=CpuInfo(
                used_ratio=0.5,
                created_at=now - timedelta(hours=delta),
            ),
            memory_info=default_memory_info,
        )
        for delta in range(5, 0, -1)
    ]
    for event_data in events_data:
        event_manager.emit(event=Event.SYSTEM_INFO, event_data=event_data)
    await event_manager.wait_for_all_listeners_to_complete()

    # When no sample duration is provided it should return all snapshots
    samples = snapshotter.get_cpu_sample()
    assert len(samples) == len(events_data)

    duration = timedelta(hours=0.5)
    samples = snapshotter.get_cpu_sample(duration)
    assert len(samples) == 1

    duration = timedelta(hours=2.5)
    samples = snapshotter.get_cpu_sample(duration)
    assert len(samples) == 3

    duration = timedelta(hours=10)
    samples = snapshotter.get_cpu_sample(duration)
    assert len(samples) == len(events_data)


async def test_methods_raise_error_when_not_active() -> None: