feat: If core.autocrlf is enabled, replace CRLF with LF when adding to index

Author: Ruben DI BATTISTA

git/index/base.py

@@ -9,7 +9,10 @@
 import os.path as osp
 import subprocess
 import tempfile
+from contextlib import contextmanager
+from enum import auto
 from io import BytesIO
+from pathlib import Path
 from stat import S_ISLNK
 from typing import (
     IO,
@@ -22,6 +25,7 @@
     Iterator,
     List,
     NoReturn,
+    Optional,
     Sequence,
     Tuple,
     Type,
@@ -72,7 +76,43 @@
 StageType = int
 Treeish = Union[Tree, Commit, str, bytes]
 
-# ------------------------------------------------------------------------------------
+# Store temporary file modifications -------------------------------------------
+
+class _FileStore:
+    """An utility class that stores original files somewhere and restores them
+    to the original content at the exit"""
+
+    _dir: PathLike
+    def __init__(self, tmp_dir: Optional[PathLike] = None):
+        self._file_map: dict[PathLike, PathLike] = {}
+        self._tmp_dir = tempfile.TemporaryDirectory(prefix=str(tmp_dir))
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc, value, tb):
+        for file, store_file in self._file_map.items():
+            with open(store_file, "rb") as rf, open(file, "wb") as wf:
+                for line in rf:
+                    wf.write(line)
+            Path(store_file).unlink()
+        self._dir.rmdir()
+
+    @property
+    def _dir(self) -> Path:
+        return Path(self._tmp_dir.name)
+
+    def save(self, file: PathLike) -> None:
+        store_file = self._dir / tempfile.mktemp()
+        self._file_map[file] = store_file
+        with open(store_file, "wb") as wf, open(file, "rb") as rf:
+            for line in rf:
+                wf.write(line)
+
+        
+
+
+# ------------------------------------------------------------------------------
 
 
 __all__ = ('IndexFile', 'CheckoutError')
@@ -578,8 +618,10 @@ def _to_relative_path(self, path: PathLike) -> PathLike:
             raise ValueError("Absolute path %r is not in git repository at %r" % (path, self.repo.working_tree_dir))
         return os.path.relpath(path, self.repo.working_tree_dir)
 
-    def _preprocess_add_items(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']]
-                              ) -> Tuple[List[PathLike], List[BaseIndexEntry]]:
+    def _preprocess_add_items(self, 
+        items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']],
+        file_store: _FileStore
+    ) -> Tuple[List[PathLike], List[BaseIndexEntry]]:
         """ Split the items into two lists of path strings and BaseEntries. """
         paths = []
         entries = []
