Type safety for homely._vcs

homely/_utils.py

@@ -268,10 +268,10 @@ def writejson(self) -> None:
 
 class RepoListEntry(TypedDict):
     repoid: str
-    localrepo: str
+    localrepo: dict[str, str | bool]
     localpath: str
     canonicalpath: NotRequired[str]
-    canonicalrepo: NotRequired[str]
+    canonicalrepo: NotRequired[dict[str, str | bool]]
 
 
 class RepoListConfig(JsonConfig[list[RepoListEntry]]):

homely/_vcs/__init__.py

@@ -1,14 +1,18 @@
 import os
+from enum import Enum
+from typing import Iterable, Optional
 
 from homely._errors import NotARepo
 
 _handlers = None
 
-HANDLER_GIT_v1 = "vcs:git"
-HANDLER_TESTHANDLER_v1 = "vcs:testhandler"
 
+class RepoType(Enum):
+    HANDLER_GIT_v1 = "vcs:git"
+    HANDLER_TESTHANDLER_v1 = "vcs:testhandler"
 
-def _gethandlers():
+
+def _gethandlers() -> Iterable[type["Repo"]]:
     global _handlers
     import homely._vcs.git
     import homely._vcs.testhandler
@@ -19,15 +23,15 @@ def _gethandlers():
     return _handlers
 
 
-def getrepohandler(repo_path):
+def getrepohandler(repo_path: str) -> "Repo":
     for class_ in _handlers or _gethandlers():
         repo = class_.frompath(repo_path)
         if repo is not None:
             return repo
     raise NotARepo(repo_path)
 
 
-def fromdict(row):
+def fromdict(row: dict[str, str | bool]) -> "Repo":
     for class_ in _handlers or _gethandlers():
         obj = class_.fromdict(row)
         if obj is not None:
@@ -39,43 +43,45 @@ class Repo:
     """
     Base class for VCS handlers
     """
-    def __init__(self,
-                 repo_path,
-                 isremote,
-                 iscanonical,
-                 suggestedlocal,
-                 canonical=None):
+    type_: RepoType
+
+    def __init__(
+        self,
+        repo_path: str,
+        isremote: bool,
+        iscanonical: bool,
+        suggestedlocal: Optional[str],
+        canonical: Optional[str] = None,
+    ):
         self.isremote = isremote
         self.iscanonical = iscanonical
         self.repo_path = repo_path
         if suggestedlocal is None:
             suggestedlocal = os.path.basename(repo_path)
         self.suggestedlocal = suggestedlocal
         self._canonical = canonical
-        super(Repo, self).__init__()
-        assert self.type in (HANDLER_GIT_v1, HANDLER_TESTHANDLER_v1)
 
     @classmethod
-    def frompath(class_, repo_path):
+    def frompath(class_, repo_path: str) -> Optional["Repo"]:
         raise Exception(
             "%s.%s needs to implement @classmethod .frompath(repo_path)" % (
                 class_.__module__, class_.__name__))
 
     @classmethod
-    def shortid(class_, repoid):
+    def shortid(class_, repoid: str) -> str:
         raise Exception(
             "%s.%s needs to implement @staticmethod .shortid(repoid)" % (
                 class_.__module__, class_.__name__))
 
-    def getrepoid(self):
+    def getrepoid(self) -> str:
         """
         Get a unique id for the repo. For example, the first commit hash.
         """
         raise Exception(
             "%s.%s needs to implement .getrepoid()" % (
                 self.__class__.__module__, self.__class__.__name__))
 
