Update how git-theta track ${path} works.

Currently, when a pattern in `.gitattributes` matches the path in
`git-theta track`, the theta attributes (filter, merge, diff) are added
to that line. This can cause issues as it may result in tracking more
files with git-theta than expected.

However, just adding a new line that is an exact match for the path to
the end of the gitattribute file is not correct either. The *last*
attribute line in the file is the one used by Git. This could result in
`git-theta track` removing an some other attribute that is set for that
file.

This PR updates the way that git attributes are set. If there are
already git attributes set to non-theta values for the file that are used by git-theta
(filter, merge, diff) an error is raised. If these attributes are all
set to git-theta, no new entry is added, even when the entry is a
pattern match instead of an exact match. If the new file has no
attributes set, or attributes that don't overlap with git-theta, then a
new entry is added. Non-overlapping attributes are copied down if they
were set before.

When a line has an attribute set multiple times, the *last* one is used
so we can add the theta filters at the end to override any previous
ones.

Added a `is_theta_tracked` function similar to the one from
r-three#214 where the test for if a
file is tracked by Git-Theta is abstracted into a function.

However it is implemented differently as it now handles the subtleties of
what attribute line is active at a given time.

Only the final line that a path matches is used to set attributes and
only the last entry for some key is used. This is now respected.

Author: blester125

git_theta/git_utils.py

@@ -1,5 +1,7 @@
 """Utilities for manipulating git."""
 
+import copy
+import dataclasses
 import filecmp
 import fnmatch
 import io
@@ -10,7 +12,7 @@
 import shutil
 import subprocess
 import sys
-from typing import List, Sequence, Union
+from typing import Dict, List, Optional, Sequence, Union
 
 import git
 
@@ -107,7 +109,24 @@ def get_gitattributes_file(repo):
     return os.path.join(repo.working_dir, ".gitattributes")
 
 
-def read_gitattributes(gitattributes_file):
+@dataclasses.dataclass
+class GitAttributes:
+    pattern: str
+    attributes: Dict[str, str]
+    raw: Optional[str] = None
+
+    def __str__(self):
+        if self.raw:
+            return self.raw
+        attrs = " ".join(f"{k}={v}" if v else k for k, v in self.attributes.items())
+        return f"{self.pattern} {attrs}"
+
+    def __eq__(self, o):
+        raw_eq = self.raw == o.raw if self.raw and o.raw else True
+        return self.pattern == o.pattern and self.attributes == o.attributes and raw_eq
+
+
+def read_gitattributes(gitattributes_file) -> List[GitAttributes]:
     """
     Read contents of this repo's .gitattributes file
 
@@ -123,14 +142,30 @@ def read_gitattributes(gitattributes_file):
     """
     if os.path.exists(gitattributes_file):
         with open(gitattributes_file, "r") as f:
-            return [line.rstrip("\n") for line in f]
+            return [parse_gitattributes(line.rstrip("\n")) for line in f]
     else:
         return []
 
 
+def parse_gitattributes(gitattributes: str) -> GitAttributes:
+    # TODO: Fix for escaped patterns
+    pattern, *attributes = gitattributes.split(" ")
+    attrs = {}
+    # Overwrite as we go to get the LAST attribute behavior
+    for attribute in attributes:
+        # TODO: Update to handle unsetting attribute, etc.
+        if "=" in attribute:
+            key, value = attribute.split("=")
+        else:
+            key = attribute
+            value = None
+        attrs[key] = value
+    return GitAttributes(pattern, attrs, gitattributes)
+
+
 @file_or_name(gitattributes_file="w")
 def write_gitattributes(
-    gitattributes_file: Union[str, io.FileIO], attributes: List[str]
+    gitattributes_file: Union[str, io.FileIO], attributes: List[GitAttributes]
 ):
     """
     Write list of attributes to this repo's .gitattributes file
@@ -143,60 +178,136 @@ def write_gitattributes(
     attributes:
         Attributes to write to .gitattributes
     """
