Repository: apify/crawlee-python
File: tests/unit/_autoscaling/test_snapshotter.py (Python)

Code before:
    now = datetime.now(timezone.utc)
    one_hour_ago = now - timedelta(hours=1)

    # Create mock snapshots with varying creation times
    snapshots = [
        CpuSnapshot(used_ratio=0.7, max_used_ratio=0.95, created_at=one_hour_ago),
        CpuSnapshot(used_ratio=0.8, max_used_ratio=0.95, created_at=now),
    ]

    # Assign these snapshots to one of the lists (e.g., CPU snapshots)
    snapshotter._cpu_snapshots = snapshots

    # Prune snapshots older than 2 hours
    snapshots_casted = cast(list[Snapshot], snapshotter._cpu_snapshots)
    snapshotter._prune_snapshots(snapshots_casted, now)

    # Check that only the last two snapshots remain
    assert len(snapshotter._cpu_snapshots) == 2
    assert snapshotter._cpu_snapshots[0].created_at == one_hour_ago
    assert snapshotter._cpu_snapshots[1].created_at == now


def test_memory_load_evaluation_logs_warning_on_high_usage(caplog: pytest.LogCaptureFixture) -> None:
    snapshotter = Snapshotter(AsyncMock(spec=EventManager), max_memory_size=ByteSize.from_gb(8))

    high_memory_usage = ByteSize.from_gb(8) * 0.95  # 90% of 8 GB

    snapshotter._evaluate_memory_load(
        current_memory_usage_size=high_memory_usage,
        snapshot_timestamp=datetime.now(timezone.utc),
    )

    assert len(caplog.records) == 1
    assert caplog.records[0].levelname.lower() == 'warning'
    assert 'Memory is critically overloaded' in caplog.records[0].msg

    # It should not log again, since the last log was short time ago
    snapshotter._evaluate_memory_load(
        current_memory_usage_size=high_memory_usage,
        snapshot_timestamp=datetime.now(timezone.utc),
    )

    assert len(caplog.records) == 1


def test_memory_load_evaluation_silent_on_acceptable_usage(
    monkeypatch: pytest.MonkeyPatch,
    snapshotter: Snapshotter,
) -> None:
    mock_logger_warn = MagicMock()
    monkeypatch.setattr(getLogger('crawlee.autoscaling.snapshotter'), 'warning', mock_logger_warn)

Review comment (suggestion):
```suggestion
    high_memory_usage = ByteSize.from_gb(8) * 0.95  # 95% of 8 GB
```

Code after:
    now = datetime.now(timezone.utc)
    one_hour_ago = now - timedelta(hours=1)

    # Create mock snapshots with varying creation times
    snapshots = [
        CpuSnapshot(used_ratio=0.7, max_used_ratio=0.95, created_at=one_hour_ago),
        CpuSnapshot(used_ratio=0.8, max_used_ratio=0.95, created_at=now),
    ]

    # Assign these snapshots to one of the lists (e.g., CPU snapshots)
    snapshotter._cpu_snapshots = snapshots

    # Prune snapshots older than 2 hours
    snapshots_casted = cast(list[Snapshot], snapshotter._cpu_snapshots)
    snapshotter._prune_snapshots(snapshots_casted, now)

    # Check that only the last two snapshots remain
    assert len(snapshotter._cpu_snapshots) == 2
    assert snapshotter._cpu_snapshots[0].created_at == one_hour_ago
    assert snapshotter._cpu_snapshots[1].created_at == now


def test_memory_load_evaluation_logs_warning_on_high_usage(caplog: pytest.LogCaptureFixture) -> None:
    snapshotter = Snapshotter(AsyncMock(spec=EventManager), max_memory_size=ByteSize.from_gb(8))

    high_memory_usage = ByteSize.from_gb(8) * 0.95  # 95% of 8 GB

    snapshotter._evaluate_memory_load(
        current_memory_usage_size=high_memory_usage,
        snapshot_timestamp=datetime.now(timezone.utc),
    )

    assert len(caplog.records) == 1
    assert caplog.records[0].levelname.lower() == 'warning'
    assert 'Memory is critically overloaded' in caplog.records[0].msg

    # It should not log again, since the last log was short time ago
    snapshotter._evaluate_memory_load(
        current_memory_usage_size=high_memory_usage,
        snapshot_timestamp=datetime.now(timezone.utc),
    )

    assert len(caplog.records) == 1


def test_memory_load_evaluation_silent_on_acceptable_usage(
    monkeypatch: pytest.MonkeyPatch,
    snapshotter: Snapshotter,
) -> None:
    mock_logger_warn = MagicMock()
    monkeypatch.setattr(getLogger('crawlee.autoscaling.snapshotter'), 'warning', mock_logger_warn)