@@ -589,6 +631,7 @@ def _preprocess_add_items(self, items: Sequence[Union[PathLike, Blob, BaseIndexE
 
         for item in items:
             if isinstance(item, (str, os.PathLike)):
+                self._autocrlf(item, file_store)
                 paths.append(self._to_relative_path(item))
             elif isinstance(item, (Blob, Submodule)):
                 entries.append(BaseIndexEntry.from_blob(item))
@@ -599,6 +642,29 @@ def _preprocess_add_items(self, items: Sequence[Union[PathLike, Blob, BaseIndexE
         # END for each item
         return paths, entries
 
+    def _autocrlf(self, file: PathLike, file_store: _FileStore) -> PathLike:
+        """If the config option `autocrlf` is True, replace CRLF with LF"""
+
+        reader = self.repo.config_reader()
+
+        autocrlf = reader.get_value("core", "autocrlf", False)
+
+        if not autocrlf:
+            return file
+
+        file_store.save(file)
+
+        with open(file, "rb") as f:
+            content = f.read()
+
+        content = content.replace(b"\r\n", b"\n")
+        
+        with open(file, "wb") as f:
+            f.write(content)
+
+        return file
+
+
     def _store_path(self, filepath: PathLike, fprogress: Callable) -> BaseIndexEntry:
         """Store file at filepath in the database and return the base index entry
         Needs the git_working_dir decorator active ! This must be assured in the calling code"""
@@ -753,70 +819,72 @@ def add(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']
         # sort the entries into strings and Entries, Blobs are converted to entries
         # automatically
         # paths can be git-added, for everything else we use git-update-index
-        paths, entries = self._preprocess_add_items(items)
-        entries_added: List[BaseIndexEntry] = []
-        # This code needs a working tree, therefore we try not to run it unless required.
-        # That way, we are OK on a bare repository as well.
-        # If there are no paths, the rewriter has nothing to do either
-        if paths:
-            entries_added.extend(self._entries_for_paths(paths, path_rewriter, fprogress, entries))
-
-        # HANDLE ENTRIES
-        if entries:
-            null_mode_entries = [e for e in entries if e.mode == 0]
-            if null_mode_entries:
-                raise ValueError(
-                    "At least one Entry has a null-mode - please use index.remove to remove files for clarity")
-            # END null mode should be remove
-
-            # HANDLE ENTRY OBJECT CREATION
-            # create objects if required, otherwise go with the existing shas
-            null_entries_indices = [i for i, e in enumerate(entries) if e.binsha == Object.NULL_BIN_SHA]
-            if null_entries_indices:
-                @ git_working_dir
-                def handle_null_entries(self: 'IndexFile') -> None:
-                    for ei in null_entries_indices:
-                        null_entry = entries[ei]
-                        new_entry = self._store_path(null_entry.path, fprogress)
-
-                        # update null entry
-                        entries[ei] = BaseIndexEntry(
-                            (null_entry.mode, new_entry.binsha, null_entry.stage, null_entry.path))
-                    # END for each entry index
-                # end closure
-                handle_null_entries(self)
-            # END null_entry handling
-
-            # REWRITE PATHS
-            # If we have to rewrite the entries, do so now, after we have generated
-            # all object sha's
-            if path_rewriter:
-                for i, e in enumerate(entries):
-                    entries[i] = BaseIndexEntry((e.mode, e.binsha, e.stage, path_rewriter(e)))
+        with _FileStore() as file_store:
+            paths, entries = self._preprocess_add_items(items, file_store)
+            import ipdb; ipdb.set_trace()
+            entries_added: List[BaseIndexEntry] = []
+            # This code needs a working tree, therefore we try not to run it unless required.
+            # That way, we are OK on a bare repository as well.
+            # If there are no paths, the rewriter has nothing to do either
+            if paths:
+                entries_added.extend(self._entries_for_paths(paths, path_rewriter, fprogress, entries))
+
+            # HANDLE ENTRIES
+            if entries:
+                null_mode_entries = [e for e in entries if e.mode == 0]
+                if null_mode_entries:
+                    raise ValueError(
+                        "At least one Entry has a null-mode - please use index.remove to remove files for clarity")
+                # END null mode should be remove
+
+                # HANDLE ENTRY OBJECT CREATION
+                # create objects if required, otherwise go with the existing shas
+                null_entries_indices = [i for i, e in enumerate(entries) if e.binsha == Object.NULL_BIN_SHA]
+                if null_entries_indices:
+                    @ git_working_dir
+                    def handle_null_entries(self: 'IndexFile') -> None:
+                        for ei in null_entries_indices:
+                            null_entry = entries[ei]
+                            new_entry = self._store_path(null_entry.path, fprogress)
+
+                            # update null entry
+                            entries[ei] = BaseIndexEntry(
+                                (null_entry.mode, new_entry.binsha, null_entry.stage, null_entry.path))
+                        # END for each entry index
+                    # end closure
+                    handle_null_entries(self)
+                # END null_entry handling
+
+                # REWRITE PATHS
+                # If we have to rewrite the entries, do so now, after we have generated
+                # all object sha's
+                if path_rewriter:
+                    for i, e in enumerate(entries):
+                        entries[i] = BaseIndexEntry((e.mode, e.binsha, e.stage, path_rewriter(e)))
+                    # END for each entry
+                # END handle path rewriting
+
+                # just go through the remaining entries and provide progress info
+                for i, entry in enumerate(entries):
+                    progress_sent = i in null_entries_indices
+                    if not progress_sent:
+                        fprogress(entry.path, False, entry)
+                        fprogress(entry.path, True, entry)
+                    # END handle progress
                 # END for each entry
-            # END handle path rewriting
-
-            # just go through the remaining entries and provide progress info
-            for i, entry in enumerate(entries):
-                progress_sent = i in null_entries_indices
-                if not progress_sent:
-                    fprogress(entry.path, False, entry)
-                    fprogress(entry.path, True, entry)
-                # END handle progress
-            # END for each entry
-            entries_added.extend(entries)
-        # END if there are base entries
-
-        # FINALIZE
-        # add the new entries to this instance
-        for entry in entries_added:
-            self.entries[(entry.path, 0)] = IndexEntry.from_base(entry)
-
-        if write:
-            self.write(ignore_extension_data=not write_extension_data)
-        # END handle write
+                entries_added.extend(entries)
+            # END if there are base entries
 
-        return entries_added
+            # FINALIZE
+            # add the new entries to this instance
+            for entry in entries_added:
+                self.entries[(entry.path, 0)] = IndexEntry.from_base(entry)
+
+            if write:
+                self.write(ignore_extension_data=not write_extension_data)
+            # END handle write
+
+            return entries_added
 
     def _items_to_rela_paths(self, items: Union[PathLike, Sequence[Union[PathLike, BaseIndexEntry, Blob, Submodule]]]
                              ) -> List[PathLike]:

test/test_index.py

@@ -29,6 +29,7 @@
 from git.cmd import Git
 from git.compat import is_win
 from git.exc import HookExecutionError, InvalidGitRepositoryError
+from git.index.base import _FileStore
 from git.index.fun import hook_path
 from git.index.typ import BaseIndexEntry, IndexEntry
 from git.objects import Blob
@@ -930,3 +931,25 @@ def test_index_add_pathlike(self, rw_repo):
         file.touch()
 
         rw_repo.index.add(file)
+
+
+
+def test_filestore(tmp_path): 
+    dummy_file = tmp_path / "dummy.txt"
+
+    content = "Dummy\n"
+
+    with open(dummy_file, "w") as f:
+        f.write(content)
+
+    with _FileStore() as fs: 
+        fs.save(dummy_file)
+
+        with open(dummy_file, "w") as f:
+            f.write(r"Something else\n")
+        
+    with open(dummy_file, "r") as f:
+        assert f.read() == content
+
+    
+