-    gitattributes_file.write("\n".join(attributes))
+    gitattributes_file.write("\n".join(map(str, attributes)))
     # End file with newline.
     gitattributes_file.write("\n")
 
 
-def add_theta_to_gitattributes(gitattributes: List[str], path: str) -> str:
-    """Add a filter=theta that covers file_name.
-
-    Parameters
-    ----------
-        gitattributes: A list of the lines from the gitattribute files.
-        path: The path to the model we are adding a filter to.
-
-    Returns
-    -------
-    List[str]
-        The lines to write to the new gitattribute file with a (possibly) new
-        filter=theta added that covers the given file.
-    """
-    pattern_found = False
-    new_gitattributes = []
-    for line in gitattributes:
-        # TODO(bdlester): Revisit this regex to see if it when the pattern
-        # is escaped due to having spaces in it.
-        match = re.match(r"^\s*(?P<pattern>[^\s]+)\s+(?P<attributes>.*)$", line)
-        if match:
-            # If there is already a pattern that covers the file, add the filter
-            # to that.
-            if fnmatch.fnmatchcase(path, match.group("pattern")):
-                pattern_found = True
-                if not "filter=theta" in match.group("attributes"):
-                    line = f"{line.rstrip()} filter=theta"
-                if not "merge=theta" in match.group("attributes"):
-                    line = f"{line.rstrip()} merge=theta"
-                if not "diff=theta" in match.group("attributes"):
-                    line = f"{line.rstrip()} diff=theta"
-        new_gitattributes.append(line)
-    # If we don't find a matching pattern, add a new line that covers just this
-    # specific file.
-    if not pattern_found:
-        new_gitattributes.append(f"{path} filter=theta merge=theta diff=theta")
-    return new_gitattributes
+def add_theta_to_gitattributes(
+    gitattributes: List[GitAttributes], path: str
+) -> List[GitAttributes]:
+    """Add git attributes required by git-theta for path."""
+    theta_attributes = ("filter", "merge", "diff")
+    previous_attribute = None
+    # Find if an active gitattribute entry applies to path
+    for gitattribute in gitattributes[::-1]:
+        if fnmatch.fnmatchcase(path, gitattribute.pattern):
+            previous_attribute = gitattribute
+            break
+    # If path is already managed by a git attributes entry.
+    if previous_attribute:
+        # If all of the theta attributes are set, we don't do anything.
+        if all(
+            previous_attribute.attributes.get(attr) == "theta"
+            for attr in theta_attributes
+        ):
+            return gitattributes
+        # If any of the attributes theta uses is set to something else, error out.
+        if any(
+            attr in previous_attribute.attributes
+            and previous_attribute.attributes[attr] != "theta"
+            for attr in theta_attributes
+        ):
+            raise ValueError(
+                f"Git Attributes used by git-theta are already set for {path}. "
+                f"Found filter={previous_attribute.attributes.get('filter')}, "
+                f"diff={previous_attribute.attributes.get('diff')}, "
+                f"merge={previous_attribute.attributes.get('merge')}."
+            )
+    # If the old entry set other attributes, make sure they are preserved.
+    attributes = (
+        copy.deepcopy(previous_attribute.attributes) if previous_attribute else {}
+    )
+    for attr in theta_attributes:
+        attributes[attr] = "theta"
+    new_attribute = GitAttributes(path, attributes)
+    gitattributes.append(new_attribute)
+    return gitattributes
+
+
+# def add_theta_to_gitattributes(gitattributes: List[str], path: str) -> str:
+#     """Add a filter=theta that covers file_name.
+
+#     If there is a pattern that covers the current file that applies the git-theta
+#     attribute, no new pattern is added. If there is a pattern that covers the
+#     current file and sets attributes used for git-theta an error is raised. If
+#     there is a pattern that sets non-overlapping attributes they are copied into
+#     a new path-specific pattern. If there is no match, a new path-specific
+#     pattern is always created.
+
+#     Parameters
+#     ----------
+#         gitattributes: A list of the lines from the gitattribute files.
+#         path: The path to the model we are adding a filter to.
+
+#     Returns
+#     -------
+#     List[str]
+#         The lines to write to the new gitattribute file with a (possibly) new
+#         filter=theta added that covers the given file.
+#     """
+#     new_gitattributes = []
+#     theta_attributes = "filter=theta merge=theta diff=theta"
+#     previous_attributes = None
+#     previous_match = None
+#     # Find the *last* attribute line that matches the path.
+#     for i, line in enumerate(gitattributes):
+#         # TODO(bdlester): Revisit this regex to see if it when the pattern
+#         # is escaped due to having spaces in it.
+#         if (m := re.match(r"^\s*(?P<pattern>[^\s]+)\s+(?P<attributes>.*)$", line)):
+#             # Record if there is already a pattern that covers the file.
+#             # Note: The *last* gitattribute line is the one that is used when
+#             # multiple lines match a file.
+#             # TODO(brianlester): Some patterns like [:space:] aren't handled by
+#             # this function. Find a replacement that does?
+#             if fnmatch.fnmatchcase(path, m.group("pattern")):
+#                 previous_attributes = m.group("attributes")
+#                 previous_match = i
+#         new_gitattributes.append(line)
+#     if previous_attributes:
+#         # If the previous match set overlapping attributes, error out.
+#         for attr in ("filter", "diff", "merge"):
+#             if attr not in previous_attributes:
+#                 continue
+#             if attr in previous_attributes and not is_attribute_active(previous_attributes, attr, "theta"):
+#                 raise ValueError()
+#         # Add the new attributes
+#         else:
+#             new_attributes[previous_match] = f"{path} {previous_attributes.strip()} {theta_attributes}"
+#     else:
+#         new_attributes.append(f"{path} {theta_attributes}")
+#     return new_gitattributes
 
 
 def get_gitattributes_tracked_patterns(gitattributes_file):
     gitattributes = read_gitattributes(gitattributes_file)
     theta_attributes = [
-        attribute for attribute in gitattributes if "filter=theta" in attribute
+        attr
+        for attr in gitattributes
+        if attr.attributes.get(a) == "theta"
+        for a in ("filter", "diff", "merge")
     ]
