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

Code before:
    # Create mock snapshots with varying creation times
    snapshots = Snapshotter._get_sorted_list_by_created_at(
        [
            CpuSnapshot(used_ratio=0.5, max_used_ratio=0.95, created_at=five_hours_ago),
            CpuSnapshot(used_ratio=0.6, max_used_ratio=0.95, created_at=three_hours_ago),
            CpuSnapshot(used_ratio=0.7, max_used_ratio=0.95, created_at=two_hours_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 == two_hours_ago
    assert snapshotter._cpu_snapshots[1].created_at == now


def test_pruning_empty_snapshot_list_remains_empty(snapshotter: Snapshotter) -> None:
    now = datetime.now(timezone.utc)
    snapshotter._cpu_snapshots = Snapshotter._get_sorted_list_by_created_at([])
    snapshots_casted = cast(list[Snapshot], snapshotter._cpu_snapshots)
    snapshotter._prune_snapshots(snapshots_casted, now)
    assert snapshotter._cpu_snapshots == []


def test_snapshot_pruning_keeps_recent_records_unaffected(snapshotter: Snapshotter) -> None:
    snapshotter._snapshot_history = timedelta(hours=2)

    # Create timestamps for testing
    now = datetime.now(timezone.utc)
    one_hour_ago = now - timedelta(hours=1)

    # Create mock snapshots with varying creation times
    snapshots = Snapshotter._get_sorted_list_by_created_at(
        [
            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)

Review comment (suggestion):
```suggestion
    snapshotter._cpu_snapshots = Snapshotter._get_sorted_list_by_created_at(list[CpuSnapshot]())
```

Code after:
    # Create mock snapshots with varying creation times
    snapshots = Snapshotter._get_sorted_list_by_created_at(
        [
            CpuSnapshot(used_ratio=0.5, max_used_ratio=0.95, created_at=five_hours_ago),
            CpuSnapshot(used_ratio=0.6, max_used_ratio=0.95, created_at=three_hours_ago),
            CpuSnapshot(used_ratio=0.7, max_used_ratio=0.95, created_at=two_hours_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 == two_hours_ago
    assert snapshotter._cpu_snapshots[1].created_at == now


def test_pruning_empty_snapshot_list_remains_empty(snapshotter: Snapshotter) -> None:
    now = datetime.now(timezone.utc)
    snapshotter._cpu_snapshots = Snapshotter._get_sorted_list_by_created_at(list[CpuSnapshot]())
    snapshots_casted = cast(list[Snapshot], snapshotter._cpu_snapshots)
    snapshotter._prune_snapshots(snapshots_casted, now)
    assert snapshotter._cpu_snapshots == []


def test_snapshot_pruning_keeps_recent_records_unaffected(snapshotter: Snapshotter) -> None:
    snapshotter._snapshot_history = timedelta(hours=2)

    # Create timestamps for testing
    now = datetime.now(timezone.utc)
    one_hour_ago = now - timedelta(hours=1)

    # Create mock snapshots with varying creation times
    snapshots = Snapshotter._get_sorted_list_by_created_at(
        [
            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)