Update porcelain and tests for improved LFS integration

Update LFS filter driver instantiation to pass repository context and
adjust tests to work with the new filter interface and improved LFS
handling.

dulwich/porcelain.py

@@ -5068,7 +5068,7 @@ def lfs_clean(repo=".", path=None):
 
         # Get LFS store
         lfs_store = LFSStore.from_repo(r)
-        filter_driver = LFSFilterDriver(lfs_store)
+        filter_driver = LFSFilterDriver(lfs_store, repo=r)
 
         # Read file content
         full_path = os.path.join(r.path, path)
@@ -5097,7 +5097,7 @@ def lfs_smudge(repo=".", pointer_content=None):
 
         # Get LFS store
         lfs_store = LFSStore.from_repo(r)
-        filter_driver = LFSFilterDriver(lfs_store)
+        filter_driver = LFSFilterDriver(lfs_store, repo=r)
 
         # Smudge the pointer (retrieve actual content)
         return filter_driver.smudge(pointer_content)
@@ -5162,7 +5162,7 @@ def lfs_migrate(repo=".", include=None, exclude=None, everything=False):
     with open_repo_closing(repo) as r:
         # Initialize LFS if needed
         lfs_store = LFSStore.from_repo(r, create=True)
-        filter_driver = LFSFilterDriver(lfs_store)
+        filter_driver = LFSFilterDriver(lfs_store, repo=r)
 
         # Get current index
         index = r.open_index()

tests/compat/test_lfs.py

@@ -355,14 +355,25 @@ class LFSCloneCompatTest(LFSCompatTestCase):
         cloned_repo = porcelain.clone(source_dir, target_dir)
         self.addCleanup(cloned_repo.close)
 
-        # Verify LFS file exists as pointer
+        # Verify LFS file exists
         cloned_file = os.path.join(target_dir, "test.bin")
         with open(cloned_file, "rb") as f:
             content = f.read()
 
-        # Should be a pointer, not the full content
-        self.assertLess(len(content), 1000)  # Pointer is much smaller
-        self.assertIn(b"version https://git-lfs.github.com/spec/v1", content)
+        # Check if filter.lfs.smudge is configured
+        cloned_config = cloned_repo.get_config()
+        try:
+            lfs_smudge = cloned_config.get((b"filter", b"lfs"), b"smudge")
+            has_lfs_config = bool(lfs_smudge)
+        except KeyError:
+            has_lfs_config = False
+
+        if has_lfs_config:
+            # git-lfs smudge filter should have converted it
+            self.assertEqual(content, test_content)
+        else:
+            # No git-lfs config (uses built-in filter), should be a pointer
+            self.assertIn(b"version https://git-lfs.github.com/spec/v1", content)
 
 
 if __name__ == "__main__":

tests/test_lfs.py

@@ -22,10 +22,13 @@
 """Tests for LFS support."""
 
 import json
+import os
 import shutil
 import tempfile
 
+from dulwich import porcelain
 from dulwich.lfs import LFSFilterDriver, LFSPointer, LFSStore
+from dulwich.repo import Repo
 
 from . import TestCase
 
@@ -306,6 +309,107 @@ class LFSIntegrationTests(TestCase):
                 f"Failed for content: {content!r}",
             )
 
