Merge pull request #1285 from Yobmod/main

Finish initial typing of Index and Submodule

Author: Byron

git/cmd.py

@@ -831,7 +831,7 @@ def execute(self,
         except cmd_not_found_exception as err:
             raise GitCommandNotFound(redacted_command, err) from err
         else:
-            proc = cast(Popen, proc)
+            # replace with a typeguard for Popen[bytes]?
             proc.stdout = cast(BinaryIO, proc.stdout)
             proc.stderr = cast(BinaryIO, proc.stderr)
 

git/config.py

@@ -31,12 +31,14 @@
 
 # typing-------------------------------------------------------
 
-from typing import Any, Callable, IO, List, Dict, Sequence, TYPE_CHECKING, Tuple, Union, cast, overload
+from typing import (Any, Callable, IO, List, Dict, Sequence,
+                    TYPE_CHECKING, Tuple, Union, cast, overload)
 
-from git.types import Literal, Lit_config_levels, PathLike, TBD
+from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, TBD, assert_never, is_config_level
 
 if TYPE_CHECKING:
     from git.repo.base import Repo
+    from io import BytesIO
 
 # -------------------------------------------------------------
 
@@ -48,8 +50,10 @@
 
 # invariants
 # represents the configuration level of a configuration file
-CONFIG_LEVELS = ("system", "user", "global", "repository"
-                 )  # type: Tuple[Literal['system'], Literal['user'], Literal['global'], Literal['repository']]
+
+
+CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository")
+
 
 # Section pattern to detect conditional includes.
 # https://git-scm.com/docs/git-config#_conditional_includes
@@ -229,8 +233,9 @@ def get_config_path(config_level: Lit_config_levels) -> str:
         return osp.normpath(osp.expanduser("~/.gitconfig"))
     elif config_level == "repository":
         raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path")
-
-    raise ValueError("Invalid configuration level: %r" % config_level)
+    else:
+        # Should not reach here. Will raise ValueError if does. Static typing will warn missing elifs
+        assert_never(config_level, ValueError(f"Invalid configuration level: {config_level!r}"))
 
 
 class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, object)):  # type: ignore ## mypy does not understand dynamic class creation # noqa: E501
@@ -271,7 +276,7 @@ class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, obje
     # list of RawConfigParser methods able to change the instance
     _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
 
