40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
import io
|
|
import tarfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from great_ai.large_file import LargeFileS3
|
|
|
|
|
|
def test_download_quota_keeps_the_requested_file_and_evicts_older_models(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
backend = MagicMock()
|
|
backend.list_objects_v2.side_effect = lambda **kwargs: {
|
|
"Contents": [{"Key": kwargs["Prefix"] + "0", "ETag": "one"}]
|
|
}
|
|
|
|
def download(
|
|
self: LargeFileS3, remote_path: Any, path: Path, hide_progress: bool
|
|
) -> None:
|
|
data = b"x" * (2000 if self._name == "oversized" else 600)
|
|
with tarfile.open(path, "w:gz") as archive:
|
|
member = tarfile.TarInfo(self._local_name)
|
|
member.size = len(data)
|
|
archive.addfile(member, io.BytesIO(data))
|
|
|
|
with (
|
|
patch.object(LargeFileS3, "_client", backend),
|
|
patch.object(LargeFileS3, "_download", download),
|
|
patch.object(LargeFileS3, "cache_path", tmp_path / "cache"),
|
|
patch.object(LargeFileS3, "max_cache_size", "1 KB"),
|
|
):
|
|
first = LargeFileS3("first").get()
|
|
second = LargeFileS3("second").get()
|
|
assert not first.exists()
|
|
assert second.read_bytes() == b"x" * 600
|
|
with pytest.raises(ValueError, match="exceeds max_cache_size"):
|
|
LargeFileS3("oversized").get()
|
|
assert second.exists()
|