+    def test_builtin_lfs_clone_no_config(self) -> None:
+        """Test cloning with LFS when no git-lfs commands are configured."""
+        # Create source repository
+        source_dir = os.path.join(self.test_dir, "source")
+        os.makedirs(source_dir)
+        source_repo = Repo.init(source_dir)
+
+        # Create empty config (no LFS commands)
+        config = source_repo.get_config()
+        config.write_to_path()
+
+        # Create .gitattributes with LFS filter
+        gitattributes_path = os.path.join(source_dir, ".gitattributes")
+        with open(gitattributes_path, "wb") as f:
+            f.write(b"*.bin filter=lfs\n")
+
+        # Create test content and store in LFS
+        test_content = b"Test binary content"
+        test_oid = LFSStore.from_repo(source_repo, create=True).write_object(
+            [test_content]
+        )
+
+        # Create LFS pointer file
+        pointer = LFSPointer(test_oid, len(test_content))
+        pointer_file = os.path.join(source_dir, "test.bin")
+        with open(pointer_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Commit files
+        porcelain.add(source_repo, paths=[".gitattributes", "test.bin"])
+        porcelain.commit(source_repo, message=b"Add LFS tracked file")
+        source_repo.close()
+
+        # Clone the repository
+        target_dir = os.path.join(self.test_dir, "target")
+        target_repo = porcelain.clone(source_dir, target_dir)
+
+        # Verify no LFS commands in config
+        target_config = target_repo.get_config_stack()
+        with self.assertRaises(KeyError):
+            target_config.get((b"filter", b"lfs"), b"smudge")
+
+        # Check the cloned file
+        cloned_file = os.path.join(target_dir, "test.bin")
+        with open(cloned_file, "rb") as f:
+            content = f.read()
+
+        # Should still be a pointer (LFS object not in target's store)
+        self.assertTrue(
+            content.startswith(b"version https://git-lfs.github.com/spec/v1")
+        )
+        self.assertIn(test_oid.encode(), content)
+        target_repo.close()
+
+    def test_builtin_lfs_with_local_objects(self) -> None:
+        """Test built-in LFS filter when objects are available locally."""
+        # No LFS config
+        config = self.repo.get_config()
+        config.write_to_path()
+
+        # Create .gitattributes
+        gitattributes_path = os.path.join(self.test_dir, ".gitattributes")
+        with open(gitattributes_path, "wb") as f:
+            f.write(b"*.dat filter=lfs\n")
+
+        # Create LFS store and add object
+        test_content = b"Hello from LFS!"
+        lfs_store = LFSStore.from_repo(self.repo, create=True)
+        test_oid = lfs_store.write_object([test_content])
+
+        # Create pointer file
+        pointer = LFSPointer(test_oid, len(test_content))
+        pointer_file = os.path.join(self.test_dir, "data.dat")
+        with open(pointer_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Commit
+        porcelain.add(self.repo, paths=[".gitattributes", "data.dat"])
+        porcelain.commit(self.repo, message=b"Add LFS file")
+
+        # Reset index to trigger checkout with filter
+        self.repo.reset_index()
+
+        # Check file content
+        with open(pointer_file, "rb") as f:
+            content = f.read()
+
+        # Built-in filter should have converted pointer to actual content
+        self.assertEqual(content, test_content)
+
+    def test_builtin_lfs_filter_used(self) -> None:
+        """Verify that built-in LFS filter is used when no config exists."""
+        # Get filter registry
+        normalizer = self.repo.get_blob_normalizer()
+        filter_registry = normalizer.filter_registry
+        lfs_driver = filter_registry.get_driver("lfs")
+
+        # Should be built-in LFS filter
+        self.assertIsInstance(lfs_driver, LFSFilterDriver)
+        self.assertEqual(type(lfs_driver).__module__, "dulwich.lfs")
+
 
 class LFSFilterDriverTests(TestCase):
     def setUp(self) -> None:
@@ -873,7 +977,7 @@ class LFSClientTests(TestCase):
         self.addCleanup(self.server.shutdown)
 
         # Create LFS client pointing to our test server
-        self.client = LFSClient(f"{self.server_url}/objects")
+        self.client = LFSClient(self.server_url)
 
     def test_client_url_normalization(self) -> None:
         """Test that client URL is normalized correctly."""

tests/test_porcelain_filters.py

@@ -21,6 +21,7 @@
 
 """Tests for porcelain filter integration."""
 
+import hashlib
 import os
 import tempfile
 from io import BytesIO
@@ -215,6 +216,39 @@ class PorcelainFilterTests(TestCase):
             # The checkout should apply the smudge filter
             self.assertIn(b"\r\n", content)
 
+    def test_process_filter_priority(self) -> None:
+        """Test that process filters take priority over built-in ones."""
+        # Create a custom filter script
+        filter_script = os.path.join(self.test_dir, "test-filter.sh")
+        with open(filter_script, "w") as f:
+            f.write("#!/bin/sh\necho 'FILTERED'")
+        os.chmod(filter_script, 0o755)
+
+        # Configure custom filter
+        config = self.repo.get_config()
+        config.set((b"filter", b"test"), b"smudge", filter_script.encode())
+        config.write_to_path()
+
+        # Create .gitattributes
+        gitattributes = os.path.join(self.test_dir, ".gitattributes")
+        with open(gitattributes, "wb") as f:
+            f.write(b"*.txt filter=test\n")
+
+        # Test filter application
+        from dulwich.filters import FilterRegistry
+
+        filter_registry = FilterRegistry(config, self.repo)
+        test_driver = filter_registry.get_driver("test")
+
+        # Should be ProcessFilterDriver, not built-in
+        from dulwich.filters import ProcessFilterDriver
+
+        self.assertIsInstance(test_driver, ProcessFilterDriver)
+
+        # Test smudge
+        result = test_driver.smudge(b"original", b"test.txt")
+        self.assertEqual(result, b"FILTERED\n")
+
     def test_commit_with_clean_filter(self) -> None:
         """Test committing with a clean filter."""
         # Set up a custom filter in git config
@@ -241,6 +275,87 @@ class PorcelainFilterTests(TestCase):
         # The committed blob should have filtered content
         # (Note: actual filter execution requires process filter support)
 
+    def test_clone_with_builtin_lfs_filter(self) -> None:
+        """Test cloning with built-in LFS filter (no subprocess)."""
+        # Create a source repository with LFS
+        source_dir = tempfile.mkdtemp()
+        self.addCleanup(rmtree_ro, source_dir)
+        source_repo = Repo.init(source_dir)
+        self.addCleanup(source_repo.close)
+
+        # Create .gitattributes with LFS filter
+        gitattributes_path = os.path.join(source_dir, ".gitattributes")
+        with open(gitattributes_path, "wb") as f:
+            f.write(b"*.bin filter=lfs\n")
+
+        # Create LFS pointer file manually
+        from dulwich.lfs import LFSPointer
+
+        pointer = LFSPointer(
+            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", 0
+        )
+        pointer_file = os.path.join(source_dir, "empty.bin")
+        with open(pointer_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Create actual LFS object in the store
+        from dulwich.lfs import LFSStore
+
+        lfs_store = LFSStore.from_repo(source_repo, create=True)
+        lfs_store.write_object([b""])  # Empty file content
+
+        # Commit the files
+        porcelain.add(source_repo, paths=[".gitattributes", "empty.bin"])
+        porcelain.commit(source_repo, message=b"Add LFS file")
+
+        # Clone the repository (should use built-in LFS filter)
+        target_dir = tempfile.mkdtemp()
+        self.addCleanup(rmtree_ro, target_dir)
+
+        # Clone with built-in filter (no git-lfs config)
+        target_repo = porcelain.clone(source_dir, target_dir)
+        self.addCleanup(target_repo.close)
+
+        # Verify the file was checked out with the filter
+        target_file = os.path.join(target_dir, "empty.bin")
+        with open(target_file, "rb") as f:
+            content = f.read()
+
+        # Without git-lfs configured, the built-in filter is used
+        # Since the LFS object isn't in the target repo's store,
+        # it should remain as a pointer
+        self.assertIn(b"version https://git-lfs", content)
+        self.assertIn(
+            b"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", content
+        )
+
+    def test_builtin_lfs_filter_with_object(self) -> None:
+        """Test built-in LFS filter when object is available in store."""
+        # Create test content
+        test_content = b"Hello, LFS!"
+        test_oid = hashlib.sha256(test_content).hexdigest()
+
+        # Create LFS pointer
+        from dulwich.lfs import LFSPointer
+
+        pointer = LFSPointer(test_oid, len(test_content))
+
+        # Create LFS store and write object
+        from dulwich.lfs import LFSStore
+
+        lfs_store = LFSStore.from_repo(self.repo, create=True)
+        lfs_store.write_object([test_content])
+
+        # Test smudge filter
+        from dulwich.filters import FilterRegistry
+
+        filter_registry = FilterRegistry(self.repo.get_config_stack(), self.repo)
+        lfs_driver = filter_registry.get_driver("lfs")
+
+        # Smudge should return actual content since object is in store
+        smudged = lfs_driver.smudge(pointer.to_bytes(), b"test.txt")
+        self.assertEqual(smudged, test_content)
+
     def test_ls_files_with_filters(self) -> None:
         """Test ls-files respects filter settings."""
         # Configure autocrlf

tests/test_porcelain_lfs.py

@@ -239,6 +239,192 @@ class LFSPorcelainTestCase(TestCase):
         self.assertIsNone(results["regular.txt"])
         self.assertIsNone(results["nonexistent.txt"])
 
+    def test_clone_with_builtin_lfs_no_config(self):
+        """Test cloning with built-in LFS filter when no git-lfs config exists."""
+        # Create a source repo with LFS content
+        source_dir = tempfile.mkdtemp()
+        self.addCleanup(lambda: self._cleanup_test_dir_path(source_dir))
+        source_repo = Repo.init(source_dir)
+
+        # Create .gitattributes
+        gitattributes_path = os.path.join(source_dir, ".gitattributes")
+        with open(gitattributes_path, "w") as f:
+            f.write("*.bin filter=lfs diff=lfs merge=lfs -text\n")
+
+        # Create test content and store in LFS
+        # LFSStore.from_repo with create=True will create the directories
+        test_content = b"This is test content for LFS"
+        lfs_store = LFSStore.from_repo(source_repo, create=True)
+        oid = lfs_store.write_object([test_content])
+
+        # Create LFS pointer file
+        pointer = LFSPointer(oid, len(test_content))
+        test_file = os.path.join(source_dir, "test.bin")
+        with open(test_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Add and commit
+        porcelain.add(source_repo, paths=[".gitattributes", "test.bin"])
+        porcelain.commit(source_repo, message=b"Add LFS file")
+
+        # Clone with empty config (no git-lfs commands)
+        clone_dir = tempfile.mkdtemp()
+        self.addCleanup(lambda: self._cleanup_test_dir_path(clone_dir))
+
+        # Verify source repo has no LFS filter config
+        config = source_repo.get_config()
+        with self.assertRaises(KeyError):
+            config.get((b"filter", b"lfs"), b"smudge")
+
+        # Clone the repository
+        cloned_repo = porcelain.clone(source_dir, clone_dir)
+
+        # Verify that built-in LFS filter was used
+        normalizer = cloned_repo.get_blob_normalizer()
+        if hasattr(normalizer, "filter_registry"):
+            lfs_driver = normalizer.filter_registry.get_driver("lfs")
+            # Should be the built-in LFSFilterDriver
+            self.assertEqual(type(lfs_driver).__name__, "LFSFilterDriver")
+            self.assertEqual(type(lfs_driver).__module__, "dulwich.lfs")
+
+        # Check that the file remains as a pointer (expected behavior)
+        # The built-in LFS filter preserves pointers when objects aren't available
+        cloned_file = os.path.join(clone_dir, "test.bin")
+        with open(cloned_file, "rb") as f:
+            content = f.read()
+
+        # Should still be a pointer since objects weren't transferred
+        self.assertTrue(
+            content.startswith(b"version https://git-lfs.github.com/spec/v1")
+        )
+        cloned_pointer = LFSPointer.from_bytes(content)
+        self.assertIsNotNone(cloned_pointer)
+        self.assertEqual(cloned_pointer.oid, pointer.oid)
+        self.assertEqual(cloned_pointer.size, pointer.size)
+
+        source_repo.close()
+        cloned_repo.close()
+
+    def _cleanup_test_dir_path(self, path):
+        """Clean up a test directory by path."""
+        import shutil
+
+        shutil.rmtree(path, ignore_errors=True)
+
+    def test_add_applies_clean_filter(self):
+        """Test that add operation applies LFS clean filter."""
+        # Don't use lfs_init to avoid configuring git-lfs commands
+        # Create LFS store manually
+        lfs_store = LFSStore.from_repo(self.repo, create=True)
+
+        # Create .gitattributes
+        gitattributes_path = os.path.join(self.repo.path, ".gitattributes")
+        with open(gitattributes_path, "w") as f:
+            f.write("*.bin filter=lfs diff=lfs merge=lfs -text\n")
+
+        # Create a file that should be cleaned to LFS
+        test_content = b"This is large file content that should be stored in LFS"
+        test_file = os.path.join(self.repo.path, "large.bin")
+        with open(test_file, "wb") as f:
+            f.write(test_content)
+
+        # Add the file - this should apply the clean filter
+        porcelain.add(self.repo, paths=["large.bin"])
+
+        # Check that the file was cleaned to a pointer in the index
+        index = self.repo.open_index()
+        entry = index[b"large.bin"]
+
+        # Get the blob from the object store
+        blob = self.repo.get_object(entry.sha)
+        content = blob.data
+
+        # Should be an LFS pointer
+        self.assertTrue(
+            content.startswith(b"version https://git-lfs.github.com/spec/v1")
+        )
+        pointer = LFSPointer.from_bytes(content)
+        self.assertIsNotNone(pointer)
+        self.assertEqual(pointer.size, len(test_content))
+
+        # Verify the actual content was stored in LFS
+        with lfs_store.open_object(pointer.oid) as f:
+            stored_content = f.read()
+        self.assertEqual(stored_content, test_content)
+
+    def test_checkout_applies_smudge_filter(self):
+        """Test that checkout operation applies LFS smudge filter."""
+        # Create LFS store and content
+        lfs_store = LFSStore.from_repo(self.repo, create=True)
+
+        # Create .gitattributes
+        gitattributes_path = os.path.join(self.repo.path, ".gitattributes")
+        with open(gitattributes_path, "w") as f:
+            f.write("*.bin filter=lfs diff=lfs merge=lfs -text\n")
+
+        # Create test content and store in LFS
+        test_content = b"This is the actual file content from LFS"
+        oid = lfs_store.write_object([test_content])
+
+        # Create LFS pointer file
+        pointer = LFSPointer(oid, len(test_content))
+        test_file = os.path.join(self.repo.path, "data.bin")
+        with open(test_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Add and commit the pointer
+        porcelain.add(self.repo, paths=[".gitattributes", "data.bin"])
+        porcelain.commit(self.repo, message=b"Add LFS file")
+
+        # Remove the file from working directory
+        os.remove(test_file)
+
+        # Checkout the file - this should apply the smudge filter
+        porcelain.checkout(self.repo, paths=["data.bin"])
+
+        # Verify the file was expanded from pointer to content
+        with open(test_file, "rb") as f:
+            content = f.read()
+
+        self.assertEqual(content, test_content)
+
+    def test_reset_hard_applies_smudge_filter(self):
+        """Test that reset --hard applies LFS smudge filter."""
+        # Create LFS store and content
+        lfs_store = LFSStore.from_repo(self.repo, create=True)
+
+        # Create .gitattributes
+        gitattributes_path = os.path.join(self.repo.path, ".gitattributes")
+        with open(gitattributes_path, "w") as f:
+            f.write("*.bin filter=lfs diff=lfs merge=lfs -text\n")
+
+        # Create test content and store in LFS
+        test_content = b"Content that should be restored by reset"
+        oid = lfs_store.write_object([test_content])
+
+        # Create LFS pointer file
+        pointer = LFSPointer(oid, len(test_content))
+        test_file = os.path.join(self.repo.path, "reset-test.bin")
+        with open(test_file, "wb") as f:
+            f.write(pointer.to_bytes())
+
+        # Add and commit
+        porcelain.add(self.repo, paths=[".gitattributes", "reset-test.bin"])
+        commit_sha = porcelain.commit(self.repo, message=b"Add LFS file for reset test")
+
+        # Modify the file in working directory
+        with open(test_file, "wb") as f:
+            f.write(b"Modified content that should be discarded")
+
+        # Reset hard - this should restore the file with smudge filter applied
+        porcelain.reset(self.repo, mode="hard", treeish=commit_sha)
+
+        # Verify the file was restored with LFS content
+        with open(test_file, "rb") as f:
+            content = f.read()
+
+        self.assertEqual(content, test_content)
+
 
 if __name__ == "__main__":
     unittest.main()