-    def __init__(self, file_or_files: Union[None, PathLike, IO, Sequence[Union[PathLike, IO]]] = None,
+    def __init__(self, file_or_files: Union[None, PathLike, 'BytesIO', Sequence[Union[PathLike, 'BytesIO']]] = None,
                  read_only: bool = True, merge_includes: bool = True,
                  config_level: Union[Lit_config_levels, None] = None,
                  repo: Union['Repo', None] = None) -> None:
@@ -300,13 +305,13 @@ def __init__(self, file_or_files: Union[None, PathLike, IO, Sequence[Union[PathL
             self._proxies = self._dict()
 
         if file_or_files is not None:
-            self._file_or_files = file_or_files  # type: Union[PathLike, IO, Sequence[Union[PathLike, IO]]]
+            self._file_or_files: Union[PathLike, 'BytesIO', Sequence[Union[PathLike, 'BytesIO']]] = file_or_files
         else:
             if config_level is None:
                 if read_only:
-                    self._file_or_files = [get_config_path(f)  # type: ignore
-                                           for f in CONFIG_LEVELS    # Can type f properly when 3.5 dropped
-                                           if f != 'repository']
+                    self._file_or_files = [get_config_path(f)
+                                           for f in CONFIG_LEVELS
+                                           if is_config_level(f) and f != 'repository']
                 else:
                     raise ValueError("No configuration level or configuration files specified")
             else:
@@ -323,15 +328,13 @@ def __init__(self, file_or_files: Union[None, PathLike, IO, Sequence[Union[PathL
     def _acquire_lock(self) -> None:
         if not self._read_only:
             if not self._lock:
-                if isinstance(self._file_or_files, (tuple, list)):
-                    raise ValueError(
-                        "Write-ConfigParsers can operate on a single file only, multiple files have been passed")
-                # END single file check
-
                 if isinstance(self._file_or_files, (str, os.PathLike)):
                     file_or_files = self._file_or_files
+                elif isinstance(self._file_or_files, (tuple, list, Sequence)):
+                    raise ValueError(
+                        "Write-ConfigParsers can operate on a single file only, multiple files have been passed")
                 else:
-                    file_or_files = cast(IO, self._file_or_files).name
+                    file_or_files = self._file_or_files.name
 
                 # END get filename from handle/stream
                 # initialize lock base - we want to write
@@ -649,7 +652,7 @@ def write(self) -> None:
             a file lock"""
         self._assure_writable("write")
         if not self._dirty:
-            return
+            return None
 
         if isinstance(self._file_or_files, (list, tuple)):
             raise AssertionError("Cannot write back if there is not exactly a single file to write to, have %i files"
@@ -665,7 +668,7 @@ def write(self) -> None:
         fp = self._file_or_files
 
         # we have a physical file on disk, so get a lock
-        is_file_lock = isinstance(fp, (str, IOBase))  # can't use Pathlike until 3.5 dropped
+        is_file_lock = isinstance(fp, (str, os.PathLike, IOBase))  # can't use Pathlike until 3.5 dropped
         if is_file_lock and self._lock is not None:  # else raise Error?
             self._lock._obtain_lock()
 
@@ -674,7 +677,7 @@ def write(self) -> None:
             with open(fp, "wb") as fp_open:
                 self._write(fp_open)
         else:
-            fp = cast(IO, fp)
+            fp = cast('BytesIO', fp)
             fp.seek(0)
             # make sure we do not overwrite into an existing file
             if hasattr(fp, 'truncate'):

git/diff.py

@@ -15,19 +15,26 @@
 
 # typing ------------------------------------------------------------------
 
-from typing import Any, Iterator, List, Match, Optional, Tuple, Type, Union, TYPE_CHECKING
-from git.types import PathLike, TBD, Literal
+from typing import Any, Iterator, List, Match, Optional, Tuple, Type, TypeVar, Union, TYPE_CHECKING
+from git.types import PathLike, TBD, Literal, TypeGuard
 
 if TYPE_CHECKING:
     from .objects.tree import Tree
+    from .objects import Commit
     from git.repo.base import Repo
-
+    from git.objects.base import IndexObject
     from subprocess import Popen
 
-Lit_change_type = Literal['A', 'D', 'M', 'R', 'T']
+Lit_change_type = Literal['A', 'D', 'C', 'M', 'R', 'T', 'U']
+
+
+def is_change_type(inp: str) -> TypeGuard[Lit_change_type]:
+    # return True
+    return inp in ['A', 'D', 'C', 'M', 'R', 'T', 'U']
 
 # ------------------------------------------------------------------------
 
+
 __all__ = ('Diffable', 'DiffIndex', 'Diff', 'NULL_TREE')
 
 # Special object to compare against the empty tree in diffs
@@ -75,15 +82,16 @@ class Diffable(object):
     class Index(object):
         pass
 
-    def _process_diff_args(self, args: List[Union[str, 'Diffable', object]]) -> List[Union[str, 'Diffable', object]]:
+    def _process_diff_args(self, args: List[Union[str, 'Diffable', Type['Diffable.Index'], object]]
+                           ) -> List[Union[str, 'Diffable', Type['Diffable.Index'], object]]:
         """
         :return:
             possibly altered version of the given args list.
             Method is called right before git command execution.
             Subclasses can use it to alter the behaviour of the superclass"""
         return args
 
-    def diff(self, other: Union[Type[Index], Type['Tree'], object, None, str] = Index,
+    def diff(self, other: Union[Type['Index'], 'Tree', 'Commit', None, str, object] = Index,
              paths: Union[PathLike, List[PathLike], Tuple[PathLike, ...], None] = None,
              create_patch: bool = False, **kwargs: Any) -> 'DiffIndex':
         """Creates diffs between two items being trees, trees and index or an
@@ -116,7 +124,7 @@ def diff(self, other: Union[Type[Index], Type['Tree'], object, None, str] = Inde
         :note:
             On a bare repository, 'other' needs to be provided as Index or as
             as Tree/Commit, or a git command error will occur"""
-        args = []  # type: List[Union[str, Diffable, object]]
+        args: List[Union[PathLike, Diffable, Type['Diffable.Index'], object]] = []
         args.append("--abbrev=40")        # we need full shas
         args.append("--full-index")       # get full index paths, not only filenames
 
@@ -134,8 +142,8 @@ def diff(self, other: Union[Type[Index], Type['Tree'], object, None, str] = Inde
         if paths is not None and not isinstance(paths, (tuple, list)):
             paths = [paths]
 
-        if hasattr(self, 'repo'):  # else raise Error?
-            self.repo = self.repo  # type: 'Repo'
+        if hasattr(self, 'Has_Repo'):
+            self.repo: Repo = self.repo
 
         diff_cmd = self.repo.git.diff
         if other is self.Index:
@@ -169,7 +177,10 @@ def diff(self, other: Union[Type[Index], Type['Tree'], object, None, str] = Inde
         return index
 
 
-class DiffIndex(list):
+T_Diff = TypeVar('T_Diff', bound='Diff')
+
+
+class DiffIndex(List[T_Diff]):
 
     """Implements an Index for diffs, allowing a list of Diffs to be queried by
     the diff properties.
@@ -183,7 +194,7 @@ class DiffIndex(list):
     # T = Changed in the type
     change_type = ("A", "C", "D", "R", "M", "T")
 
-    def iter_change_type(self, change_type: Lit_change_type) -> Iterator['Diff']:
+    def iter_change_type(self, change_type: Lit_change_type) -> Iterator[T_Diff]:
         """
         :return:
             iterator yielding Diff instances that match the given change_type
@@ -200,19 +211,19 @@ def iter_change_type(self, change_type: Lit_change_type) -> Iterator['Diff']:
         if change_type not in self.change_type:
             raise ValueError("Invalid change type: %s" % change_type)
 
-        for diff in self:   # type: 'Diff'
-            if diff.change_type == change_type:
-                yield diff
-            elif change_type == "A" and diff.new_file:
-                yield diff
-            elif change_type == "D" and diff.deleted_file:
-                yield diff
-            elif change_type == "C" and diff.copied_file:
-                yield diff
-            elif change_type == "R" and diff.renamed:
-                yield diff
-            elif change_type == "M" and diff.a_blob and diff.b_blob and diff.a_blob != diff.b_blob:
-                yield diff
+        for diffidx in self:
+            if diffidx.change_type == change_type:
+                yield diffidx
+            elif change_type == "A" and diffidx.new_file:
+                yield diffidx
+            elif change_type == "D" and diffidx.deleted_file:
+                yield diffidx
+            elif change_type == "C" and diffidx.copied_file:
+                yield diffidx
+            elif change_type == "R" and diffidx.renamed:
+                yield diffidx
+            elif change_type == "M" and diffidx.a_blob and diffidx.b_blob and diffidx.a_blob != diffidx.b_blob:
+                yield diffidx
         # END for each diff
 
 
@@ -281,7 +292,7 @@ def __init__(self, repo: 'Repo',
                  a_mode: Union[bytes, str, None], b_mode: Union[bytes, str, None],
                  new_file: bool, deleted_file: bool, copied_file: bool,
                  raw_rename_from: Optional[bytes], raw_rename_to: Optional[bytes],
-                 diff: Union[str, bytes, None], change_type: Optional[str], score: Optional[int]) -> None:
+                 diff: Union[str, bytes, None], change_type: Optional[Lit_change_type], score: Optional[int]) -> None:
 
         assert a_rawpath is None or isinstance(a_rawpath, bytes)
         assert b_rawpath is None or isinstance(b_rawpath, bytes)
@@ -300,19 +311,21 @@ def __init__(self, repo: 'Repo',
                         repo = submodule.module()
                     break
 
+        self.a_blob: Union['IndexObject', None]
         if a_blob_id is None or a_blob_id == self.NULL_HEX_SHA:
             self.a_blob = None
         else:
             self.a_blob = Blob(repo, hex_to_bin(a_blob_id), mode=self.a_mode, path=self.a_path)
 
+        self.b_blob: Union['IndexObject', None]
         if b_blob_id is None or b_blob_id == self.NULL_HEX_SHA:
             self.b_blob = None
         else:
             self.b_blob = Blob(repo, hex_to_bin(b_blob_id), mode=self.b_mode, path=self.b_path)
 
-        self.new_file = new_file
-        self.deleted_file = deleted_file
-        self.copied_file = copied_file
+        self.new_file: bool = new_file
+        self.deleted_file: bool = deleted_file
+        self.copied_file: bool = copied_file
 
         # be clear and use None instead of empty strings
         assert raw_rename_from is None or isinstance(raw_rename_from, bytes)
@@ -321,7 +334,7 @@ def __init__(self, repo: 'Repo',
         self.raw_rename_to = raw_rename_to or None
 
         self.diff = diff
-        self.change_type = change_type
+        self.change_type: Union[Lit_change_type, None] = change_type
         self.score = score
 
     def __eq__(self, other: object) -> bool:
@@ -386,36 +399,36 @@ def __str__(self) -> str:
         # end
         return res
 
-    @property
+    @ property
     def a_path(self) -> Optional[str]:
         return self.a_rawpath.decode(defenc, 'replace') if self.a_rawpath else None
 
-    @property
+    @ property
     def b_path(self) -> Optional[str]:
         return self.b_rawpath.decode(defenc, 'replace') if self.b_rawpath else None
 
-    @property
+    @ property
     def rename_from(self) -> Optional[str]:
         return self.raw_rename_from.decode(defenc, 'replace') if self.raw_rename_from else None
 
-    @property
+    @ property
     def rename_to(self) -> Optional[str]:
         return self.raw_rename_to.decode(defenc, 'replace') if self.raw_rename_to else None
 
-    @property
+    @ property
     def renamed(self) -> bool:
         """:returns: True if the blob of our diff has been renamed
         :note: This property is deprecated, please use ``renamed_file`` instead.
         """
         return self.renamed_file
 
-    @property
+    @ property
     def renamed_file(self) -> bool:
         """:returns: True if the blob of our diff has been renamed
         """
         return self.rename_from != self.rename_to
 
-    @classmethod
+    @ classmethod
     def _pick_best_path(cls, path_match: bytes, rename_match: bytes, path_fallback_match: bytes) -> Optional[bytes]:
         if path_match:
             return decode_path(path_match)
@@ -428,7 +441,7 @@ def _pick_best_path(cls, path_match: bytes, rename_match: bytes, path_fallback_m
 
         return None
 
-    @classmethod
+    @ classmethod
     def _index_from_patch_format(cls, repo: 'Repo', proc: TBD) -> DiffIndex:
         """Create a new DiffIndex from the given text which must be in patch format
         :param repo: is the repository we are operating on - it is required
@@ -441,7 +454,7 @@ def _index_from_patch_format(cls, repo: 'Repo', proc: TBD) -> DiffIndex:
 
         # for now, we have to bake the stream
         text = b''.join(text_list)
-        index = DiffIndex()
+        index: 'DiffIndex' = DiffIndex()
         previous_header = None
         header = None
         a_path, b_path = None, None  # for mypy
@@ -491,19 +504,21 @@ def _index_from_patch_format(cls, repo: 'Repo', proc: TBD) -> DiffIndex:
 
         return index
 
-    @staticmethod
+    @ staticmethod
     def _handle_diff_line(lines_bytes: bytes, repo: 'Repo', index: DiffIndex) -> None:
         lines = lines_bytes.decode(defenc)
 
         for line in lines.split(':')[1:]:
             meta, _, path = line.partition('\x00')
             path = path.rstrip('\x00')
-            a_blob_id, b_blob_id = None, None  # Type: Optional[str]
+            a_blob_id: Optional[str]
+            b_blob_id: Optional[str]
             old_mode, new_mode, a_blob_id, b_blob_id, _change_type = meta.split(None, 4)
             # Change type can be R100
             # R: status letter
             # 100: score (in case of copy and rename)
-            change_type = _change_type[0]
+            assert is_change_type(_change_type[0]), f"Unexpected value for change_type received: {_change_type[0]}"
+            change_type: Lit_change_type = _change_type[0]
             score_str = ''.join(_change_type[1:])
             score = int(score_str) if score_str.isdigit() else None
             path = path.strip()
@@ -543,14 +558,14 @@ def _handle_diff_line(lines_bytes: bytes, repo: 'Repo', index: DiffIndex) -> Non
                         '', change_type, score)
             index.append(diff)
 
-    @classmethod
+    @ classmethod
     def _index_from_raw_format(cls, repo: 'Repo', proc: 'Popen') -> 'DiffIndex':
         """Create a new DiffIndex from the given stream which must be in raw format.
         :return: git.DiffIndex"""
         # handles
         # :100644 100644 687099101... 37c5e30c8... M    .gitignore
 
-        index = DiffIndex()
+        index: 'DiffIndex' = DiffIndex()
         handle_process_output(proc, lambda byt: cls._handle_diff_line(byt, repo, index),
                               None, finalize_process, decode_streams=False)