|
| 1 | +import io |
| 2 | + |
| 3 | +import pytest |
| 4 | +from pytest_mock import MockerFixture |
| 5 | +from pytest_test_utils import TempDirFactory, TmpDir |
| 6 | + |
| 7 | +from scmrepo.git import Git |
| 8 | +from scmrepo.git.lfs import LFSStorage, Pointer, smudge |
| 9 | + |
| 10 | +FOO_OID = "2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae" |
| 11 | +FOO_POINTER = ( |
| 12 | + f"version https://git-lfs.github.com/spec/v1\noid sha256:{FOO_OID}\nsize 3\n" |
| 13 | +).encode() |
| 14 | + |
| 15 | + |
| 16 | +@pytest.fixture |
| 17 | +def storage(tmp_dir_factory: TempDirFactory) -> LFSStorage: |
| 18 | + storage_path = tmp_dir_factory.mktemp("lfs") |
| 19 | + return LFSStorage(storage_path) |
| 20 | + |
| 21 | + |
| 22 | +@pytest.fixture |
| 23 | +def lfs(tmp_dir: TmpDir, scm: Git) -> None: |
| 24 | + tmp_dir.gen(".gitattributes", "*.lfs filter=lfs diff=lfs merge=lfs -text") |
| 25 | + scm.add([".gitattributes"]) |
| 26 | + scm.commit("init lfs attributes") |
| 27 | + |
| 28 | + |
| 29 | +@pytest.fixture |
| 30 | +def lfs_objects(tmp_dir: TmpDir) -> TmpDir: |
| 31 | + objects = tmp_dir / ".git" / "lfs" / "objects" |
| 32 | + objects.mkdir(parents=True) |
| 33 | + return objects |
| 34 | + |
| 35 | + |
| 36 | +def test_pointer_build(tmp_dir: TmpDir): |
| 37 | + tmp_dir.gen("foo", "foo") |
| 38 | + with open(tmp_dir / "foo", "rb") as fobj: |
| 39 | + pointer = Pointer.build(fobj) |
| 40 | + |
| 41 | + assert pointer.dump() == FOO_POINTER.decode("utf-8") |
| 42 | + |
| 43 | + |
| 44 | +def test_pointer_load(tmp_dir: TmpDir): |
| 45 | + tmp_dir.gen("foo.lfs", FOO_POINTER) |
| 46 | + with open(tmp_dir / "foo.lfs", "rb") as fobj: |
| 47 | + pointer = Pointer.load(fobj) |
| 48 | + assert pointer.oid == FOO_OID |
| 49 | + assert pointer.size == 3 |
| 50 | + |
| 51 | + |
| 52 | +def test_smudge(tmp_dir: TmpDir, storage: LFSStorage, mocker: MockerFixture): |
| 53 | + tmp_dir.gen("foo.lfs", FOO_POINTER) |
| 54 | + with open(tmp_dir / "foo.lfs", "rb") as fobj: |
| 55 | + assert smudge(storage, fobj).read() == FOO_POINTER |
| 56 | + |
| 57 | + mocker.patch.object(storage, "open", return_value=io.BytesIO(b"foo")) |
| 58 | + with open(tmp_dir / "foo.lfs", "rb") as fobj: |
| 59 | + assert smudge(storage, fobj).read() == b"foo" |
| 60 | + |
| 61 | + |
| 62 | +@pytest.mark.usefixtures("lfs") |
| 63 | +def test_lfs(tmp_dir: TmpDir, scm: Git, lfs_objects: TmpDir): |
| 64 | + # NOTE: scmrepo does not currently support LFS clean (writes), this writes |
| 65 | + # the pointer to the git odb (simulating an actual LFS clean) |
| 66 | + tmp_dir.gen("foo.lfs", FOO_POINTER) |
| 67 | + scm.add(["foo.lfs"]) |
| 68 | + scm.commit("add foo") |
| 69 | + lfs_objects.gen({FOO_OID[0:2]: {FOO_OID[2:4]: {FOO_OID: "foo"}}}) |
| 70 | + |
| 71 | + fs = scm.get_fs("HEAD") |
| 72 | + with fs.open("foo.lfs", "rb", raw=True) as fobj: |
| 73 | + assert fobj.read() == FOO_POINTER |
| 74 | + with fs.open("foo.lfs", "rb", raw=False) as fobj: |
| 75 | + assert fobj.read() == b"foo" |
0 commit comments