-    def clonetopath(self, dest):
+    def clonetopath(self, dest: str) -> None:
         """
         Clone the repo at <self.pushablepath> into <dest>
         Note that if self.pushablepath is None, then self.path will be used
@@ -85,30 +91,38 @@ def clonetopath(self, dest):
             "%s.%s needs to implement @classmethod .clonetopath(dest)" % (
                 self.__class__.__module__, self.__class__.__name__))
 
-    def isdirty(self):
+    def isdirty(self) -> bool:
         raise Exception(
             "%s.%s needs to implement .isdirty()" % (
                 self.__class__.__module__, self.__class__.__name__))
 
-    def pullchanges(self):
+    def pullchanges(self) -> None:
         raise Exception(
             "%s.%s needs to implement .pullchanges()" % (
                 self.__class__.__module__, self.__class__.__name__))
 
-    def asdict(self):
+    def asdict(self) -> dict[str, str | bool]:
         return dict(
-            type=self.type,
+            type=self.type_.value,
             repo_path=self.repo_path,
             isremote=self.isremote,
             iscanonical=self.iscanonical,
             suggestedlocal=self.suggestedlocal,
         )
 
     @classmethod
-    def fromdict(class_, row):
-        if row["type"] == class_.type:
-            return class_(row["repo_path"],
-                          row["isremote"],
-                          row["iscanonical"],
-                          row["suggestedlocal"],
-                          )
+    def fromdict(class_, row: dict[str, str | bool]) -> Optional["Repo"]:
+        if row["type"] != class_.type_.value:
+            return None
+
+        assert isinstance(row["repo_path"], str)
+        assert isinstance(row["suggestedlocal"], str)
+        assert isinstance(row["isremote"], bool)
+        assert isinstance(row["iscanonical"], bool)
+
+        return class_(
+            row["repo_path"],
+            row["isremote"],
+            row["iscanonical"],
+            row["suggestedlocal"],
+        )

homely/_vcs/git.py

@@ -1,5 +1,6 @@
 import os
 import re
+from typing import Optional
 
 import homely._vcs
 from homely._errors import ConnectionError, RepoError, RepoHasNoCommitsError
@@ -8,11 +9,11 @@
 
 
 class Repo(homely._vcs.Repo):
-    type = homely._vcs.HANDLER_GIT_v1
+    type_ = homely._vcs.RepoType.HANDLER_GIT_v1
     pulldesc = 'git pull'
 
     @classmethod
-    def _from_parts(class_, repo_path, user, domain, name):
+    def _from_parts(class_, repo_path: str, user: str, domain: str, name: str) -> "Repo":
         if name.endswith('.git'):
             name = name[0:-4]
 
@@ -27,15 +28,17 @@ def _from_parts(class_, repo_path, user, domain, name):
         )
 
     @classmethod
-    def frompath(class_, repo_path):
+    def frompath(class_, repo_path: str) -> Optional["Repo"]:
         if os.path.isdir(repo_path):
             if not os.path.isdir(os.path.join(repo_path, '.git')):
-                return
+                return None
+
             return class_(_expandpath(repo_path),
                           isremote=False,
                           iscanonical=False,
                           suggestedlocal=None
                           )
+
         if (repo_path.startswith('ssh://') or
                 repo_path.startswith('https://') or
                 repo_path.startswith('git@')):
@@ -58,7 +61,9 @@ def frompath(class_, repo_path):
                           iscanonical=False,
                           suggestedlocal=None)
 
-    def pullchanges(self):
+        return None
+
+    def pullchanges(self) -> None:
         assert not self.isremote
         cmd = ['git', 'pull']
         code, _, err = execute(cmd,
@@ -69,23 +74,25 @@ def pullchanges(self):
             return
 
         assert code == 1
+        assert isinstance(err, bytes)  # TODO: replace this with a type check
         needle = b'fatal: Could not read from remote repository.'
         if needle in err:
             raise ConnectionError()
 
-        raise SystemError("Unexpected output from 'git pull': {}".format(err))
+        raise SystemError(f"Unexpected output from 'git pull': {err!r}")
 
-    def clonetopath(self, dest):
+    def clonetopath(self, dest: str) -> None:
         origin = self.repo_path
         execute(['git', 'clone', origin, dest])
 
-    def getrepoid(self):
+    def getrepoid(self) -> str:
         assert not self.isremote
         cmd = ['git', 'rev-list', '--max-parents=0', 'HEAD']
         returncode, stdout = run(cmd,
                                  cwd=self.repo_path,
                                  stdout=True,
                                  stderr="STDOUT")[:2]
+        assert isinstance(stdout, bytes)  # TODO: replace this with a type check
         if returncode == 0:
             return self._getfirsthash(stdout)
         if returncode != 128:
@@ -115,19 +122,20 @@ def getrepoid(self):
 
         raise SystemError("Unexpected exitcode {}".format(returncode))
 
-    def _getfirsthash(self, stdout):
+    def _getfirsthash(self, stdout: bytes) -> str:
         stripped = stdout.rstrip().decode('utf-8')
         if '\n' in stripped:
             raise RepoError("Git repo has multiple initial commits")
         return stripped
 
     @staticmethod
-    def shortid(repoid):
+    def shortid(repoid: str) -> str:
         return repoid[0:8]
 
-    def isdirty(self):
+    def isdirty(self) -> bool:
         cmd = ['git', 'status', '--porcelain']
         out = execute(cmd, cwd=self.repo_path, stdout=True)[1]
+        assert isinstance(out, bytes)  # TODO: replace this with a type check
         for line in out.split(b'\n'):
             if len(line) and not line.startswith(b'?? '):
                 return True

homely/_vcs/testhandler.py

@@ -13,11 +13,11 @@
 
 
 class Repo(homely._vcs.Repo):
-    type = homely._vcs.HANDLER_TESTHANDLER_v1
+    type_ = homely._vcs.RepoType.HANDLER_TESTHANDLER_v1
     pulldesc = 'fake repo pull'
 
     @classmethod
-    def frompath(class_, repo_path) -> "Optional[homely._vcs.Repo]":
+    def frompath(class_, repo_path: str) -> "Optional[homely._vcs.Repo]":
         if repo_path.startswith(PREFIX):
             dirpart = repo_path[len(PREFIX):]
             return class_(repo_path,
@@ -37,7 +37,7 @@ def frompath(class_, repo_path) -> "Optional[homely._vcs.Repo]":
                       iscanonical=False,
                       suggestedlocal=None)
 
-    def clonetopath(self, dest_path):
+    def clonetopath(self, dest_path: str) -> None:
         assert not os.path.exists(dest_path)
         os.mkdir(dest_path)
         with open(os.path.join(dest_path, ORIGINFILE), 'w') as f:
@@ -47,7 +47,7 @@ def clonetopath(self, dest_path):
             origin = origin[len(PREFIX):]
         self._pull(origin, dest_path)
 
-    def _pull(self, origin, local):
+    def _pull(self, origin: str, local: str) -> None:
         # delete every local file except the special ones
         for thing in os.listdir(local):
             if thing not in (ORIGINFILE, MARKERFILE, DIRTYFILE):
@@ -70,23 +70,23 @@ def _pull(self, origin, local):
             else:
                 shutil.copy2(src, dst)
 
-    def pullchanges(self):
+    def pullchanges(self) -> None:
         assert not self.isdirty()
         with open(os.path.join(self.repo_path, ORIGINFILE), 'r') as f:
             origin = f.read().strip()
             if origin.startswith(PREFIX):
                 origin = origin[len(PREFIX):]
         self._pull(origin, self.repo_path)
 
-    def getrepoid(self):
+    def getrepoid(self) -> str:
         assert not self.isremote
         with open(os.path.join(self.repo_path, MARKERFILE), 'r') as f:
             return f.read().strip()
 
     @staticmethod
-    def shortid(repoid):
+    def shortid(repoid: str) -> str:
         return repoid[0:5]
 
-    def isdirty(self):
+    def isdirty(self) -> bool:
         assert not self.isremote
         return os.path.exists(os.path.join(self.repo_path, DIRTYFILE))

pyproject.toml

@@ -70,9 +70,6 @@ module = [
     "homely._test.system",
     "homely._test",
     "homely._ui",
-    "homely._vcs.git",
-    "homely._vcs.testhandler",
-    "homely._vcs",
     "homely.files",
     "homely.general",
     "homely.install",