diff --git a/.tmuxp.yaml b/.tmuxp.yaml index b406f94fc..df20071e1 100644 --- a/.tmuxp.yaml +++ b/.tmuxp.yaml @@ -11,6 +11,7 @@ windows: panes: - focus: true - pane + - make watch_mypy - make watch_test - window_name: docs layout: main-horizontal diff --git a/CHANGES b/CHANGES index 545dd0b5d..57d8a9f5a 100644 --- a/CHANGES +++ b/CHANGES @@ -21,6 +21,11 @@ $ pipx install --suffix=@next 'vcspull' --pip-args '\--pre' --force + +### Internal + +- mypy: Add `--strict` typings (#386) + ## vcspull v1.18.0 (2022-10-31) ### Python 3.11 support (#409) @@ -218,7 +223,13 @@ $ pipx install --suffix=@next 'vcspull' --pip-args '\--pre' --force - Move to `src/` directory structure (#382) - libvcs: Update to 0.17.x (#373) -- Basic mypy annotations (#373) +- mypy: + + - Basic mypy annotations (#373) + - [`mypy --strict`] compliant (#386) + + [`mypy --strict`]: https://mypy.readthedocs.io/en/stable/command_line.html#cmdoption-mypy-strict + - Remove `.pre-commit-config.yaml`: Let's not automate what the contributor could / should do themselves. - Add [flake8-bugbear](https://github.com/PyCQA/flake8-bugbear) (#379) diff --git a/conftest.py b/conftest.py index 7edd34fe6..fa3932c9d 100644 --- a/conftest.py +++ b/conftest.py @@ -34,18 +34,20 @@ def cwd_default(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path) -> None @pytest.fixture(autouse=True, scope="session") @pytest.mark.usefixtures("set_home") -def xdg_config_path(user_path: pathlib.Path): +def xdg_config_path(user_path: pathlib.Path) -> pathlib.Path: p = user_path / ".config" p.mkdir() return p @pytest.fixture(scope="function") -def config_path(xdg_config_path: pathlib.Path, request: pytest.FixtureRequest): +def config_path( + xdg_config_path: pathlib.Path, request: pytest.FixtureRequest +) -> pathlib.Path: conf_path = xdg_config_path / "vcspull" conf_path.mkdir(exist_ok=True) - def clean(): + def clean() -> None: shutil.rmtree(conf_path) request.addfinalizer(clean) @@ -53,17 +55,19 @@ def clean(): @pytest.fixture(autouse=True) -def set_xdg_config_path(monkeypatch: pytest.MonkeyPatch, xdg_config_path: pathlib.Path): +def set_xdg_config_path( + monkeypatch: pytest.MonkeyPatch, xdg_config_path: pathlib.Path +) -> None: monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg_config_path)) @pytest.fixture(scope="function") -def repos_path(user_path: pathlib.Path, request: pytest.FixtureRequest): +def repos_path(user_path: pathlib.Path, request: pytest.FixtureRequest) -> pathlib.Path: """Return temporary directory for repository checkout guaranteed unique.""" dir = user_path / "repos" dir.mkdir(exist_ok=True) - def clean(): + def clean() -> None: shutil.rmtree(dir) request.addfinalizer(clean) diff --git a/docs/conf.py b/docs/conf.py index 02c0c4a69..8ce9f2e57 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -1,8 +1,13 @@ # flake8: noqa E501 +import inspect import os import sys +import typing as t +from os.path import dirname, relpath from pathlib import Path +import vcspull + # Get the project root dir, which is the parent dir of this cwd = Path.cwd() project_root = cwd.parent @@ -60,7 +65,7 @@ html_css_files = ["css/custom.css"] html_extra_path = ["manifest.json"] html_theme = "furo" -html_theme_path: list = [] +html_theme_path: list[str] = [] html_theme_options = { "light_logo": "img/vcspull.svg", "dark_logo": "img/vcspull-dark.svg", @@ -171,13 +176,9 @@ } -def linkcode_resolve(domain, info): # NOQA: C901 - import inspect - import sys - from os.path import dirname, relpath - - import vcspull - +def linkcode_resolve( + domain: str, info: dict[str, str] +) -> t.Union[None, str]: # NOQA: C901 """ Determine the URL corresponding to Python object @@ -210,7 +211,8 @@ def linkcode_resolve(domain, info): # NOQA: C901 except AttributeError: pass else: - obj = unwrap(obj) + if callable(obj): + obj = unwrap(obj) try: fn = inspect.getsourcefile(obj) diff --git a/pyproject.toml b/pyproject.toml index b9f7a5c99..c0437b69a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -143,6 +143,7 @@ files = [ "src", "tests" ] +strict = true [[tool.mypy.overrides]] module = [ diff --git a/scripts/generate_gitlab.py b/scripts/generate_gitlab.py index aa1c3e0ef..bbe93e6b2 100755 --- a/scripts/generate_gitlab.py +++ b/scripts/generate_gitlab.py @@ -7,6 +7,10 @@ import requests import yaml +from libvcs.sync.git import GitRemote +from vcspull.cli.sync import guess_vcs +from vcspull.types import RawConfig + try: gitlab_token = os.environ["GITLAB_TOKEN"] except KeyError: @@ -69,14 +73,14 @@ sys.exit(1) path_prefix = os.getcwd() -config: dict = {} +config: RawConfig = {} for group in response.json(): url_to_repo = group["ssh_url_to_repo"].replace(":", "/") namespace_path = group["namespace"]["full_path"] reponame = group["path"] - path = "%s/%s" % (path_prefix, namespace_path) + path = f"{path_prefix}/{namespace_path}" if path not in config: config[path] = {} @@ -84,9 +88,22 @@ # simplified config not working - https://github.com/vcs-python/vcspull/issues/332 # config[path][reponame] = 'git+ssh://%s' % (url_to_repo) + vcs = guess_vcs(url_to_repo) + if vcs is None: + raise Exception(f"Could not guess VCS for URL: {url_to_repo}") + config[path][reponame] = { - "url": "git+ssh://%s" % (url_to_repo), - "remotes": {"origin": "ssh://%s" % (url_to_repo)}, + "name": reponame, + "dir": path / reponame, + "url": f"git+ssh://{url_to_repo}", + "remotes": { + "origin": GitRemote( + name="origin", + fetch_url=f"ssh://{url_to_repo}", + push_url=f"ssh://{url_to_repo}", + ) + }, + "vcs": vcs, } config_yaml = yaml.dump(config) diff --git a/src/vcspull/_internal/config_reader.py b/src/vcspull/_internal/config_reader.py index 32ccbfe26..23ca22fde 100644 --- a/src/vcspull/_internal/config_reader.py +++ b/src/vcspull/_internal/config_reader.py @@ -22,11 +22,11 @@ class ConfigReader: '{\n "session_name": "my session"\n}' """ - def __init__(self, content: "RawConfigData"): + def __init__(self, content: "RawConfigData") -> None: self.content = content @staticmethod - def _load(format: "FormatLiteral", content: str): + def _load(format: "FormatLiteral", content: str) -> t.Dict[str, t.Any]: """Load raw config data and directly return it. >>> ConfigReader._load("json", '{ "session_name": "my session" }') @@ -36,17 +36,20 @@ def _load(format: "FormatLiteral", content: str): {'session_name': 'my session'} """ if format == "yaml": - return yaml.load( - content, - Loader=yaml.SafeLoader, + return t.cast( + t.Dict[str, t.Any], + yaml.load( + content, + Loader=yaml.SafeLoader, + ), ) elif format == "json": - return json.loads(content) + return t.cast(t.Dict[str, t.Any], json.loads(content)) else: raise NotImplementedError(f"{format} not supported in configuration") @classmethod - def load(cls, format: "FormatLiteral", content: str): + def load(cls, format: "FormatLiteral", content: str) -> "ConfigReader": """Load raw config data into a ConfigReader instance (to dump later). >>> cfg = ConfigReader.load("json", '{ "session_name": "my session" }') @@ -69,7 +72,7 @@ def load(cls, format: "FormatLiteral", content: str): ) @classmethod - def _from_file(cls, path: pathlib.Path): + def _from_file(cls, path: pathlib.Path) -> t.Dict[str, t.Any]: r"""Load data from file path directly to dictionary. **YAML file** @@ -102,7 +105,7 @@ def _from_file(cls, path: pathlib.Path): content = open(path).read() if path.suffix in [".yaml", ".yml"]: - format: FormatLiteral = "yaml" + format: "FormatLiteral" = "yaml" elif path.suffix == ".json": format = "json" else: @@ -114,7 +117,7 @@ def _from_file(cls, path: pathlib.Path): ) @classmethod - def from_file(cls, path: pathlib.Path): + def from_file(cls, path: pathlib.Path) -> "ConfigReader": r"""Load data from file path **YAML file** @@ -169,11 +172,14 @@ def _dump( '{\n "session_name": "my session"\n}' """ if format == "yaml": - return yaml.dump( - content, - indent=2, - default_flow_style=False, - Dumper=yaml.SafeDumper, + return t.cast( + str, + yaml.dump( + content, + indent=2, + default_flow_style=False, + Dumper=yaml.SafeDumper, + ), ) elif format == "json": return json.dumps( diff --git a/src/vcspull/cli/__init__.py b/src/vcspull/cli/__init__.py index 42f526f51..abe4b911c 100644 --- a/src/vcspull/cli/__init__.py +++ b/src/vcspull/cli/__init__.py @@ -7,6 +7,8 @@ import argparse import logging import textwrap +import typing as t +from typing import overload from libvcs.__about__ import __version__ as libvcs_version @@ -30,7 +32,21 @@ ).strip() -def create_parser(return_subparsers: bool = False): +@overload +def create_parser( + return_subparsers: t.Literal[True], +) -> t.Tuple[argparse.ArgumentParser, t.Any]: + ... + + +@overload +def create_parser(return_subparsers: t.Literal[False]) -> argparse.ArgumentParser: + ... + + +def create_parser( + return_subparsers: bool = False, +) -> t.Union[argparse.ArgumentParser, t.Tuple[argparse.ArgumentParser, t.Any]]: parser = argparse.ArgumentParser( prog="vcspull", formatter_class=argparse.RawDescriptionHelpFormatter, @@ -64,9 +80,9 @@ def create_parser(return_subparsers: bool = False): return parser -def cli(args=None): +def cli(_args: t.Optional[t.List[str]] = None) -> None: parser, sync_parser = create_parser(return_subparsers=True) - args = parser.parse_args(args) + args = parser.parse_args(_args) setup_logger(log=log, level=args.log_level.upper()) diff --git a/src/vcspull/cli/sync.py b/src/vcspull/cli/sync.py index 0f9d8966b..d828e4a37 100644 --- a/src/vcspull/cli/sync.py +++ b/src/vcspull/cli/sync.py @@ -1,10 +1,14 @@ import argparse import logging +import pathlib import sys import typing as t from copy import deepcopy +from datetime import datetime from libvcs._internal.shortcuts import create_project +from libvcs._internal.types import VCSLiteral +from libvcs.sync.git import GitSync from libvcs.url import registry as url_tools from ..config import filter_repos, find_config_files, load_configs @@ -12,7 +16,7 @@ log = logging.getLogger(__name__) -def clamp(n, _min, _max): +def clamp(n: int, _min: int, _max: int) -> int: return max(_min, min(n, _max)) @@ -51,8 +55,8 @@ def create_sync_subparser(parser: argparse.ArgumentParser) -> argparse.ArgumentP def sync( - repo_patterns, - config, + repo_patterns: t.List[str], + config: pathlib.Path, exit_on_error: bool, parser: t.Optional[ argparse.ArgumentParser @@ -102,12 +106,28 @@ def sync( raise SystemExit(EXIT_ON_ERROR_MSG) -def progress_cb(output, timestamp): +def progress_cb(output: str, timestamp: datetime) -> None: sys.stdout.write(output) sys.stdout.flush() -def update_repo(repo_dict): +def guess_vcs(url: str) -> t.Optional[VCSLiteral]: + vcs_matches = url_tools.registry.match(url=url, is_explicit=True) + + if len(vcs_matches) == 0: + log.warning(f"No vcs found for {url}") + return None + if len(vcs_matches) > 1: + log.warning(f"No exact matches for {url}") + return None + + return t.cast(VCSLiteral, vcs_matches[0].vcs) + + +def update_repo( + repo_dict: t.Any, + # repo_dict: Dict[str, Union[str, Dict[str, GitRemote], pathlib.Path]] +) -> GitSync: repo_dict = deepcopy(repo_dict) if "pip_url" not in repo_dict: repo_dict["pip_url"] = repo_dict.pop("url") @@ -116,16 +136,16 @@ def update_repo(repo_dict): repo_dict["progress_callback"] = progress_cb if repo_dict.get("vcs") is None: - vcs_matches = url_tools.registry.match(url=repo_dict["url"], is_explicit=True) - - if len(vcs_matches) == 0: - raise Exception(f"No vcs found for {repo_dict}") - if len(vcs_matches) > 1: - raise Exception(f"No exact matches for {repo_dict}") + vcs = guess_vcs(url=repo_dict["url"]) + if vcs is None: + raise Exception( + f'Could not automatically determine VCS for {repo_dict["url"]}' + ) - repo_dict["vcs"] = vcs_matches[0].vcs + repo_dict["vcs"] = vcs r = create_project(**repo_dict) # Creates the repo object r.update_repo(set_remotes=True) # Creates repo if not exists and fetches - return r + # TODO: Fix this + return r # type:ignore diff --git a/src/vcspull/config.py b/src/vcspull/config.py index d1f5c4422..da1a71787 100644 --- a/src/vcspull/config.py +++ b/src/vcspull/config.py @@ -11,11 +11,11 @@ import pathlib import typing as t -from libvcs._internal.types import StrPath from libvcs.sync.git import GitRemote -from vcspull._internal.config_reader import ConfigReader +from vcspull.validator import is_valid_config from . import exc +from ._internal.config_reader import ConfigReader from .types import ConfigDict, RawConfigDict from .util import get_config_dir, update_dict @@ -49,7 +49,9 @@ def expand_dir( return _dir -def extract_repos(config: RawConfigDict, cwd=pathlib.Path.cwd()) -> list[ConfigDict]: +def extract_repos( + config: RawConfigDict, cwd: pathlib.Path = pathlib.Path.cwd() +) -> list[ConfigDict]: """Return expanded configuration. end-user configuration permit inline configuration shortcuts, expand to @@ -70,7 +72,7 @@ def extract_repos(config: RawConfigDict, cwd=pathlib.Path.cwd()) -> list[ConfigD for directory, repos in config.items(): assert isinstance(repos, dict) for repo, repo_data in repos.items(): - conf: dict = {} + conf: t.Dict[str, t.Any] = {} """ repo_name: http://myrepo.com/repo.git @@ -134,7 +136,7 @@ def find_home_config_files( filetype: list[str] = ["json", "yaml"] ) -> list[pathlib.Path]: """Return configs of ``.vcspull.{yaml,json}`` in user's home directory.""" - configs = [] + configs: list[pathlib.Path] = [] yaml_config = pathlib.Path(os.path.expanduser("~/.vcspull.yaml")) has_yaml_config = yaml_config.exists() @@ -165,7 +167,7 @@ def find_config_files( t.Literal["json", "yaml", "*"], list[t.Literal["json", "yaml", "*"]] ] = ["json", "yaml"], include_home: bool = False, -): +) -> t.List[pathlib.Path]: """Return repos from a directory and match. Not recursive. Parameters @@ -189,34 +191,36 @@ def find_config_files( list : list of absolute paths to config files. """ - configs = [] + config_files = [] if path is None: path = get_config_dir() if include_home is True: - configs.extend(find_home_config_files()) + config_files.extend(find_home_config_files()) if isinstance(path, list): for p in path: - configs.extend(find_config_files(p, match, filetype)) - return configs + config_files.extend(find_config_files(p, match, filetype)) + return config_files else: path = pathlib.Path(os.path.expanduser(path)) if isinstance(match, list): for m in match: - configs.extend(find_config_files(path, m, filetype)) + config_files.extend(find_config_files(path, m, filetype)) else: if isinstance(filetype, list): for f in filetype: - configs.extend(find_config_files(path, match, f)) + config_files.extend(find_config_files(path, match, f)) else: match = f"{match}.{filetype}" - configs = list(path.glob(match)) + config_files = list(path.glob(match)) - return configs + return config_files -def load_configs(files: list[StrPath], cwd=pathlib.Path.cwd()): +def load_configs( + files: list[pathlib.Path], cwd: pathlib.Path = pathlib.Path.cwd() +) -> t.List[ConfigDict]: """Return repos from a list of files. Parameters @@ -241,6 +245,7 @@ def load_configs(files: list[StrPath], cwd=pathlib.Path.cwd()): file = pathlib.Path(file) assert isinstance(file, pathlib.Path) conf = ConfigReader._from_file(file) + assert is_valid_config(conf) newrepos = extract_repos(conf, cwd=cwd) if not repos: @@ -295,7 +300,10 @@ def detect_duplicate_repos( return dupes -def in_dir(config_dir=None, extensions: list[str] = [".yml", ".yaml", ".json"]): +def in_dir( + config_dir: t.Optional[pathlib.Path] = None, + extensions: list[str] = [".yml", ".yaml", ".json"], +) -> t.List[str]: """Return a list of configs in ``config_dir``. Parameters @@ -321,8 +329,8 @@ def in_dir(config_dir=None, extensions: list[str] = [".yml", ".yaml", ".json"]): def filter_repos( - config: list[ConfigDict], - dir: t.Union[pathlib.Path, t.Literal["*"], None] = None, + config: t.List[ConfigDict], + dir: t.Union[pathlib.Path, t.Literal["*"], str, None] = None, vcs_url: t.Union[str, None] = None, name: t.Union[str, None] = None, ) -> list[ConfigDict]: @@ -374,7 +382,7 @@ def filter_repos( def is_config_file( filename: str, extensions: t.Union[list[str], str] = [".yml", ".yaml", ".json"] -): +) -> bool: """Return True if file has a valid config file type. Parameters diff --git a/src/vcspull/log.py b/src/vcspull/log.py index 7538b5c7c..81e78c34a 100644 --- a/src/vcspull/log.py +++ b/src/vcspull/log.py @@ -12,6 +12,7 @@ """ import logging import time +import typing as t from colorama import Fore, Style @@ -24,7 +25,10 @@ } -def setup_logger(log=None, level="INFO"): +def setup_logger( + log: t.Optional[logging.Logger] = None, + level: t.Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO", +) -> None: """Setup logging for CLI use. Parameters @@ -50,138 +54,134 @@ def setup_logger(log=None, level="INFO"): repo_logger.addHandler(channel) -def default_log_template(self, record): - """Return the prefix for the log message. Template for Formatter. - - Parameters - ---------- - record : :py:class:`logging.LogRecord` - This is passed in from inside the :py:meth:`logging.Formatter.format` record. - """ - reset = [Style.RESET_ALL] - levelname = [ - LEVEL_COLORS.get(record.levelname), - Style.BRIGHT, - "(%(levelname)s)", - Style.RESET_ALL, - " ", - ] - asctime = [ - "[", - Fore.BLACK, - Style.DIM, - Style.BRIGHT, - "%(asctime)s", - Fore.RESET, - Style.RESET_ALL, - "]", - ] - name = [ - " ", - Fore.WHITE, - Style.DIM, - Style.BRIGHT, - "%(name)s", - Fore.RESET, - Style.RESET_ALL, - " ", - ] - - tpl = "".join(reset + levelname + asctime + name + reset) - - return tpl +class LogFormatter(logging.Formatter): + def template(self, record: logging.LogRecord) -> str: + """Return the prefix for the log message. Template for Formatter. + + Parameters + ---------- + record : :py:class:`logging.LogRecord` + Passed in from inside the :py:meth:`logging.Formatter.format` record. + """ + reset = [Style.RESET_ALL] + levelname = [ + LEVEL_COLORS.get(record.levelname, ""), + Style.BRIGHT, + "(%(levelname)s)", + Style.RESET_ALL, + " ", + ] + asctime = [ + "[", + Fore.BLACK, + Style.DIM, + Style.BRIGHT, + "%(asctime)s", + Fore.RESET, + Style.RESET_ALL, + "]", + ] + name = [ + " ", + Fore.WHITE, + Style.DIM, + Style.BRIGHT, + "%(name)s", + Fore.RESET, + Style.RESET_ALL, + " ", + ] + tpl = "".join(reset + levelname + asctime + name + reset) -class LogFormatter(logging.Formatter): - template = default_log_template + return tpl - def __init__(self, color=True, *args, **kwargs): - logging.Formatter.__init__(self, *args, **kwargs) + def __init__(self, color: bool = True, **kwargs: t.Any) -> None: + logging.Formatter.__init__(self, **kwargs) - def format(self, record): + def format(self, record: logging.LogRecord) -> str: try: record.message = record.getMessage() except Exception as e: record.message = f"Bad message ({e!r}): {record.__dict__!r}" date_format = "%H:%m:%S" - record.asctime = time.strftime(date_format, self.converter(record.created)) - + formatting = self.converter(record.created) # type:ignore + record.asctime = time.strftime(date_format, formatting) prefix = self.template(record) % record.__dict__ formatted = prefix + " " + record.message return formatted.replace("\n", "\n ") -def debug_log_template(self, record): - """Return the prefix for the log message. Template for Formatter. - - Parameters - ---------- - record : :class:`logging.LogRecord` - This is passed in from inside the :py:meth:`logging.Formatter.format` record. - """ - - reset = [Style.RESET_ALL] - levelname = [ - LEVEL_COLORS.get(record.levelname), - Style.BRIGHT, - "(%(levelname)1.1s)", - Style.RESET_ALL, - " ", - ] - asctime = [ - "[", - Fore.BLACK, - Style.DIM, - Style.BRIGHT, - "%(asctime)s", - Fore.RESET, - Style.RESET_ALL, - "]", - ] - name = [ - " ", - Fore.WHITE, - Style.DIM, - Style.BRIGHT, - "%(name)s", - Fore.RESET, - Style.RESET_ALL, - " ", - ] - module_funcName = [Fore.GREEN, Style.BRIGHT, "%(module)s.%(funcName)s()"] - lineno = [ - Fore.BLACK, - Style.DIM, - Style.BRIGHT, - ":", - Style.RESET_ALL, - Fore.CYAN, - "%(lineno)d", - ] - - tpl = "".join(reset + levelname + asctime + name + module_funcName + lineno + reset) - - return tpl - - class DebugLogFormatter(LogFormatter): """Provides greater technical details than standard log Formatter.""" - template = debug_log_template + def template(self, record: logging.LogRecord) -> str: + """Return the prefix for the log message. Template for Formatter. + + Parameters + ---------- + record : :class:`logging.LogRecord` + Passed from inside the :py:meth:`logging.Formatter.format` record. + """ + + reset = [Style.RESET_ALL] + levelname = [ + LEVEL_COLORS.get(record.levelname, ""), + Style.BRIGHT, + "(%(levelname)1.1s)", + Style.RESET_ALL, + " ", + ] + asctime = [ + "[", + Fore.BLACK, + Style.DIM, + Style.BRIGHT, + "%(asctime)s", + Fore.RESET, + Style.RESET_ALL, + "]", + ] + name = [ + " ", + Fore.WHITE, + Style.DIM, + Style.BRIGHT, + "%(name)s", + Fore.RESET, + Style.RESET_ALL, + " ", + ] + module_funcName = [Fore.GREEN, Style.BRIGHT, "%(module)s.%(funcName)s()"] + lineno = [ + Fore.BLACK, + Style.DIM, + Style.BRIGHT, + ":", + Style.RESET_ALL, + Fore.CYAN, + "%(lineno)d", + ] + + tpl = "".join( + reset + levelname + asctime + name + module_funcName + lineno + reset + ) + + return tpl class RepoLogFormatter(LogFormatter): - def template(self, record): + def template(self, record: logging.LogRecord) -> str: record.message = "".join( [Fore.MAGENTA, Style.BRIGHT, record.message, Fore.RESET, Style.RESET_ALL] ) return "{}|{}| {}({}) {}".format( Fore.GREEN + Style.DIM, - record.bin_name, + record.bin_name, # type:ignore Fore.YELLOW, - record.keyword, + record.keyword, # type:ignore Fore.RESET, ) @@ -189,6 +189,6 @@ def template(self, record): class RepoFilter(logging.Filter): """Only include repo logs for this type of record.""" - def filter(self, record): + def filter(self, record: logging.LogRecord) -> bool: """Only return a record if a keyword object.""" return True if "keyword" in record.__dict__ else False diff --git a/src/vcspull/types.py b/src/vcspull/types.py index 6e088dc99..35f4e1473 100644 --- a/src/vcspull/types.py +++ b/src/vcspull/types.py @@ -1,3 +1,4 @@ +import pathlib import typing as t from typing_extensions import NotRequired, TypedDict @@ -21,7 +22,11 @@ class RawConfigDict(t.TypedDict): class ConfigDict(TypedDict): vcs: t.Optional[VCSLiteral] name: str - dir: StrPath + dir: pathlib.Path url: str remotes: NotRequired[t.Optional[GitSyncRemoteDict]] shell_command_after: NotRequired[t.Optional[t.List[str]]] + + +ConfigDir = dict[str, ConfigDict] +Config = dict[str, ConfigDir] diff --git a/src/vcspull/util.py b/src/vcspull/util.py index c056f3a9d..faa267067 100644 --- a/src/vcspull/util.py +++ b/src/vcspull/util.py @@ -6,6 +6,7 @@ """ import os import pathlib +import typing as t from collections.abc import Mapping LEGACY_CONFIG_DIR = os.path.expanduser("~/.vcspull/") # remove dupes of this @@ -45,7 +46,13 @@ def get_config_dir() -> pathlib.Path: return pathlib.Path(path) -def update_dict(d, u): +T = t.TypeVar("T", bound=t.Dict[str, t.Any]) + + +def update_dict( + d: T, + u: T, +) -> T: """Return updated dict. Parameters diff --git a/src/vcspull/validator.py b/src/vcspull/validator.py new file mode 100644 index 000000000..0bad7c0e1 --- /dev/null +++ b/src/vcspull/validator.py @@ -0,0 +1,32 @@ +import pathlib +import typing as t + +from vcspull.types import RawConfigDict + +if t.TYPE_CHECKING: + from typing_extensions import TypeGuard + + +def is_valid_config(config: t.Dict[str, t.Any]) -> "TypeGuard[RawConfigDict]": + if not isinstance(config, dict): + return False + + for k, v in config.items(): + if k is None or v is None: + return False + + if not isinstance(k, str) and not isinstance(k, pathlib.Path): + return False + + if not isinstance(v, dict): + return False + + for repo_name, repo in v.items(): + if not isinstance(repo, (str, dict, pathlib.Path)): + return False + + if isinstance(repo, dict): + if "url" not in repo and "repo" not in repo: + return False + + return True diff --git a/tests/fixtures/_util.py b/tests/fixtures/_util.py index 74b40ab11..3ca85b883 100644 --- a/tests/fixtures/_util.py +++ b/tests/fixtures/_util.py @@ -1,5 +1,5 @@ import os -def curjoin(_file): # return filepath relative to __file__ (this file) +def curjoin(_file: str) -> str: # return filepath relative to __file__ (this file) return os.path.join(os.path.dirname(__file__), _file) diff --git a/tests/fixtures/example.py b/tests/fixtures/example.py index 9d2ed1ae2..06ba292b7 100644 --- a/tests/fixtures/example.py +++ b/tests/fixtures/example.py @@ -1,4 +1,4 @@ -import os +import pathlib from libvcs.sync.git import GitRemote from vcspull.types import ConfigDict @@ -37,32 +37,32 @@ { "vcs": "git", "name": "linux", - "dir": os.path.join("/home/me/myproject/study/", "linux"), + "dir": pathlib.Path("/home/me/myproject/study/linux"), "url": "git+git://git.kernel.org/linux/torvalds/linux.git", }, { "vcs": "git", "name": "freebsd", - "dir": os.path.join("/home/me/myproject/study/", "freebsd"), + "dir": pathlib.Path("/home/me/myproject/study/freebsd"), "url": "git+https://github.com/freebsd/freebsd.git", }, { "vcs": "git", "name": "sphinx", - "dir": os.path.join("/home/me/myproject/study/", "sphinx"), + "dir": pathlib.Path("/home/me/myproject/study/sphinx"), "url": "hg+https://bitbucket.org/birkenfeld/sphinx", }, { "vcs": "git", "name": "docutils", - "dir": os.path.join("/home/me/myproject/study/", "docutils"), + "dir": pathlib.Path("/home/me/myproject/study/docutils"), "url": "svn+http://svn.code.sf.net/p/docutils/code/trunk", }, { "vcs": "git", "name": "kaptan", "url": "git+git@github.com:tony/kaptan.git", - "dir": os.path.join("/home/me/myproject/github_projects/", "kaptan"), + "dir": pathlib.Path("/home/me/myproject/github_projects/kaptan"), "remotes": { "upstream": GitRemote( **{ @@ -83,14 +83,14 @@ { "vcs": "git", "name": ".vim", - "dir": os.path.join("/home/me/myproject", ".vim"), + "dir": pathlib.Path("/home/me/myproject/.vim"), "url": "git+git@github.com:tony/vim-config.git", "shell_command_after": ["ln -sf /home/me/.vim/.vimrc /home/me/.vimrc"], }, { "vcs": "git", "name": ".tmux", - "dir": os.path.join("/home/me/myproject", ".tmux"), + "dir": pathlib.Path("/home/me/myproject/.tmux"), "url": "git+git@github.com:tony/tmux-config.git", "shell_command_after": ["ln -sf /home/me/.tmux/.tmux.conf /home/me/.tmux.conf"], }, diff --git a/tests/helpers.py b/tests/helpers.py index be496840f..acb0b22f8 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -1,7 +1,7 @@ """Helpers for vcspull.""" import os import pathlib -from typing import Literal +import typing as t from vcspull._internal.config_reader import ConfigReader @@ -15,27 +15,27 @@ class EnvironmentVarGuard: including test module, see #121. """ - def __init__(self): + def __init__(self) -> None: self._environ = os.environ - self._unset = set() - self._reset = dict() + self._unset: t.Set[str] = set() + self._reset: t.Dict[str, str] = dict() - def set(self, envvar, value): + def set(self, envvar: str, value: str) -> None: if envvar not in self._environ: self._unset.add(envvar) else: self._reset[envvar] = self._environ[envvar] self._environ[envvar] = value - def unset(self, envvar): + def unset(self, envvar: str) -> None: if envvar in self._environ: self._reset[envvar] = self._environ[envvar] del self._environ[envvar] - def __enter__(self): + def __enter__(self) -> "EnvironmentVarGuard": return self - def __exit__(self, *ignore_exc): + def __exit__(self, *ignore_exc: object) -> None: for envvar, value in self._reset.items(): self._environ[envvar] = value for unset in self._unset: @@ -47,5 +47,5 @@ def write_config(config_path: pathlib.Path, content: str) -> pathlib.Path: return config_path -def load_raw(data: str, format: Literal["yaml", "json"]) -> dict: +def load_raw(data: str, format: t.Literal["yaml", "json"]) -> dict[str, t.Any]: return ConfigReader._load(format=format, content=data) diff --git a/tests/test_cli.py b/tests/test_cli.py index 586f71ffa..0b3fc70ee 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -61,7 +61,7 @@ class SyncCLINonExistentRepo(t.NamedTuple): ) def test_sync_cli_filter_non_existent( tmp_path: pathlib.Path, - capsys: pytest.CaptureFixture, + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, user_path: pathlib.Path, config_path: pathlib.Path, @@ -192,7 +192,7 @@ class SyncFixture(t.NamedTuple): ) def test_sync( tmp_path: pathlib.Path, - capsys: pytest.CaptureFixture, + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, user_path: pathlib.Path, config_path: pathlib.Path, @@ -324,7 +324,7 @@ class SyncBrokenFixture(t.NamedTuple): ) def test_sync_broken( tmp_path: pathlib.Path, - capsys: pytest.CaptureFixture, + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, user_path: pathlib.Path, config_path: pathlib.Path, diff --git a/tests/test_config.py b/tests/test_config.py index e5d7713af..190da1683 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,13 +1,29 @@ import pathlib +import typing as t import pytest from vcspull import config +from vcspull.types import ConfigDict + + +class LoadYAMLFn(t.Protocol): + def __call__( + self, + content: str, + dir: str = "randomdir", + filename: str = "randomfilename.yaml", + ) -> t.Tuple[ + pathlib.Path, t.List[t.Union[t.Any, pathlib.Path]], t.List[ConfigDict] + ]: + ... @pytest.fixture -def load_yaml(tmp_path: pathlib.Path): - def fn(content, dir="randomdir", filename="randomfilename.yaml"): +def load_yaml(tmp_path: pathlib.Path) -> LoadYAMLFn: + def fn( + content: str, dir: str = "randomdir", filename: str = "randomfilename.yaml" + ) -> t.Tuple[pathlib.Path, t.List[pathlib.Path], t.List[ConfigDict]]: _dir = tmp_path / dir _dir.mkdir() _config = _dir / filename @@ -20,7 +36,7 @@ def fn(content, dir="randomdir", filename="randomfilename.yaml"): return fn -def test_simple_format(load_yaml): +def test_simple_format(load_yaml: LoadYAMLFn) -> None: dir, _, repos = load_yaml( """ vcspull: @@ -35,7 +51,7 @@ def test_simple_format(load_yaml): assert dir / "vcspull" / "libvcs" == repo["dir"] -def test_relative_dir(load_yaml): +def test_relative_dir(load_yaml: LoadYAMLFn) -> None: dir, _, repos = load_yaml( """ ./relativedir: @@ -43,8 +59,8 @@ def test_relative_dir(load_yaml): """ ) - configs = config.find_config_files(path=dir) - repos = config.load_configs(configs, dir) + config_files = config.find_config_files(path=dir) + repos = config.load_configs(config_files, dir) assert len(repos) == 1 repo = repos[0] diff --git a/tests/test_config_file.py b/tests/test_config_file.py index 9219e2f59..520790799 100644 --- a/tests/test_config_file.py +++ b/tests/test_config_file.py @@ -8,26 +8,27 @@ from vcspull import config, exc from vcspull._internal.config_reader import ConfigReader from vcspull.config import expand_dir, extract_repos +from vcspull.validator import is_valid_config from .fixtures import example as fixtures from .helpers import EnvironmentVarGuard, load_raw, write_config @pytest.fixture(scope="function") -def yaml_config(config_path: pathlib.Path): +def yaml_config(config_path: pathlib.Path) -> pathlib.Path: yaml_file = config_path / "repos1.yaml" yaml_file.touch() return yaml_file @pytest.fixture(scope="function") -def json_config(config_path: pathlib.Path): +def json_config(config_path: pathlib.Path) -> pathlib.Path: json_file = config_path / "repos2.json" json_file.touch() return json_file -def test_dict_equals_yaml(): +def test_dict_equals_yaml() -> None: # Verify that example YAML is returning expected dict format. config = ConfigReader._load( format="yaml", @@ -56,7 +57,7 @@ def test_dict_equals_yaml(): assert fixtures.config_dict == config -def test_export_json(tmp_path: pathlib.Path): +def test_export_json(tmp_path: pathlib.Path) -> None: json_config = tmp_path / ".vcspull.json" config = ConfigReader(content=fixtures.config_dict) @@ -69,7 +70,7 @@ def test_export_json(tmp_path: pathlib.Path): assert fixtures.config_dict == new_config -def test_export_yaml(tmp_path: pathlib.Path): +def test_export_yaml(tmp_path: pathlib.Path) -> None: yaml_config = tmp_path / ".vcspull.yaml" config = ConfigReader(content=fixtures.config_dict) @@ -81,7 +82,7 @@ def test_export_yaml(tmp_path: pathlib.Path): assert fixtures.config_dict == new_config -def test_scan_config(tmp_path: pathlib.Path): +def test_scan_config(tmp_path: pathlib.Path) -> None: config_files = [] exists = os.path.exists @@ -106,14 +107,15 @@ def test_scan_config(tmp_path: pathlib.Path): assert len(config_files) == files -def test_expand_shell_command_after(): +def test_expand_shell_command_after() -> None: # Expand shell commands from string to list. + assert is_valid_config(fixtures.config_dict) config = extract_repos(fixtures.config_dict) assert config, fixtures.config_dict_expanded -def test_expandenv_and_homevars(): +def test_expandenv_and_homevars() -> None: # Assure ~ tildes and environment template vars expand. config1 = load_raw( """\ @@ -155,20 +157,23 @@ def test_expandenv_and_homevars(): format="json", ) + assert is_valid_config(config1) + assert is_valid_config(config2) + config1_expanded = extract_repos(config1) config2_expanded = extract_repos(config2) paths = [r["dir"].parent for r in config1_expanded] - assert expand_dir("${HOME}/github_projects/") in paths - assert expand_dir("~/study/") in paths - assert expand_dir("~") in paths + assert expand_dir(pathlib.Path("${HOME}/github_projects/")) in paths + assert expand_dir(pathlib.Path("~/study/")) in paths + assert expand_dir(pathlib.Path("~")) in paths paths = [r["dir"].parent for r in config2_expanded] - assert expand_dir("${HOME}/github_projects/") in paths - assert expand_dir("~/study/") in paths + assert expand_dir(pathlib.Path("${HOME}/github_projects/")) in paths + assert expand_dir(pathlib.Path("~/study/")) in paths -def test_find_config_files(tmp_path: pathlib.Path): +def test_find_config_files(tmp_path: pathlib.Path) -> None: # Test find_config_files in home directory. pull_config = tmp_path / ".vcspull.yaml" @@ -182,7 +187,7 @@ def test_find_config_files(tmp_path: pathlib.Path): assert expected_in in results -def test_multiple_config_files_raises_exception(tmp_path: pathlib.Path): +def test_multiple_config_files_raises_exception(tmp_path: pathlib.Path) -> None: json_conf_file = tmp_path / ".vcspull.json" json_conf_file.touch() yaml_conf_file = tmp_path / ".vcspull.yaml" @@ -199,23 +204,27 @@ def test_in_dir( config_path: pathlib.Path, yaml_config: pathlib.Path, json_config: pathlib.Path, -): +) -> None: expected = [yaml_config.stem, json_config.stem] - result = config.in_dir(str(config_path)) + result = config.in_dir(config_path) assert len(expected) == len(result) def test_find_config_path_string( config_path: pathlib.Path, yaml_config: pathlib.Path, json_config: pathlib.Path -): +) -> None: config_files = config.find_config_files(path=config_path) assert yaml_config in config_files assert json_config in config_files -def test_find_config_path_list(config_path, yaml_config, json_config): +def test_find_config_path_list( + config_path: pathlib.Path, + yaml_config: pathlib.Path, + json_config: pathlib.Path, +) -> None: config_files = config.find_config_files(path=[config_path]) assert yaml_config in config_files @@ -227,7 +236,7 @@ def test_find_config_match_string( yaml_config: pathlib.Path, json_config: pathlib.Path, monkeypatch: pytest.MonkeyPatch, -): +) -> None: config_files = config.find_config_files(path=config_path, match=yaml_config.stem) assert yaml_config in config_files assert json_config not in config_files @@ -254,7 +263,11 @@ def test_find_config_match_string( assert json_config in config_files -def test_find_config_match_list(config_path, yaml_config, json_config): +def test_find_config_match_list( + config_path: pathlib.Path, + yaml_config: pathlib.Path, + json_config: pathlib.Path, +) -> None: config_files = config.find_config_files( path=[config_path], match=[yaml_config.stem, json_config.stem], @@ -273,7 +286,7 @@ def test_find_config_match_list(config_path, yaml_config, json_config): def test_find_config_filetype_string( config_path: pathlib.Path, yaml_config: pathlib.Path, json_config: pathlib.Path -): +) -> None: config_files = config.find_config_files( path=[config_path], match=yaml_config.stem, filetype="yaml" ) @@ -301,7 +314,7 @@ def test_find_config_filetype_string( def test_find_config_filetype_list( config_path: pathlib.Path, yaml_config: pathlib.Path, json_config: pathlib.Path -): +) -> None: config_files = config.find_config_files( path=[config_path], match=["repos*"], filetype=["*"] ) @@ -326,7 +339,7 @@ def test_find_config_include_home_config_files( config_path: pathlib.Path, yaml_config: pathlib.Path, json_config: pathlib.Path, -): +) -> None: with EnvironmentVarGuard() as env: env.set("HOME", str(tmp_path)) config_files = config.find_config_files( @@ -346,7 +359,7 @@ def test_find_config_include_home_config_files( assert json_config in results -def test_merge_nested_dict(tmp_path: pathlib.Path, config_path: pathlib.Path): +def test_merge_nested_dict(tmp_path: pathlib.Path, config_path: pathlib.Path) -> None: config1 = write_config( config_path=config_path / "repoduplicate1.yaml", content=textwrap.dedent( diff --git a/tests/test_repo.py b/tests/test_repo.py index ae6d2e08a..50bcb31fe 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -8,7 +8,7 @@ from .fixtures import example as fixtures -def test_filter_dir(): +def test_filter_dir() -> None: """`filter_repos` filter by dir""" repo_list = filter_repos(fixtures.config_dict_expanded, dir="*github_project*") @@ -17,7 +17,7 @@ def test_filter_dir(): assert r["name"] == "kaptan" -def test_filter_name(): +def test_filter_name() -> None: """`filter_repos` filter by name""" repo_list = filter_repos(fixtures.config_dict_expanded, name=".vim") @@ -26,7 +26,7 @@ def test_filter_name(): assert r["name"] == ".vim" -def test_filter_vcs(): +def test_filter_vcs() -> None: """`filter_repos` filter by vcs remote url""" repo_list = filter_repos(fixtures.config_dict_expanded, vcs_url="*kernel.org*") @@ -35,7 +35,7 @@ def test_filter_vcs(): assert r["name"] == "linux" -def test_to_dictlist(): +def test_to_dictlist() -> None: """`filter_repos` pulls the repos in dict format from the config.""" repo_list = filter_repos(fixtures.config_dict_expanded) @@ -54,7 +54,7 @@ def test_to_dictlist(): assert "url" == remote -def test_vcs_url_scheme_to_object(tmp_path: pathlib.Path): +def test_vcs_url_scheme_to_object(tmp_path: pathlib.Path) -> None: """Verify `url` return {Git,Mercurial,Subversion}Project. :class:`GitSync`, :class:`HgSync` or :class:`SvnSync` @@ -91,7 +91,7 @@ def test_vcs_url_scheme_to_object(tmp_path: pathlib.Path): assert isinstance(svn_repo, BaseSync) -def test_to_repo_objects(tmp_path: pathlib.Path): +def test_to_repo_objects(tmp_path: pathlib.Path) -> None: """:py:obj:`dict` objects into Project objects.""" repo_list = filter_repos(fixtures.config_dict_expanded) for repo_dict in repo_list: diff --git a/tests/test_sync.py b/tests/test_sync.py index 78ae95314..8ce342bea 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -11,6 +11,7 @@ from vcspull.cli.sync import update_repo from vcspull.config import extract_repos, filter_repos, load_configs from vcspull.types import ConfigDict +from vcspull.validator import is_valid_config from .helpers import write_config @@ -18,7 +19,7 @@ def test_makes_recursive( tmp_path: pathlib.Path, git_remote_repo: pathlib.Path, -): +) -> None: """Ensure that directories in pull are made recursively.""" conf = ConfigReader._load( format="yaml", @@ -29,23 +30,28 @@ def test_makes_recursive( """ ), ) - repos = extract_repos(conf) - assert len(repos) > 0 + if is_valid_config(conf): + repos = extract_repos(config=conf) + assert len(repos) > 0 - filtered_repos = filter_repos(repos, dir="*") - assert len(filtered_repos) > 0 + filtered_repos = filter_repos(repos, dir="*") + assert len(filtered_repos) > 0 - for r in filtered_repos: - assert isinstance(r, dict) - repo = create_project(**r) # type: ignore - repo.obtain() + for r in filtered_repos: + assert isinstance(r, dict) + repo = create_project(**r) # type: ignore + repo.obtain() - assert repo.dir.exists() + assert repo.dir.exists() def write_config_remote( - config_path: pathlib.Path, tmp_path: pathlib.Path, config_tpl, dir, clone_name -): + config_path: pathlib.Path, + tmp_path: pathlib.Path, + config_tpl: str, + dir: pathlib.Path, + clone_name: str, +) -> pathlib.Path: return write_config( config_path=config_path, content=config_tpl.format( @@ -102,7 +108,7 @@ def test_config_variations( dir=dummy_repo, clone_name="myclone", ) - configs = load_configs([str(config_file)]) + configs = load_configs([config_file]) # TODO: Merge repos repos = filter_repos(configs, dir="*") @@ -173,7 +179,7 @@ def test_updating_remote( initial_config: ConfigDict = { "vcs": "git", "name": "myclone", - "dir": f"{tmp_path}/study/myrepo/myclone", + "dir": tmp_path / "study/myrepo/myclone", "url": f"git+file://{dummy_repo}", "remotes": { mirror_name: GitRemote( @@ -201,6 +207,7 @@ def test_updating_remote( ) repo_dict = filter_repos([expected_config], name="myclone")[0] + assert isinstance(repo_dict, dict) repo = update_repo(repo_dict) for remote_name, remote_info in repo.remotes().items(): remote = repo.remote(remote_name) diff --git a/tests/test_utils.py b/tests/test_utils.py index 364e6db6c..7c0b434ce 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -7,7 +7,7 @@ def test_vcspull_configdir_env_var( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch -): +) -> None: monkeypatch.setenv("VCSPULL_CONFIGDIR", str(tmp_path)) assert get_config_dir() == tmp_path @@ -15,7 +15,7 @@ def test_vcspull_configdir_env_var( def test_vcspull_configdir_xdg_config_dir( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch -): +) -> None: monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) vcspull_dir = tmp_path / "vcspull" vcspull_dir.mkdir() @@ -23,6 +23,6 @@ def test_vcspull_configdir_xdg_config_dir( assert get_config_dir() == vcspull_dir -def test_vcspull_configdir_no_xdg(monkeypatch: pytest.MonkeyPatch): +def test_vcspull_configdir_no_xdg(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("XDG_CONFIG_HOME") assert get_config_dir()