WIP: Updates for mypy and new libvcs versions

Author: tony

docs/conf.py

@@ -61,7 +61,7 @@
 html_css_files = ["css/custom.css"]
 html_extra_path = ["manifest.json"]
 html_theme = "furo"
-html_theme_path = []
+html_theme_path: list = []
 html_theme_options = {
     "light_logo": "img/vcspull.svg",
     "dark_logo": "img/vcspull-dark.svg",

scripts/generate_gitlab.py

@@ -58,20 +58,20 @@
     print("File %s not accesible" % (config_filename))
     sys.exit(1)
 
-result = requests.get(
+response = requests.get(
     "%s/api/v4/groups/%s/projects" % (gitlab_host, gitlab_namespace),
     params={"include_subgroups": "true", "per_page": "100"},
     headers={"Authorization": "Bearer %s" % (gitlab_token)},
 )
 
-if 200 != result.status_code:
-    print("Error: ", result)
+if 200 != response.status_code:
+    print("Error: ", response)
     sys.exit(1)
 
 path_prefix = os.getcwd()
-config = {}
+config: dict = {}
 
-for group in result.json():
+for group in response.json():
     url_to_repo = group["ssh_url_to_repo"].replace(":", "/")
     namespace_path = group["namespace"]["full_path"]
     reponame = group["path"]

src/vcspull/cli/sync.py

@@ -6,7 +6,8 @@
 import click.shell_completion
 from click.shell_completion import CompletionItem
 
-from libvcs.shortcuts import create_project_from_pip_url
+from libvcs._internal.shortcuts import create_project
+from vcspull.types import ConfigDict
 
 from ..config import filter_repos, find_config_files, load_configs
 
@@ -21,13 +22,13 @@ def get_repo_completions(
         if ctx.params["config"] is None
         else load_configs(files=[ctx.params["config"]])
     )
-    found_repos = []
+    found_repos: list[ConfigDict] = []
     repo_terms = [incomplete]
 
     for repo_term in repo_terms:
         dir, vcs_url, name = None, None, None
         if any(repo_term.startswith(n) for n in ["./", "/", "~", "$HOME"]):
-            dir = repo_term
+            dir = dir
         elif any(repo_term.startswith(n) for n in ["http", "git", "svn", "hg"]):
             vcs_url = repo_term
         else:
@@ -105,9 +106,11 @@ def update_repo(repo_dict):
     repo_dict = deepcopy(repo_dict)
     if "pip_url" not in repo_dict:
         repo_dict["pip_url"] = repo_dict.pop("url")
+    if "url" not in repo_dict:
+        repo_dict["url"] = repo_dict.pop("pip_url")
     repo_dict["progress_callback"] = progress_cb
 
-    r = create_project_from_pip_url(**repo_dict)  # Creates the repo object
+    r = create_project(**repo_dict)  # Creates the repo object
     r.update_repo(set_remotes=True)  # Creates repo if not exists and fetches
 
     return r

src/vcspull/config.py

@@ -13,9 +13,11 @@
 
 import kaptan
 
-from libvcs.projects.git import GitRemote
+from libvcs._internal.types import StrPath
+from libvcs.sync.git import GitRemote
 
 from . import exc
+from .types import ConfigDict
 from .util import get_config_dir, update_dict
 
 log = logging.getLogger(__name__)
@@ -45,7 +47,7 @@ def expand_dir(
     return _dir
 
 
-def extract_repos(config: dict, cwd=pathlib.Path.cwd()) -> list[dict]:
+def extract_repos(config: dict, cwd=pathlib.Path.cwd()) -> list[ConfigDict]:
     """Return expanded configuration.
 
     end-user configuration permit inline configuration shortcuts, expand to
@@ -62,11 +64,11 @@ def extract_repos(config: dict, cwd=pathlib.Path.cwd()) -> list[dict]:
     -------
     list : List of normalized repository information
     """
-    configs = []
+    configs: list[ConfigDict] = []
     for directory, repos in config.items():
         for repo, repo_data in repos.items():
 
-            conf = {}
+            conf: ConfigDict = {}
 
             """
             repo_name: http://myrepo.com/repo.git
@@ -94,18 +96,26 @@ def extract_repos(config: dict, cwd=pathlib.Path.cwd()) -> list[dict]:
             if "parent_dir" not in conf:
                 conf["parent_dir"] = expand_dir(directory, cwd=cwd)
 
-            # repo_dir -> dir in libvcs 0.12.0b25
-            if "repo_dir" in conf and "dir" not in conf:
-                conf["dir"] = conf.pop("repo_dir")
-
             if "dir" not in conf:
-                conf["dir"] = expand_dir(conf["parent_dir"] / conf["name"], cwd)
+                conf["dir"] = expand_dir(
+                    pathlib.Path(conf["parent_dir"]) / conf["name"], cwd
+                )
 
             if "remotes" in conf:
                 for remote_name, url in conf["remotes"].items():
-                    conf["remotes"][remote_name] = GitRemote(
-                        name=remote_name, fetch_url=url, push_url=url
-                    )
+                    if isinstance(url, GitRemote):
+                        continue
+                    if isinstance(url, str):
+                        conf["remotes"][remote_name] = GitRemote(
+                            name=remote_name, fetch_url=url, push_url=url
+                        )
+                    elif isinstance(url, dict):
+                        assert "push_url" in url
+                        assert "fetch_url" in url
+                        conf["remotes"][remote_name] = GitRemote(
+                            name=remote_name, **url
+                        )
+
             configs.append(conf)
 
     return configs
@@ -192,12 +202,12 @@ def find_config_files(
                     configs.extend(find_config_files(path, match, f))
             else:
                 match = f"{match}.{filetype}"
-                configs = path.glob(match)
+                configs = list(path.glob(match))
 
     return configs
 
 
-def load_configs(files: list[Union[str, pathlib.Path]], cwd=pathlib.Path.cwd()):
+def load_configs(files: list[StrPath], cwd=pathlib.Path.cwd()):
     """Return repos from a list of files.
 
     Parameters
@@ -216,10 +226,11 @@ def load_configs(files: list[Union[str, pathlib.Path]], cwd=pathlib.Path.cwd()):
     ----
     Validate scheme, check for duplicate destinations, VCS urls
     """
-    repos = []
+    repos: list[ConfigDict] = []
     for file in files:
         if isinstance(file, str):
             file = pathlib.Path(file)
+        assert isinstance(file, pathlib.Path)
         ext = file.suffix.lstrip(".")
         conf = kaptan.Kaptan(handler=ext).import_config(str(file))
         newrepos = extract_repos(conf.export("dict"), cwd=cwd)
@@ -230,51 +241,49 @@ def load_configs(files: list[Union[str, pathlib.Path]], cwd=pathlib.Path.cwd()):
 
         dupes = detect_duplicate_repos(repos, newrepos)
 
-        if dupes:
+        if len(dupes) > 0:
             msg = ("repos with same path + different VCS detected!", dupes)
             raise exc.VCSPullException(msg)
         repos.extend(newrepos)
 
     return repos
 
 
-def detect_duplicate_repos(repos1: list[dict], repos2: list[dict]):
+ConfigDictTuple = tuple[ConfigDict, ConfigDict]
+
+
+def detect_duplicate_repos(
+    config1: list[ConfigDict], config2: list[ConfigDict]
+) -> list[ConfigDictTuple]:
     """Return duplicate repos dict if repo_dir same and vcs different.
 
     Parameters
     ----------
-    repos1 : dict
-        list of repo expanded dicts
+    config1 : list[ConfigDict]
 
-    repos2 : dict
-        list of repo expanded dicts
+    config2 : list[ConfigDict]
 
     Returns
     -------
-    list of dict, or None
-        Duplicate repos
+    list[ConfigDictTuple]
+        List of duplicate tuples
     """
-    dupes = []
-    path_dupe_repos = []
-
-    curpaths = [r["dir"] for r in repos1]
-    newpaths = [r["dir"] for r in repos2]
-    path_duplicates = list(set(curpaths).intersection(newpaths))
+    if not config1:
+        return []
 
-    if not path_duplicates:
-        return None
+    dupes: list[ConfigDictTuple] = []
 
-    path_dupe_repos.extend(
-        [r for r in repos2 if any(r["dir"] == p for p in path_duplicates)]
-    )
+    repo_dirs = {
+        pathlib.Path(repo["parent_dir"]) / repo["name"]: repo for repo in config1
+    }
+    repo_dirs_2 = {
+        pathlib.Path(repo["parent_dir"]) / repo["name"]: repo for repo in config2
+    }
 
-    if not path_dupe_repos:
-        return None
+    for repo_dir, repo in repo_dirs.items():
+        if repo_dir in repo_dirs_2:
+            dupes.append((repo, repo_dirs_2[repo_dir]))
 
-    for n in path_dupe_repos:
-        currepo = next((r for r in repos1 if r["dir"] == n["dir"]), None)
-        if n["url"] != currepo["url"]:
-            dupes += (n, currepo)
     return dupes
 
 
@@ -304,11 +313,11 @@ def in_dir(config_dir=None, extensions: list[str] = [".yml", ".yaml", ".json"]):
 
 
 def filter_repos(
-    config: dict,
-    dir: Union[pathlib.Path, None] = None,
+    config: list[ConfigDict],
+    dir: Union[pathlib.Path, Literal["*"], None] = None,
     vcs_url: Union[str, None] = None,
     name: Union[str, None] = None,
-):
+) -> list[ConfigDict]:
     """Return a :py:obj:`list` list of repos from (expanded) config file.
 
     dir, vcs_url and name all support fnmatch.
@@ -329,23 +338,31 @@ def filter_repos(
     list :
         Repos
     """
-    repo_list = []
+    repo_list: list[ConfigDict] = []
 
     if dir:
-        repo_list.extend([r for r in config if fnmatch.fnmatch(r["parent_dir"], dir)])
+        repo_list.extend(
+            [r for r in config if fnmatch.fnmatch(str(r["parent_dir"]), str(dir))]
+        )
 
     if vcs_url:
         repo_list.extend(
-            r for r in config if fnmatch.fnmatch(r.get("url", r.get("repo")), vcs_url)
+            r
+            for r in config
+            if fnmatch.fnmatch(str(r.get("url", r.get("repo"))), vcs_url)
         )
 
     if name:
-        repo_list.extend([r for r in config if fnmatch.fnmatch(r.get("name"), name)])
+        repo_list.extend(
+            [r for r in config if fnmatch.fnmatch(str(r.get("name")), name)]
+        )
 
     return repo_list
 
 
-def is_config_file(filename: str, extensions: list[str] = [".yml", ".yaml", ".json"]):
+def is_config_file(
+    filename: str, extensions: Union[list[str], str] = [".yml", ".yaml", ".json"]
+):
     """Return True if file has a valid config file type.
 
     Parameters

src/vcspull/types.py

@@ -0,0 +1,25 @@
+import typing as t
+
+from libvcs._internal.types import StrPath, VCSLiteral
+from libvcs.sync.git import GitSyncRemoteDict
+
+
+class RawConfigDict(t.TypedDict):
+    vcs: VCSLiteral
+    name: str
+    dir: StrPath
+    url: str
+    remotes: GitSyncRemoteDict
+
+
+RawConfigDir = dict[str, RawConfigDict]
+RawConfig = dict[str, RawConfigDir]
+
+
+class ConfigDict(t.TypedDict):
+    vcs: VCSLiteral
+    name: str
+    dir: StrPath
+    parent_dir: StrPath
+    url: str
+    remotes: GitSyncRemoteDict

tests/conftest.py

@@ -2,12 +2,13 @@
 import pathlib
 import shutil
 import textwrap
+import typing as t
 
 import pytest
 
 from libvcs._internal.run import run
-from libvcs.projects.git import GitProject
-from libvcs.shortcuts import create_project_from_pip_url
+from libvcs._internal.shortcuts import create_project
+from libvcs.sync.git import GitSync
 
 
 @pytest.fixture(autouse=True, scope="session")
@@ -67,7 +68,7 @@ def clean():
 
 @pytest.fixture
 def git_repo_kwargs(repos_path: pathlib.Path, git_dummy_repo_dir):
-    """Return kwargs for :func:`create_project_from_pip_url`."""
+    """Return kwargs for :func:`create_project`."""
     return {
         "url": "git+file://" + git_dummy_repo_dir,
         "parent_dir": str(repos_path),
@@ -76,16 +77,26 @@ def git_repo_kwargs(repos_path: pathlib.Path, git_dummy_repo_dir):
 
 
 @pytest.fixture
-def git_repo(git_repo_kwargs) -> GitProject:
+def git_repo(git_repo_kwargs) -> GitSync:
     """Create an git repository for tests. Return repo."""
-    repo = create_project_from_pip_url(**git_repo_kwargs)
+    repo = create_project(vcs="git", **git_repo_kwargs)
     repo.obtain(quiet=True)
     return repo
 
 
+class DummyRepoProtocol(t.Protocol):
+    """Callback for repo fixture factory."""
+
+    def __call__(self, repo_name: str, testfile_filename: str = ...) -> str:
+        """Callback signature for subprocess communication."""
+        ...
+
+
 @pytest.fixture
-def create_git_dummy_repo(repos_path: pathlib.Path) -> pathlib.Path:
-    def fn(repo_name, testfile_filename="testfile.test"):
+def create_git_dummy_repo(
+    repos_path: pathlib.Path,
+) -> t.Generator[DummyRepoProtocol, None, None]:
+    def fn(repo_name: str, testfile_filename: str = "testfile.test"):
         repo_path = str(repos_path / repo_name)
 
         run(["git", "init", repo_name], cwd=str(repos_path))
@@ -100,7 +111,9 @@ def fn(repo_name, testfile_filename="testfile.test"):
 
 
 @pytest.fixture
-def git_dummy_repo_dir(repos_path: pathlib.Path, create_git_dummy_repo):
+def git_dummy_repo_dir(
+    repos_path: pathlib.Path, create_git_dummy_repo: DummyRepoProtocol
+):
     """Create a git repo with 1 commit, used as a remote."""
     return create_git_dummy_repo("dummyrepo")