+    return [attr.pattern for attr in theta_attributes]
     # TODO: Correctly handle patterns with escaped spaces in them
     patterns = [attribute.split(" ")[0] for attribute in theta_attributes]
     return patterns
 
 
+def is_theta_tracked(path: str, gitattributes: List[GitAttributes]) -> bool:
+    """Check if `path` is tracked by git-theta based on `.gitattributes`.
+
+    Note: The last line that matches in .gitattributes is the active one so
+      start from the end. If the first match (really last) does not have the
+      theta filter active then the file is not tracked by Git-Theta.
+    """
+    for attr in gitattributes[::-1]:
+        if fnmatch.fnmatchcase(path, attr.pattern):
+            return all(
+                attr.attributes.get(a) == "theta" for a in ("filter", "diff", "merge")
+            )
+    return False
+
+
 def add_file(f, repo):
     """
     Add file to git staging area

git_theta/scripts/git_theta.py

@@ -80,12 +80,12 @@ def post_commit(args):
     theta_commits = theta.ThetaCommits(repo)
 
     gitattributes_file = git_utils.get_gitattributes_file(repo)
-    patterns = git_utils.get_gitattributes_tracked_patterns(gitattributes_file)
+    gitattributes = git_utils.read_gitattributes(gitattributes_file)
 
     oids = set()
     commit = repo.commit("HEAD")
     for path in commit.stats.files.keys():
-        if any([fnmatch.fnmatchcase(path, pattern) for pattern in patterns]):
+        if git_utils.is_theta_tracked(path, gitattributes):
             curr_metadata = metadata.Metadata.from_file(commit.tree[path].data_stream)
             prev_metadata = metadata.Metadata.from_commit(repo, path, "HEAD~1")