diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 000000000..1246879c4 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +*.dump eol=lf diff --git a/CHANGES b/CHANGES index 34f2e89f3..ce24d15f7 100644 --- a/CHANGES +++ b/CHANGES @@ -15,6 +15,48 @@ $ pip install --user --upgrade --pre libvcs +### New features + +#### Commands + +via #430 + +- Git + + - Support for progress bar + - Add subcommands for: + - stash: {attr}`Git.stash ` -> {class}`libvcs.cmd.git.GitStashCmd` + - remote: {attr}`Git.remote ` -> {class}`libvcs.cmd.git.GitRemoteCmd` + - submodule: {attr}`Git.submodule ` -> + {class}`libvcs.cmd.git.GitSubmoduleCmd` + - Added commands for: + - {meth}`libvcs.cmd.git.Git.rev_parse` + - {meth}`libvcs.cmd.git.Git.rev_list` + - {meth}`libvcs.cmd.git.Git.symbolic_ref` + - {meth}`libvcs.cmd.git.Git.show_ref` + +- SVN + + New and improved: + + - {meth}`libvcs.cmd.svn.Svn.unlock` + - {meth}`libvcs.cmd.svn.Svn.lock` + - {meth}`libvcs.cmd.svn.Svn.propset` + +- Mercurial + + New and improved: + + - {meth}`libvcs.cmd.hg.Hg.pull` + - {meth}`libvcs.cmd.hg.Hg.clone` + - {meth}`libvcs.cmd.hg.Hg.update` + +#### Syncing + +via #430 + +Git, SVN, and Mercurial have moved to `libvcs.cmd` + ## libvcs 0.18.1 (2022-10-23) _Maintenance only release, no bug fixes or features_ diff --git a/src/libvcs/_internal/run.py b/src/libvcs/_internal/run.py index 05a2708b1..e7c8edc11 100644 --- a/src/libvcs/_internal/run.py +++ b/src/libvcs/_internal/run.py @@ -126,7 +126,7 @@ def run( # Not until sys.version_info >= (3, 10) # pipesize: int = -1, # custom - log_in_real_time: bool = True, + log_in_real_time: bool = False, check_returncode: bool = True, callback: Optional[ProgressCallbackProtocol] = None, ) -> str: @@ -196,6 +196,13 @@ def progress_cb(output, timestamp): all_output: list[str] = [] code = None line = None + if log_in_real_time and callback is None: + + def progress_cb(output: AnyStr, timestamp: datetime.datetime) -> None: + sys.stdout.write(str(output)) + sys.stdout.flush() + + callback = progress_cb while code is None: code = proc.poll() diff --git a/src/libvcs/cmd/git.py b/src/libvcs/cmd/git.py index d9e1de5e9..46633d60d 100644 --- a/src/libvcs/cmd/git.py +++ b/src/libvcs/cmd/git.py @@ -1,16 +1,29 @@ +import datetime import pathlib import shlex from collections.abc import Sequence from typing import Any, Literal, Optional, Union -from libvcs._internal.run import run +from libvcs._internal.run import ProgressCallbackProtocol, run from libvcs._internal.types import StrOrBytesPath, StrPath _CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]] class Git: - def __init__(self, *, dir: StrPath) -> None: + progress_callback: Optional[ProgressCallbackProtocol] = None + + # Sub-commands + submodule: "GitSubmoduleCmd" + remote: "GitRemoteCmd" + stash: "GitStashCmd" + + def __init__( + self, + *, + dir: StrPath, + progress_callback: Optional[ProgressCallbackProtocol] = None, + ) -> None: """Lite, typed, pythonic wrapper for git(1). Parameters @@ -30,6 +43,13 @@ def __init__(self, *, dir: StrPath) -> None: else: self.dir = pathlib.Path(dir) + self.progress_callback = progress_callback + + # Initial git-submodule + self.submodule = GitSubmoduleCmd(dir=self.dir, cmd=self) + self.remote = GitRemoteCmd(dir=self.dir, cmd=self) + self.stash = GitStashCmd(dir=self.dir, cmd=self) + def __repr__(self) -> str: return f"" @@ -58,8 +78,10 @@ def run( noglob_pathspecs: Optional[bool] = None, icase_pathspecs: Optional[bool] = None, no_optional_locks: Optional[bool] = None, - config: Optional[str] = None, + config: Optional[dict[str, Any]] = None, config_env: Optional[str] = None, + # Pass-through to run() + log_in_real_time: bool = False, **kwargs: Any, ) -> str: """ @@ -156,6 +178,18 @@ def run( C = [C] C = [str(c) for c in C] cli_args.extend(["-C", C]) + if config is not None: + assert isinstance(config, dict) + + def stringify(v: Any) -> str: + if isinstance(v, bool): + return "true" if True else "false" + elif not isinstance(v, str): + return str(v) + return v + + for k, v in config.items(): + cli_args.extend(["--config", f"{k}={stringify(v)}"]) if git_dir is not None: cli_args.extend(["--git-dir", str(git_dir)]) if work_tree is not None: @@ -181,6 +215,9 @@ def run( if no_optional_locks is True: cli_args.append("--no-optional-locks") + if self.progress_callback is not None: + kwargs["callback"] = self.progress_callback + return run(args=cli_args, **kwargs) def clone( @@ -189,7 +226,7 @@ def clone( url: str, separate_git_dir: Optional[StrOrBytesPath] = None, template: Optional[str] = None, - depth: Optional[str] = None, + depth: Optional[int] = None, branch: Optional[str] = None, origin: Optional[str] = None, upload_pack: Optional[str] = None, @@ -216,7 +253,11 @@ def clone( no_remote_submodules: Optional[bool] = None, verbose: Optional[bool] = None, quiet: Optional[bool] = None, + # Pass-through to run + config: Optional[dict[str, Any]] = None, + log_in_real_time: bool = False, # Special behavior + check_returncode: Optional[bool] = None, make_parents: Optional[bool] = True, **kwargs: Any, ) -> str: @@ -254,7 +295,7 @@ def clone( if (filter := kwargs.pop("filter", None)) is not None: local_flags.append(f"--filter={filter}") if depth is not None: - local_flags.extend(["--depth", depth]) + local_flags.extend(["--depth", str(depth)]) if branch is not None: local_flags.extend(["--branch", branch]) if origin is not None: @@ -308,7 +349,10 @@ def clone( if make_parents and not self.dir.exists(): self.dir.mkdir(parents=True) return self.run( - ["clone", *local_flags, "--", *required_flags], check_returncode=False + ["clone", *local_flags, "--", *required_flags], + config=config, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, ) def fetch( @@ -365,6 +409,8 @@ def fetch( show_forced_updates: Optional[bool] = None, no_show_forced_updates: Optional[bool] = None, negotiate_only: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Download from repo. Wraps `git fetch `_. @@ -466,7 +512,8 @@ def fetch( if negotiate_only: local_flags.append("--negotiate-only") return self.run( - ["fetch", *local_flags, "--", *required_flags], check_returncode=False + ["fetch", *local_flags, "--", *required_flags], + check_returncode=check_returncode, ) def rebase( @@ -524,6 +571,8 @@ def rebase( show_current_patch: Optional[bool] = None, abort: Optional[bool] = None, quit: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Reapply commit on top of another tip. @@ -663,7 +712,7 @@ def rebase( local_flags.append("--quit") return self.run( - ["rebase", *local_flags, *required_flags], check_returncode=False + ["rebase", *local_flags, *required_flags], check_returncode=check_returncode ) def pull( @@ -761,6 +810,9 @@ def pull( show_forced_updates: Optional[bool] = None, no_show_forced_updates: Optional[bool] = None, negotiate_only: Optional[bool] = None, + # Pass-through to run + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Download from repo. Wraps `git pull `_. @@ -942,7 +994,9 @@ def pull( if negotiate_only: local_flags.append("--negotiate-only") return self.run( - ["pull", *local_flags, "--", *required_flags], check_returncode=False + ["pull", *local_flags, "--", *required_flags], + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, ) def init( @@ -956,6 +1010,8 @@ def init( shared: Optional[bool] = None, quiet: Optional[bool] = None, bare: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Create empty repo. Wraps `git init `_. @@ -1022,7 +1078,8 @@ def init( local_flags.append("--bare") return self.run( - ["init", *local_flags, "--", *required_flags], check_returncode=False + ["init", *local_flags, "--", *required_flags], + check_returncode=check_returncode, ) def help( @@ -1037,6 +1094,8 @@ def help( info: Optional[bool] = None, man: Optional[bool] = None, web: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Help info. Wraps `git help `_. @@ -1107,7 +1166,7 @@ def help( if web is True: local_flags.append("--web") - return self.run(["help", *local_flags], check_returncode=False) + return self.run(["help", *local_flags], check_returncode=check_returncode) def reset( self, @@ -1125,6 +1184,8 @@ def reset( commit: Optional[str] = None, recurse_submodules: Optional[bool] = None, no_recurse_submodules: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Reset HEAD. Wraps `git help `_. @@ -1198,7 +1259,7 @@ def reset( return self.run( ["reset", *local_flags, *(["--", *pathspec] if len(pathspec) else [])], - check_returncode=False, + check_returncode=check_returncode, ) def checkout( @@ -1233,6 +1294,8 @@ def checkout( new_branch: Optional[str] = None, start_point: Optional[str] = None, treeish: Optional[str] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Switches branches or checks out files. Wraps @@ -1332,7 +1395,7 @@ def checkout( return self.run( ["checkout", *local_flags, *(["--", *pathspec] if len(pathspec) else [])], - check_returncode=False, + check_returncode=check_returncode, ) def status( @@ -1355,6 +1418,8 @@ def status( ignored: Optional[Literal["traditional", "no", "matching"]] = None, ignored_submodules: Optional[Literal["untracked", "dirty", "all"]] = None, pathspec: Optional[Union[StrOrBytesPath, list[StrOrBytesPath]]] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Status of working tree. Wraps @@ -1402,6 +1467,9 @@ def status( >>> git.status(C=git_local_clone.dir / '.git', porcelain='2') '? new_file.txt' + + >>> git.status(porcelain=True, untracked_files="no") + '' """ local_flags: list[str] = [] @@ -1460,7 +1528,7 @@ def status( return self.run( ["status", *local_flags, *(["--", *pathspec] if len(pathspec) else [])], - check_returncode=False, + check_returncode=check_returncode, ) def config( @@ -1497,6 +1565,8 @@ def config( no_includes: Optional[bool] = None, includes: Optional[bool] = None, add: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """Status of working tree. Wraps @@ -1646,5 +1716,1197 @@ def config( return self.run( ["config", *local_flags], - check_returncode=False, + check_returncode=check_returncode, + ) + + def version( + self, + *, + build_options: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """Version. Wraps `git version `_. + + Examples + -------- + >>> git = Git(dir=git_local_clone.dir) + + >>> git.version() + 'git version ...' + + >>> git.version(build_options=True) + 'git version ...' + """ + local_flags: list[str] = [] + + if build_options is True: + local_flags.append("--build-options") + + return self.run( + ["version", *local_flags], + check_returncode=check_returncode, + ) + + def rev_parse( + self, + *, + parseopt: Optional[bool] = None, + sq_quote: Optional[bool] = None, + keep_dashdash: Optional[bool] = None, + stop_at_non_option: Optional[bool] = None, + stuck_long: Optional[bool] = None, + verify: Optional[bool] = None, + args: Optional[str] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """rev-parse. Wraps `git rev-parse `_. + + Examples + -------- + >>> git = Git(dir=git_local_clone.dir) + + >>> git.rev_parse() + '' + + >>> git.rev_parse(parseopt=True) + 'usage: git rev-parse --parseopt...' + + >>> git.rev_parse(verify=True, args='HEAD') + '...' + """ + local_flags: list[str] = [] + + if parseopt is True: + local_flags.append("--parseopt") + + if keep_dashdash is True: + local_flags.append("--keep-dashdash") + if stop_at_non_option is True: + local_flags.append("--stop-at-non-option") + if stuck_long is True: + local_flags.append("--stuck-long") + if sq_quote is True: + local_flags.append("--sq-quote") + if verify is True: + local_flags.append("--verify") + + if parseopt is True: + if args is not None: + local_flags.extend(["--", args]) + else: + if args is not None: + local_flags.append(args) + + return self.run( + ["rev-parse", *local_flags], + check_returncode=check_returncode, + ) + + def rev_list( + self, + *, + commit: Optional[Union[list[str], str]], + path: Optional[Union[list[StrPath], StrPath]] = None, + # + # Limiting + # + max_count: Optional[int] = None, + skip: Optional[int] = None, + since: Optional[str] = None, + after: Optional[str] = None, + until: Optional[str] = None, + before: Optional[str] = None, + max_age: Optional[str] = None, + min_age: Optional[str] = None, + author: Optional[str] = None, + committer: Optional[str] = None, + grep: Optional[str] = None, + all_match: Optional[bool] = None, + invert_grep: Optional[bool] = None, + regexp_ignore_case: Optional[bool] = None, + basic_regexp: Optional[bool] = None, + extended_regexp: Optional[bool] = None, + fixed_strings: Optional[bool] = None, + perl_regexp: Optional[bool] = None, + remove_empty: Optional[bool] = None, + merges: Optional[bool] = None, + no_merges: Optional[bool] = None, + no_min_parents: Optional[bool] = None, + min_parents: Optional[int] = None, + no_max_parents: Optional[bool] = None, + max_parents: Optional[int] = None, + first_parent: Optional[bool] = None, + exclude_first_parent_only: Optional[bool] = None, + _not: Optional[bool] = None, + all: Optional[bool] = None, + branches: Optional[Union[str, bool]] = None, + tags: Optional[Union[str, bool]] = None, + remotes: Optional[Union[str, bool]] = None, + exclude: Optional[bool] = None, + reflog: Optional[bool] = None, + alternative_refs: Optional[bool] = None, + single_worktree: Optional[bool] = None, + ignore_missing: Optional[bool] = None, + stdin: Optional[bool] = None, + disk_usage: Optional[Union[bool, str]] = None, + cherry_mark: Optional[bool] = None, + cherry_pick: Optional[bool] = None, + left_only: Optional[bool] = None, + right_only: Optional[bool] = None, + cherry: Optional[bool] = None, + walk_reflogs: Optional[bool] = None, + merge: Optional[bool] = None, + boundary: Optional[bool] = None, + use_bitmap_index: Optional[bool] = None, + progress: Optional[Union[str, bool]] = None, + # Formatting + # + # --parents + # --children + # --objects | --objects-edge + # --disk-usage[=human] + # --unpacked + # --header | --pretty + # --[no-]object-names + # --abbrev= | --no-abbrev + # --abbrev-commit + # --left-right + # --count + header: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = True, + log_in_real_time: bool = False, + **kwargs: Any, + ) -> str: + """rev-list. Wraps `git rev-list `_. + + Examples + -------- + >>> git = Git(dir=git_local_clone.dir) + + >>> git.rev_list(commit="HEAD") + '...' + + >>> git.run(['commit', '--allow-empty', '--message=Moo']) + '[master ...] Moo' + + >>> git.rev_list(commit="HEAD", max_count=1) + '' + + >>> git.rev_list(commit="HEAD", path=".", max_count=1, header=True) + '' + + >>> git.rev_list(commit="origin..HEAD", max_count=1, all=True, header=True) + '' + + >>> git.rev_list(commit="origin..HEAD", max_count=1, header=True) + '' + """ + required_flags: list[str] = [] + path_flags: list[str] = [] + local_flags: list[str] = [] + + if isinstance(commit, str): + required_flags.append(commit) + if isinstance(commit, list): + required_flags.extend(commit) + + if isinstance(path, str): + path_flags.append(path) + if isinstance(path, list): + path_flags.extend(str(pathlib.Path(p).absolute()) for p in path) + elif isinstance(path, pathlib.Path): + path_flags.append(str(pathlib.Path(path).absolute())) + + for kwarg, kwarg_shell_flag in [ + (branches, "--branches"), + (tags, "--tags"), + (remotes, "--remotes"), + (disk_usage, "--disk-usage"), + (progress, "--progress"), + ]: + if kwarg is not None: + if isinstance(kwarg, str): + local_flags.extend([kwarg_shell_flag, kwarg]) + elif kwarg: + local_flags.append(kwarg_shell_flag) + + for datetime_kwarg, datetime_shell_flag in [ # 1.year.ago + (since, "--since"), + (after, "--after"), + (until, "--until"), + (before, "--before"), + (max_age, "--max-age"), + (min_age, "--min-age"), + ]: + if datetime_kwarg is not None: + if isinstance(datetime, str): + local_flags.extend([datetime_shell_flag, datetime_kwarg]) + + for int_flag, int_shell_flag in [ + (max_count, "--max-count"), + (skip, "--skip"), + (min_parents, "--min-parents"), + (max_parents, "--max-parents"), + ]: + if int_flag is not None: + local_flags.extend([int_shell_flag, str(int_shell_flag)]) + + for flag, shell_flag in [ + # Limiting output + (all, "--all"), + (author, "--author"), + (committer, "--committer"), + (grep, "--grep"), + (all_match, "--all-match"), + (invert_grep, "--invert-grep"), + (regexp_ignore_case, "--regexp-ignore-case"), + (basic_regexp, "--basic-regexp"), + (extended_regexp, "--extended-regexp"), + (fixed_strings, "--fixed-strings"), + (perl_regexp, "--perl-regexp"), + (remove_empty, "--remove-empty"), + (merges, "--merges"), + (no_merges, "--no-merges"), + (no_min_parents, "--no-min-parents"), + (no_max_parents, "--no-max-parents"), + (first_parent, "--first-parent"), + (exclude_first_parent_only, "--exclude-first-parent-only"), + (_not, "--not"), + (all, "--all"), + (exclude, "--exclude"), + (reflog, "--reflog"), + (alternative_refs, "--alternative-refs"), + (single_worktree, "--single-worktree"), + (ignore_missing, "--ignore-missing"), + (stdin, "--stdin"), + (cherry_mark, "--cherry-mark"), + (cherry_pick, "--cherry-pick"), + (left_only, "--left-only"), + (right_only, "--right-only"), + (cherry, "--cherry"), + (walk_reflogs, "--walk-reflogs"), + (merge, "--merge"), + (boundary, "--boundary"), + (use_bitmap_index, "--use-bitmap-index"), + # Formatting outputs + (header, "--header"), + ]: + if flag is not None and flag: + local_flags.append(shell_flag) + + return self.run( + [ + "rev-list", + *local_flags, + *required_flags, + *(["--", *path_flags] if len(path_flags) else []), + ], + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def symbolic_ref( + self, + *, + name: str, + ref: Optional[str] = None, + message: Optional[str] = None, + short: Optional[bool] = None, + delete: Optional[bool] = None, + quiet: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """symbolic-ref. Wraps `git symbolic-ref + `_. + + Examples + -------- + >>> git = Git(dir=git_local_clone.dir) + + >>> git.symbolic_ref(name="test") + 'fatal: ref test is not a symbolic ref' + + >>> git.symbolic_ref(name="test") + 'fatal: ref test is not a symbolic ref' + """ + required_flags: list[str] = [name] + local_flags: list[str] = [] + + if message is not None and isinstance(message, str): + local_flags.extend(["-m", message]) + + if delete is True: + local_flags.append("--delete") + if short is True: + local_flags.append("--short") + if quiet is True: + local_flags.append("--quiet") + + return self.run( + ["symbolic-ref", *required_flags, *local_flags], + check_returncode=check_returncode, + ) + + def show_ref( + self, + *, + pattern: Optional[Union[list[str], str]] = None, + quiet: Optional[bool] = None, + verify: Optional[bool] = None, + head: Optional[bool] = None, + dereference: Optional[bool] = None, + tags: Optional[bool] = None, + hash: Optional[Union[str, bool]] = None, + abbrev: Optional[Union[str, bool]] = None, + # libvcs special behavior + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + r"""show-ref. Wraps `git show-ref `_. + + Examples + -------- + >>> git = Git(dir=git_local_clone.dir) + + >>> git.show_ref() + '...' + + >>> git.show_ref(pattern='master') + '...' + + >>> git.show_ref(pattern='master', head=True) + '...' + + >>> git.show_ref(pattern='HEAD', verify=True) + '... HEAD' + + >>> git.show_ref(pattern='master', dereference=True) + '... refs/heads/master\n... refs/remotes/origin/master' + + >>> git.show_ref(pattern='HEAD', tags=True) + '' + """ + local_flags: list[str] = [] + pattern_flags: list[str] = [] + + if pattern is not None: + if isinstance(pattern, str): + pattern_flags.append(pattern) + elif isinstance(pattern, list): + pattern_flags.extend(pattern) + + for kwarg, kwarg_shell_flag in [ + (hash, "--hash"), + (abbrev, "--abbrev"), + ]: + if kwarg is not None: + if isinstance(kwarg, str): + local_flags.extend([kwarg_shell_flag, kwarg]) + elif kwarg: + local_flags.append(kwarg_shell_flag) + + for flag, shell_flag in [ + (quiet, "--quiet"), + (verify, "--verify"), + (head, "--head"), + (dereference, "--dereference"), + (tags, "--tags"), + ]: + if flag is not None and flag: + local_flags.append(shell_flag) + + return self.run( + [ + "show-ref", + *local_flags, + *(["--", *pattern_flags] if len(pattern_flags) else []), + ], + check_returncode=check_returncode, + ) + + +GitSubmoduleCmdCommandLiteral = Literal[ + "status", + "init", + "deinit", + "update", + "set-branch", + "set-url", + "summary", + "foreach", + "sync", + "absorbgitdirs", +] + + +class GitSubmoduleCmd: + def __init__(self, *, dir: StrPath, cmd: Optional[Git] = None) -> None: + """Lite, typed, pythonic wrapper for git-submodule(1). + + Parameters + ---------- + dir : + Operates as PATH in the corresponding git subcommand. + + Examples + -------- + >>> GitSubmoduleCmd(dir=tmp_path) + + + >>> GitSubmoduleCmd(dir=tmp_path).run(quiet=True) + 'fatal: not a git repository (or any of the parent directories): .git' + + >>> GitSubmoduleCmd(dir=git_local_clone.dir).run(quiet=True) + '' + """ + #: Directory to check out + self.dir: pathlib.Path + if isinstance(dir, pathlib.Path): + self.dir = dir + else: + self.dir = pathlib.Path(dir) + + self.cmd = cmd if isinstance(cmd, Git) else Git(dir=self.dir) + + def __repr__(self) -> str: + return f"" + + def run( + self, + command: Optional[GitSubmoduleCmdCommandLiteral] = None, + local_flags: Optional[list[str]] = None, + *, + quiet: Optional[bool] = None, + cached: Optional[bool] = None, # Only when no command entered and status + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """Wraps `git submodule `_. + + Examples + -------- + >>> GitSubmoduleCmd(dir=git_local_clone.dir).run() + '' + """ + local_flags = local_flags if isinstance(local_flags, list) else [] + if command is not None: + local_flags.insert(0, command) + + if quiet is True: + local_flags.append("--quiet") + if cached is True: + local_flags.append("--cached") + + return self.cmd.run( + ["submodule", *local_flags], + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def init( + self, + *, + path: Optional[Union[list[StrPath], StrPath]] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule init + + Examples + -------- + >>> GitSubmoduleCmd(dir=git_local_clone.dir).init() + '' + """ + local_flags: list[str] = [] + required_flags: list[str] = [] + + if isinstance(path, list): + required_flags.extend(str(pathlib.Path(p).absolute()) for p in path) + elif isinstance(path, pathlib.Path): + required_flags.append(str(pathlib.Path(path).absolute())) + + return self.run( + "init", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def update( + self, + *, + path: Optional[Union[list[StrPath], StrPath]] = None, + init: Optional[bool] = None, + force: Optional[bool] = None, + checkout: Optional[bool] = None, + rebase: Optional[bool] = None, + merge: Optional[bool] = None, + recursive: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """git submodule update + + Examples + -------- + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update() + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(init=True) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(init=True, recursive=True) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update( + ... init=True, filter="blob:none" + ... ) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(force=True) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(checkout=True) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(rebase=True) + '' + >>> GitSubmoduleCmd(dir=git_local_clone.dir).update(merge=True) + '' + """ + local_flags: list[str] = [] + required_flags: list[str] = [] + + if isinstance(path, list): + required_flags.extend(str(pathlib.Path(p).absolute()) for p in path) + elif isinstance(path, pathlib.Path): + required_flags.append(str(pathlib.Path(path).absolute())) + + if init is True: + local_flags.append("--init") + if force is True: + local_flags.append("--force") + + if checkout is True: + local_flags.append("--checkout") + elif rebase is True: + local_flags.append("--rebase") + elif merge is True: + local_flags.append("--merge") + if (filter := kwargs.pop("filter", None)) is not None: + local_flags.append(f"--filter={filter}") + + return self.run( + "update", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + +GitRemoteCommandLiteral = Literal[ + "add", + "rename", + "remove", + "set-branches", + "set-head", + "set-branch", + "get-url", + "set-url", + "set-url --add", + "set-url --delete", + "prune", + "show", + "update", +] + + +class GitRemoteCmd: + def __init__(self, *, dir: StrPath, cmd: Optional[Git] = None) -> None: + r"""Lite, typed, pythonic wrapper for git-remote(1). + + Parameters + ---------- + dir : + Operates as PATH in the corresponding git subcommand. + + Examples + -------- + >>> GitRemoteCmd(dir=tmp_path) + + + >>> GitRemoteCmd(dir=tmp_path).run(verbose=True) + 'fatal: not a git repository (or any of the parent directories): .git' + + >>> GitRemoteCmd(dir=git_local_clone.dir).run(verbose=True) + 'origin\tfile:///...' + """ + #: Directory to check out + self.dir: pathlib.Path + if isinstance(dir, pathlib.Path): + self.dir = dir + else: + self.dir = pathlib.Path(dir) + + self.cmd = cmd if isinstance(cmd, Git) else Git(dir=self.dir) + + def __repr__(self) -> str: + return f"" + + def run( + self, + command: Optional[GitRemoteCommandLiteral] = None, + local_flags: Optional[list[str]] = None, + *, + verbose: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + r"""Wraps `git submodule `_. + + Examples + -------- + >>> GitRemoteCmd(dir=git_local_clone.dir).run() + 'origin' + >>> GitRemoteCmd(dir=git_local_clone.dir).run(verbose=True) + 'origin\tfile:///...' + """ + local_flags = local_flags if isinstance(local_flags, list) else [] + if command is not None: + local_flags.insert(0, command) + + if verbose is True: + local_flags.append("--verbose") + + return self.cmd.run( + ["remote", *local_flags], + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def add( + self, + *, + name: str, + url: str, + fetch: Optional[bool] = None, + track: Optional[str] = None, + master: Optional[str] = None, + mirror: Optional[Union[Literal["push", "fetch"], bool]] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule add + + Examples + -------- + >>> git_remote_repo = create_git_remote_repo() + >>> GitRemoteCmd(dir=git_local_clone.dir).add( + ... name='my_remote', url=f'file://{git_remote_repo}' + ... ) + '' + """ + local_flags: list[str] = [] + required_flags: list[str] = [name, url] + + if mirror is not None: + if isinstance(mirror, str): + assert any(f for f in ["push", "fetch"]) + local_flags.extend(["--mirror", mirror]) + if isinstance(mirror, bool) and mirror: + local_flags.append("--mirror") + return self.run( + "add", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def rename( + self, + *, + old: str, + new: str, + progress: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule rename + + Examples + -------- + >>> git_remote_repo = create_git_remote_repo() + >>> GitRemoteCmd(dir=git_local_clone.dir).rename(old='origin', new='new_name') + '' + >>> GitRemoteCmd(dir=git_local_clone.dir).run() + 'new_name' + """ + local_flags: list[str] = [] + required_flags: list[str] = [old, new] + + if progress is not None: + if progress: + local_flags.append("--progress") + else: + local_flags.append("--no-progress") + return self.run( + "rename", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def remove( + self, + *, + name: str, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule remove + + Examples + -------- + >>> GitRemoteCmd(dir=git_local_clone.dir).remove(name='origin') + '' + >>> GitRemoteCmd(dir=git_local_clone.dir).run() + '' + """ + local_flags: list[str] = [] + required_flags: list[str] = [name] + + return self.run( + "remove", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def show( + self, + *, + name: Optional[str] = None, + verbose: Optional[bool] = None, + no_query_remotes: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule show + + Examples + -------- + >>> GitRemoteCmd(dir=git_local_clone.dir).show() + 'origin' + """ + local_flags: list[str] = [] + required_flags: list[str] = [] + + if name is not None: + required_flags.append(name) + + if verbose is not None: + local_flags.append("--verbose") + + if no_query_remotes is not None or no_query_remotes: + local_flags.append("-n") + + return self.run( + "show", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def prune( + self, + *, + name: str, + dry_run: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule prune + + Examples + -------- + >>> git_remote_repo = create_git_remote_repo() + >>> GitRemoteCmd(dir=git_local_clone.dir).prune(name='origin') + '' + + >>> GitRemoteCmd(dir=git_local_clone.dir).prune(name='origin', dry_run=True) + '' + """ + local_flags: list[str] = [] + required_flags: list[str] = [name] + + if dry_run: + local_flags.append("--dry-run") + + return self.run( + "prune", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def get_url( + self, + *, + name: str, + push: Optional[bool] = None, + all: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule get-url + + Examples + -------- + >>> git_remote_repo = create_git_remote_repo() + >>> GitRemoteCmd(dir=git_local_clone.dir).get_url(name='origin') + 'file:///...' + + >>> GitRemoteCmd(dir=git_local_clone.dir).get_url(name='origin', push=True) + 'file:///...' + + >>> GitRemoteCmd(dir=git_local_clone.dir).get_url(name='origin', all=True) + 'file:///...' + """ + local_flags: list[str] = [] + required_flags: list[str] = [name] + + if push: + local_flags.append("--push") + if all: + local_flags.append("--all") + + return self.run( + "get-url", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def set_url( + self, + *, + name: str, + url: str, + old_url: Optional[str] = None, + push: Optional[bool] = None, + add: Optional[bool] = None, + delete: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git submodule set-url + + Examples + -------- + >>> git_remote_repo = create_git_remote_repo() + >>> GitRemoteCmd(dir=git_local_clone.dir).set_url( + ... name='origin', + ... url='http://localhost' + ... ) + '' + + >>> GitRemoteCmd(dir=git_local_clone.dir).set_url( + ... name='origin', + ... url='http://localhost', + ... push=True + ... ) + '' + + >>> GitRemoteCmd(dir=git_local_clone.dir).set_url( + ... name='origin', + ... url='http://localhost', + ... add=True + ... ) + '' + + >>> current_url = GitRemoteCmd(dir=git_local_clone.dir).get_url(name='origin') + >>> GitRemoteCmd(dir=git_local_clone.dir).set_url( + ... name='origin', + ... url=current_url, + ... delete=True + ... ) + 'fatal: Will not delete all non-push URLs' + + """ + local_flags: list[str] = [] + required_flags: list[str] = [name, url] + if old_url is not None: + required_flags.append(old_url) + + if push: + local_flags.append("--push") + if add: + local_flags.append("--add") + if delete: + local_flags.append("--delete") + + return self.run( + "set-url", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + +GitStashCommandLiteral = Literal[ + "list", + "show", + "save", + "drop", + "branch", + "pop", + "apply", + "push", + "clear", + "create", + "store", +] + + +class GitStashCmd: + def __init__(self, *, dir: StrPath, cmd: Optional[Git] = None) -> None: + """Lite, typed, pythonic wrapper for git-stash(1). + + Parameters + ---------- + dir : + Operates as PATH in the corresponding git subcommand. + + Examples + -------- + >>> GitStashCmd(dir=tmp_path) + + + >>> GitStashCmd(dir=tmp_path).run(quiet=True) + 'fatal: not a git repository (or any of the parent directories): .git' + + >>> GitStashCmd(dir=git_local_clone.dir).run(quiet=True) + '' + """ + #: Directory to check out + self.dir: pathlib.Path + if isinstance(dir, pathlib.Path): + self.dir = dir + else: + self.dir = pathlib.Path(dir) + + self.cmd = cmd if isinstance(cmd, Git) else Git(dir=self.dir) + + def __repr__(self) -> str: + return f"" + + def run( + self, + command: Optional[GitStashCommandLiteral] = None, + local_flags: Optional[list[str]] = None, + *, + quiet: Optional[bool] = None, + cached: Optional[bool] = None, # Only when no command entered and status + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """Wraps `git stash `_. + + Examples + -------- + >>> GitStashCmd(dir=git_local_clone.dir).run() + 'No local changes to save' + """ + local_flags = local_flags if isinstance(local_flags, list) else [] + if command is not None: + local_flags.insert(0, command) + + if quiet is True: + local_flags.append("--quiet") + if cached is True: + local_flags.append("--cached") + + return self.cmd.run( + ["stash", *local_flags], + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def _list( + self, + *, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + ) -> str: + """git stash list + + Examples + -------- + >>> GitStashCmd(dir=git_local_clone.dir)._list() + '' + """ + return self.run( + "list", + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def push( + self, + *, + path: Optional[Union[list[StrPath], StrPath]] = None, + patch: Optional[bool] = None, + staged: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """git stash update + + TODO: Fill-in + + Examples + -------- + >>> GitStashCmd(dir=git_local_clone.dir).push() + 'No local changes to save' + + >>> GitStashCmd(dir=git_local_clone.dir).push(path='.') + 'No local changes to save' + """ + local_flags: list[str] = [] + required_flags: list[str] = [] + + if isinstance(path, list): + required_flags.extend(str(pathlib.Path(p).absolute()) for p in path) + elif isinstance(path, pathlib.Path): + required_flags.append(str(pathlib.Path(path).absolute())) + + if patch is True: + local_flags.append("--patch") + if staged is True: + local_flags.append("--staged") + + return self.run( + "push", + local_flags=local_flags + ["--"] + required_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def pop( + self, + *, + stash: Optional[int] = None, + index: Optional[bool] = None, + quiet: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """git stash pop + + Examples + -------- + >>> GitStashCmd(dir=git_local_clone.dir).pop() + 'No stash entries found.' + + >>> GitStashCmd(dir=git_local_clone.dir).pop(stash=0) + 'error: refs/stash@{0} is not a valid reference' + + >>> GitStashCmd(dir=git_local_clone.dir).pop(stash=1, index=True) + 'error: refs/stash@{1} is not a valid reference' + + >>> GitStashCmd(dir=git_local_clone.dir).pop(stash=1, quiet=True) + 'error: refs/stash@{1} is not a valid reference' + + >>> GitStashCmd(dir=git_local_clone.dir).push(path='.') + 'No local changes to save' + """ + local_flags: list[str] = [] + stash_flags: list[str] = [] + + if stash is not None: + stash_flags.extend(["--", str(stash)]) + + if index is True: + local_flags.append("--index") + if quiet is True: + local_flags.append("--quiet") + + return self.run( + "pop", + local_flags=local_flags + stash_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, + ) + + def save( + self, + *, + message: Optional[str] = None, + staged: Optional[int] = None, + keep_index: Optional[int] = None, + patch: Optional[bool] = None, + include_untracked: Optional[bool] = None, + all: Optional[bool] = None, + quiet: Optional[bool] = None, + # Pass-through to run() + log_in_real_time: bool = False, + check_returncode: Optional[bool] = None, + **kwargs: Any, + ) -> str: + """git stash save + + Examples + -------- + >>> GitStashCmd(dir=git_local_clone.dir).save() + 'No local changes to save' + + >>> GitStashCmd(dir=git_local_clone.dir).save(message="Message") + 'No local changes to save' + """ + local_flags: list[str] = [] + stash_flags: list[str] = [] + + if all is True: + local_flags.append("--all") + if staged is True: + local_flags.append("--staged") + if patch is True: + local_flags.append("--patch") + if include_untracked is True: + local_flags.append("--include-untracked") + if keep_index is True: + local_flags.append("--keep-index") + if quiet is True: + local_flags.append("--quiet") + + if message is not None: + local_flags.extend(["--message", message]) + + return self.run( + "save", + local_flags=local_flags + stash_flags, + check_returncode=check_returncode, + log_in_real_time=log_in_real_time, ) diff --git a/src/libvcs/cmd/hg.py b/src/libvcs/cmd/hg.py index 8f59cf82c..797a6ff9c 100644 --- a/src/libvcs/cmd/hg.py +++ b/src/libvcs/cmd/hg.py @@ -12,7 +12,7 @@ from collections.abc import Sequence from typing import Any, Optional, Union -from libvcs._internal.run import run +from libvcs._internal.run import ProgressCallbackProtocol, run from libvcs._internal.types import StrOrBytesPath, StrPath _CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]] @@ -34,7 +34,14 @@ class HgPagerType(enum.Enum): class Hg: - def __init__(self, *, dir: StrPath) -> None: + progress_callback: Optional[ProgressCallbackProtocol] = None + + def __init__( + self, + *, + dir: StrPath, + progress_callback: Optional[ProgressCallbackProtocol] = None, + ) -> None: """Lite, typed, pythonic wrapper for hg(1). Parameters @@ -54,6 +61,8 @@ def __init__(self, *, dir: StrPath) -> None: else: self.dir = pathlib.Path(dir) + self.progress_callback = progress_callback + def __repr__(self) -> str: return f"" @@ -77,6 +86,7 @@ def run( time: Optional[bool] = None, pager: Optional[HgPagerType] = None, color: Optional[HgColorType] = None, + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """ @@ -125,6 +135,8 @@ def run( ``--pager TYPE`` config : ``--config CONFIG [+]``, ``section.name=value`` + check_returncode : bool, default: ``True`` + Passthrough to :func:`libvcs._internal.run.run()` Examples -------- @@ -150,7 +162,7 @@ def run( if color is not None: cli_args.append(["--color", color]) if verbose is True: - cli_args.append("verbose") + cli_args.append("--verbose") if quiet is True: cli_args.append("--quiet") if debug is True: @@ -168,7 +180,14 @@ def run( if help is True: cli_args.append("--help") - return run(args=cli_args, **kwargs) + if self.progress_callback is not None: + kwargs["callback"] = self.progress_callback + + return run( + args=cli_args, + check_returncode=True if check_returncode is None else check_returncode, + **kwargs, + ) def clone( self, @@ -183,11 +202,22 @@ def clone( pull: Optional[bool] = None, stream: Optional[bool] = None, insecure: Optional[bool] = None, + quiet: Optional[bool] = None, + # Special behavior + make_parents: Optional[bool] = True, + check_returncode: Optional[bool] = None, ) -> str: """Clone a working copy from a mercurial repo. Wraps `hg clone `_. + Parameters + ---------- + make_parents : bool, default: ``True`` + Creates checkout directory (`:attr:`self.dir`) if it doesn't already exist. + check_returncode : bool, default: ``None`` + Passthrough to :meth:`Hg.run` + Examples -------- >>> hg = Hg(dir=tmp_path) @@ -216,6 +246,80 @@ def clone( local_flags.append("--stream") if insecure is True: local_flags.append("--insecure") + if quiet is True: + local_flags.append("--quiet") + + # libvcs special behavior + if make_parents and not self.dir.exists(): + self.dir.mkdir(parents=True) return self.run( - ["clone", *local_flags, "--", *required_flags], check_returncode=False + ["clone", *local_flags, "--", *required_flags], + check_returncode=check_returncode, ) + + def update( + self, + quiet: Optional[bool] = None, + verbose: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = True, + *args: object, + **kwargs: object, + ) -> str: + """Update working directory + + Wraps `hg update `_. + + Examples + -------- + >>> hg = Hg(dir=tmp_path) + >>> hg_remote_repo = create_hg_remote_repo() + >>> hg.clone(url=f'file://{hg_remote_repo}') + 'updating to branch default...1 files updated, 0 files merged, ...' + >>> hg.update() + '0 files updated, 0 files merged, 0 files removed, 0 files unresolved' + """ + local_flags: list[str] = [] + + if quiet: + local_flags.append("--quiet") + if verbose: + local_flags.append("--verbose") + + return self.run(["update", *local_flags], check_returncode=check_returncode) + + def pull( + self, + quiet: Optional[bool] = None, + verbose: Optional[bool] = None, + update: Optional[bool] = None, + # libvcs special behavior + check_returncode: Optional[bool] = True, + *args: object, + **kwargs: object, + ) -> str: + """Update working directory + + Wraps `hg update `_. + + Examples + -------- + >>> hg = Hg(dir=tmp_path) + >>> hg_remote_repo = create_hg_remote_repo() + >>> hg.clone(url=f'file://{hg_remote_repo}') + 'updating to branch default...1 files updated, 0 files merged, ...' + >>> hg.pull() + 'pulling from ...searching for changes...no changes found' + >>> hg.pull(update=True) + 'pulling from ...searching for changes...no changes found' + """ + local_flags: list[str] = [] + + if quiet: + local_flags.append("--quiet") + if verbose: + local_flags.append("--verbose") + if update: + local_flags.append("--update") + + return self.run(["pull", *local_flags], check_returncode=check_returncode) diff --git a/src/libvcs/cmd/svn.py b/src/libvcs/cmd/svn.py index 7a0dfd0b5..b3d1ffed1 100644 --- a/src/libvcs/cmd/svn.py +++ b/src/libvcs/cmd/svn.py @@ -8,9 +8,9 @@ """ import pathlib from collections.abc import Sequence -from typing import Any, Literal, Optional, Union +from typing import Any, List, Literal, Optional, Union -from libvcs._internal.run import run +from libvcs._internal.run import ProgressCallbackProtocol, run from libvcs._internal.types import StrOrBytesPath, StrPath _CMD = Union[StrOrBytesPath, Sequence[StrOrBytesPath]] @@ -20,7 +20,14 @@ class Svn: - def __init__(self, *, dir: StrPath) -> None: + progress_callback: Optional[ProgressCallbackProtocol] = None + + def __init__( + self, + *, + dir: StrPath, + progress_callback: Optional[ProgressCallbackProtocol] = None, + ) -> None: """Lite, typed, pythonic wrapper for svn(1). Parameters @@ -40,6 +47,8 @@ def __init__(self, *, dir: StrPath) -> None: else: self.dir = pathlib.Path(dir) + self.progress_callback = progress_callback + def __repr__(self) -> str: return f"" @@ -55,6 +64,9 @@ def run( trust_server_cert: Optional[bool] = None, config_dir: Optional[pathlib.Path] = None, config_option: Optional[pathlib.Path] = None, + # Special behavior + make_parents: Optional[bool] = True, + check_returncode: Optional[bool] = None, **kwargs: Any, ) -> str: """ @@ -86,6 +98,10 @@ def run( --config-option, ``FILE:SECTION:OPTION=[VALUE]`` cwd : :attr:`libvcs._internal.types.StrOrBytesPath`, optional Defaults to :attr:`~.cwd` + make_parents : bool, default: ``True`` + Creates checkout directory (`:attr:`self.dir`) if it doesn't already exist. + check_returncode : bool, default: ``None`` + Passthrough to :meth:`Svn.run` Examples -------- @@ -117,7 +133,14 @@ def run( if config_option is not None: cli_args.extend(["--config-option", str(config_option)]) - return run(args=cli_args, **kwargs) + if self.progress_callback is not None: + kwargs["callback"] = self.progress_callback + + return run( + args=cli_args, + check_returncode=True if check_returncode is None else check_returncode, + **kwargs, + ) def checkout( self, @@ -127,6 +150,15 @@ def checkout( force: Optional[bool] = None, ignore_externals: Optional[bool] = None, depth: DepthLiteral = None, + quiet: Optional[bool] = None, + username: Optional[str] = None, + password: Optional[str] = None, + no_auth_cache: Optional[bool] = None, + non_interactive: Optional[bool] = True, + trust_server_cert: Optional[bool] = None, + # Special behavior + make_parents: Optional[bool] = True, + check_returncode: Optional[bool] = False, ) -> str: """Check out a working copy from an SVN repo. @@ -144,28 +176,45 @@ def checkout( ignore externals definitions depth : Sparse checkout support, Optional + make_parents : bool, default: ``True`` + Creates checkout directory (`:attr:`self.dir`) if it doesn't already exist. + check_returncode : bool, default: True + Passthrough to :meth:`Svn.run` Examples -------- >>> svn = Svn(dir=tmp_path) >>> svn_remote_repo = create_svn_remote_repo() >>> svn.checkout(url=f'file://{svn_remote_repo}') - 'Checked out revision ...' - >>> svn.checkout(url=f'file://{svn_remote_repo}', revision=1) - 'svn: E160006: No such revision 1...' + '...Checked out revision ...' + >>> svn.checkout(url=f'file://{svn_remote_repo}', revision=10) + 'svn: E160006: No such revision 10...' """ local_flags: list[str] = [url, str(self.dir)] if revision is not None: - local_flags.append(f"--revision={revision}") + local_flags.extend(["--revision", str(revision)]) if depth is not None: - local_flags.append(depth) + local_flags.extend(["--depth", depth]) if force is True: local_flags.append("--force") if ignore_externals is True: local_flags.append("--ignore-externals") - return self.run(["checkout", *local_flags], check_returncode=False) + # libvcs special behavior + if make_parents and not self.dir.exists(): + self.dir.mkdir(parents=True) + + return self.run( + ["checkout", *local_flags], + quiet=quiet, + username=username, + password=password, + no_auth_cache=no_auth_cache, + non_interactive=non_interactive, + trust_server_cert=trust_server_cert, + check_returncode=check_returncode, + ) def add( self, @@ -302,8 +351,9 @@ def blame( Examples -------- >>> svn = Svn(dir=tmp_path) - >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') - 'Checked out revision ...' + >>> repo = create_svn_remote_repo() + >>> svn.checkout(url=f'file://{repo}') + '...Checked out revision ...' >>> new_file = tmp_path / 'new.txt' >>> new_file.write_text('example text', encoding="utf-8") 12 @@ -312,12 +362,12 @@ def blame( >>> svn.commit(path=new_file, message='My new commit') '...' >>> svn.blame('new.txt') - '1 ... example text' + '4 ... example text' """ local_flags: list[str] = [str(target)] if revision is not None: - local_flags.append(f"--revision={revision}") + local_flags.extend(["--revision", revision]) if verbose is True: local_flags.append("--verbose") if use_merge_history is True: @@ -412,7 +462,7 @@ def commit( >>> svn.add(path=new_file) 'A new.txt' >>> svn.commit(path=new_file, message='My new commit') - 'Adding new.txt...Transmitting file data...Committed revision 1.' + 'Adding new.txt...Transmitting file data...Committed revision 4.' """ local_flags: list[str] = [] @@ -511,19 +561,62 @@ def import_(self, *args: Any, **kwargs: Any) -> str: return self.run(["import", *local_flags]) - def info(self, *args: Any, **kwargs: Any) -> str: + def info( + self, + target: Optional[StrPath] = None, + targets: Optional[Union[list[StrPath], StrPath]] = None, + changelist: Optional[List[str]] = None, + revision: Optional[str] = None, + depth: DepthLiteral = None, + incremental: Optional[bool] = None, + recursive: Optional[bool] = None, + xml: Optional[bool] = None, + *args: Any, + **kwargs: Any, + ) -> str: """ Wraps `svn info `_. Parameters ---------- + targets : pathlib.Path + `--targets ARG`: contents of file ARG as additional args + xml : bool + `--xml`, xml output + revision : Union[RevisionLiteral, str] + Number, '{ DATE }', 'HEAD', 'BASE', 'COMMITTED', 'PREV' + depth : + `--depth ARG`, Sparse checkout support, Optional + incremental : bool + `--incremental`, give output suitable for concatenation """ local_flags: list[str] = [*args] + if isinstance(target, pathlib.Path): + local_flags.append(str(target.absolute())) + elif isinstance(target, str): + local_flags.append(target) + + if revision is not None: + local_flags.extend(["--revision", revision]) + if targets is not None: + if isinstance(targets, Sequence): + local_flags.extend(["--targets", *[str(t) for t in targets]]) + else: + local_flags.extend(["--targets", str(targets)]) + if changelist is not None: + local_flags.extend(["--changelist", *changelist]) + if recursive is True: + local_flags.append("--recursive") + if xml is True: + local_flags.append("--xml") + if incremental is True: + local_flags.append("--incremental") + return self.run(["info", *local_flags]) - def list(self, *args: Any, **kwargs: Any) -> str: + def _list(self, *args: Any, **kwargs: Any) -> str: """ Wraps `svn list `_ (ls). @@ -535,15 +628,34 @@ def list(self, *args: Any, **kwargs: Any) -> str: return self.run(["list", *local_flags]) - def lock(self, *args: Any, **kwargs: Any) -> str: + def lock( + self, + targets: Optional[pathlib.Path] = None, + force: Optional[bool] = None, + **kwargs: Any, + ) -> str: """ Wraps `svn lock `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.lock(targets='samplepickle') + "'samplepickle' locked by user '...'." """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + + if targets is not None: + if isinstance(targets, str): + local_flags.extend([str(targets)]) + elif isinstance(targets, Sequence): + local_flags.extend([*[str(t) for t in targets]]) + + if force: + local_flags.append("--force") return self.run(["lock", *local_flags]) @@ -668,63 +780,221 @@ def proplist(self, *args: Any, **kwargs: Any) -> str: return self.run(["proplist", *local_flags]) - def propset(self, *args: Any, **kwargs: Any) -> str: + def propset( + self, + name: str, + path: Optional[StrPath] = None, + value: Optional[str] = None, + value_path: Optional[StrPath] = None, + target: Optional[StrOrBytesPath] = None, + *args: Any, + **kwargs: Any, + ) -> str: """ Wraps `svn propset `_ (pset, ps). Parameters ---------- + name : str + propname + value_path : + VALFILE + + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.propset(name="my_prop", value="value", path=".") + "property 'my_prop' set on '.'" """ - local_flags: list[str] = [*args] + local_flags: list[str] = [name, *args] + + if value is not None: + local_flags.append(value) + elif isinstance(value_path, pathlib.Path): + local_flags.extend(["--file", str(pathlib.Path(value_path).absolute())]) + else: + raise ValueError("Must enter a value or value_path") + + if path is not None: + if isinstance(path, (str, pathlib.Path)): + local_flags.append(str(pathlib.Path(path).absolute())) + elif target is not None: + local_flags.append(str(target)) return self.run(["propset", *local_flags]) - def relocate(self, *args: Any, **kwargs: Any) -> str: + def relocate(self, *, to_path: StrPath, **kwargs: Any) -> str: """ Wraps `svn relocate `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path / 'initial_place') + >>> repo_path = create_svn_remote_repo() + >>> svn.checkout(url=repo_path.as_uri()) + '...Checked out revision ...' + >>> new_place = repo_path.rename(tmp_path / 'new_place') + >>> svn.relocate(to_path=new_place.absolute().as_uri()) + '' """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + required_flags: list[str] = [] + + if isinstance(to_path, str): + if to_path.startswith("file://"): + required_flags.append(to_path) + else: + required_flags.append(str(pathlib.Path(to_path).absolute().as_uri())) + elif isinstance(to_path, pathlib.Path): + required_flags.append(str(to_path.absolute().as_uri())) - return self.run(["relocate", *local_flags]) + return self.run(["relocate", *local_flags, *required_flags]) - def resolve(self, *args: Any, **kwargs: Any) -> str: + def resolve( + self, + path: Union[list[pathlib.Path], pathlib.Path], + targets: Optional[pathlib.Path] = None, + depth: DepthLiteral = None, + force: Optional[bool] = None, + *args: Any, + **kwargs: Any, + ) -> str: """ Wraps `svn resolve `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.resolve(path='.') + '' """ local_flags: list[str] = [*args] + if isinstance(path, list): + local_flags.extend(str(p.absolute()) for p in path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) + + if targets is not None: + if isinstance(targets, Sequence): + local_flags.extend(["--targets", *[str(t) for t in targets]]) + else: + local_flags.extend(["--targets", str(targets)]) + + if depth is not None: + local_flags.extend(["--depth", depth]) + if force is not None: + local_flags.append("--force") + return self.run(["resolve", *local_flags]) - def resolved(self, *args: Any, **kwargs: Any) -> str: + def resolved( + self, + *, + path: Union[list[pathlib.Path], pathlib.Path], + targets: Optional[pathlib.Path] = None, + depth: DepthLiteral = None, + force: Optional[bool] = None, + **kwargs: Any, + ) -> str: """ Wraps `svn resolved `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.resolved(path='.') + '' """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + + if isinstance(path, list): + local_flags.extend(str(p.absolute()) for p in path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) + + if path is not None: + if isinstance(path, str): + local_flags.append(str(pathlib.Path(path).absolute())) + if isinstance(path, list): + local_flags.extend(str(p.absolute()) for p in path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) + elif targets is not None: + if isinstance(targets, Sequence): + local_flags.extend(["--targets", *[str(t) for t in targets]]) + else: + local_flags.extend(["--targets", str(targets)]) + + if depth is not None: + local_flags.extend(["--depth", depth]) + if force is not None: + local_flags.append("--force") return self.run(["resolved", *local_flags]) - def revert(self, *args: Any, **kwargs: Any) -> str: + def revert( + self, + *, + path: Union[list[pathlib.Path], pathlib.Path], + targets: Optional[pathlib.Path] = None, + depth: DepthLiteral = None, + force: Optional[bool] = None, + **kwargs: Any, + ) -> str: """ Wraps `svn revert `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> new_file = tmp_path / 'new.txt' + >>> new_file.write_text('example text', encoding="utf-8") + 12 + >>> svn.add(path=new_file) + 'A new.txt' + >>> svn.commit(path=new_file, message='My new commit') + '...' + >>> svn.revert(path=new_file) + '' """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + + if isinstance(path, list): + local_flags.extend(str(p.absolute()) for p in path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) + + if path is not None: + if isinstance(path, str): + local_flags.append(str(pathlib.Path(path).absolute())) + if isinstance(path, list): + local_flags.extend(str(p.absolute()) for p in path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) + elif targets is not None: + if isinstance(targets, Sequence): + local_flags.extend(["--targets", *[str(t) for t in targets]]) + else: + local_flags.extend(["--targets", str(targets)]) + + if depth is not None: + local_flags.extend(["--depth", depth]) + if force is not None: + local_flags.append("--force") return self.run(["revert", *local_flags]) @@ -733,47 +1003,143 @@ def status(self, *args: Any, **kwargs: Any) -> str: Wraps `svn status `_ (stat, st). - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.status() + '' """ local_flags: list[str] = [*args] return self.run(["status", *local_flags]) - def switch(self, *args: Any, **kwargs: Any) -> str: + def switch( + self, + *, + to_path: StrPath, + path: StrPath, + ignore_ancestry: Optional[bool] = None, + **kwargs: Any, + ) -> str: """ Wraps `svn switch `_ (sw). - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path / 'initial_place') + >>> repo_path = create_svn_remote_repo() + >>> svn.checkout(url=(repo_path / 'sampledir').as_uri()) + '...Checked out revision ...' + >>> other_dir = repo_path / 'otherdir' + >>> svn.switch(to_path=other_dir.as_uri(), path='.', ignore_ancestry=True) + 'D...Updated to revision...' """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + required_flags: list[str] = [] + + if isinstance(to_path, str): + if to_path.startswith("file://"): + local_flags.append(to_path) + else: + local_flags.append(str(pathlib.Path(to_path).absolute().as_uri())) + elif isinstance(to_path, pathlib.Path): + local_flags.append(str(to_path.absolute().as_uri())) + + if path is not None: + if isinstance(path, str): + local_flags.append(path) + elif isinstance(path, pathlib.Path): + local_flags.append(str(path.absolute())) - return self.run(["switch", *local_flags]) + if ignore_ancestry: + local_flags.append("--ignore-ancestry") - def unlock(self, *args: Any, **kwargs: Any) -> str: + return self.run(["switch", *local_flags, *required_flags]) + + def unlock( + self, + targets: Optional[pathlib.Path] = None, + force: Optional[bool] = None, + **kwargs: Any, + ) -> str: """ Wraps `svn unlock `_. - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.lock(targets='samplepickle') + "'samplepickle' locked by user '...'." + >>> svn.unlock(targets='samplepickle') + "'samplepickle' unlocked." """ - local_flags: list[str] = [*args] + local_flags: list[str] = [] + + if targets is not None: + if isinstance(targets, str): + local_flags.extend([str(targets)]) + elif isinstance(targets, Sequence): + local_flags.extend([*[str(t) for t in targets]]) + + if force: + local_flags.append("--force") return self.run(["unlock", *local_flags]) - def update(self, *args: Any, **kwargs: Any) -> str: + def update( + self, + accept: Optional[str] = None, + changelist: Optional[List[str]] = None, + diff3_cmd: Optional[str] = None, + editor_cmd: Optional[str] = None, + force: Optional[bool] = None, + ignore_externals: Optional[bool] = None, + parents: Optional[bool] = None, + quiet: Optional[bool] = None, + revision: Optional[str] = None, + set_depth: Optional[str] = None, + *args: Any, + **kwargs: Any, + ) -> str: """ Wraps `svn update `_ (up). - Parameters - ---------- + Examples + -------- + >>> svn = Svn(dir=tmp_path) + >>> svn.checkout(url=f'file://{create_svn_remote_repo()}') + '...Checked out revision ...' + >>> svn.update() + "Updating ..." """ local_flags: list[str] = [*args] + if revision is not None: + local_flags.extend(["--revision", revision]) + if diff3_cmd is not None: + local_flags.extend(["--diff3-cmd", diff3_cmd]) + if editor_cmd is not None: + local_flags.extend(["--editor-cmd", editor_cmd]) + if set_depth is not None: + local_flags.extend(["--set-depth", set_depth]) + if changelist is not None: + local_flags.extend(["--changelist", *changelist]) + if force is True: + local_flags.append("--force") + if quiet is True: + local_flags.append("--quiet") + if parents is True: + local_flags.append("--parents") + if ignore_externals is True: + local_flags.append("--ignore-externals") + return self.run(["update", *local_flags]) def upgrade(self, *args: Any, **kwargs: Any) -> str: diff --git a/src/libvcs/data/repotest.dump b/src/libvcs/data/repotest.dump new file mode 100644 index 000000000..a1d92d550 --- /dev/null +++ b/src/libvcs/data/repotest.dump @@ -0,0 +1,227 @@ +SVN-fs-dump-format-version: 2 + +UUID: 876a30f4-1eed-0310-aeb7-ae314d1e5934 + +Revision-number: 0 +Prop-content-length: 56 +Content-length: 56 + +K 8 +svn:date +V 27 +2005-01-07T23:55:31.755989Z +PROPS-END + +Revision-number: 1 +Prop-content-length: 118 +Content-length: 118 + +K 7 +svn:log +V 20 +testrepo setup rev 1 +K 10 +svn:author +V 3 +hpk +K 8 +svn:date +V 27 +2005-01-07T23:55:37.815386Z +PROPS-END + +Node-path: execfile +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 4 +Text-content-md5: d4b5bc61e16310f08c5d11866eba0a22 +Content-length: 14 + +PROPS-END +x=42 + +Node-path: otherdir +Node-kind: dir +Node-action: add +Prop-content-length: 10 +Content-length: 10 + +PROPS-END + + +Node-path: otherdir/__init__.py +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 0 +Text-content-md5: d41d8cd98f00b204e9800998ecf8427e +Content-length: 10 + +PROPS-END + + +Node-path: otherdir/a.py +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 30 +Text-content-md5: 247c7daeb2ee5dcab0aba7bd12bad665 +Content-length: 40 + +PROPS-END +from b import stuff as result + + +Node-path: otherdir/b.py +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 15 +Text-content-md5: c1b13503469a7711306d03a4b0721bc6 +Content-length: 25 + +PROPS-END +stuff="got it" + + +Node-path: otherdir/c.py +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 75 +Text-content-md5: 250cdb6b5df68536152c681f48297569 +Content-length: 85 + +PROPS-END +import py; py.magic.autopath() +import otherdir.a +value = otherdir.a.result + + +Node-path: otherdir/d.py +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 72 +Text-content-md5: 940c9c621e7b198e081459642c37f5a7 +Content-length: 82 + +PROPS-END +import py; py.magic.autopath() +from otherdir import a +value2 = a.result + + +Node-path: sampledir +Node-kind: dir +Node-action: add +Prop-content-length: 10 +Content-length: 10 + +PROPS-END + + +Node-path: sampledir/otherfile +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 0 +Text-content-md5: d41d8cd98f00b204e9800998ecf8427e +Content-length: 10 + +PROPS-END + + +Node-path: samplefile +Node-kind: file +Node-action: add +Prop-content-length: 40 +Text-content-length: 11 +Text-content-md5: 9225ac28b32156979ab6482b8bb5fb8c +Content-length: 51 + +K 13 +svn:eol-style +V 6 +native +PROPS-END +samplefile + + +Node-path: samplepickle +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 56 +Text-content-md5: 719d85c1329a33134bb98f56b756c545 +Content-length: 66 + +PROPS-END +(dp1 +S'answer' +p2 +I42 +sI1 +I2 +sS'hello' +p3 +S'world' +p4 +s. + +Revision-number: 2 +Prop-content-length: 108 +Content-length: 108 + +K 7 +svn:log +V 10 +second rev +K 10 +svn:author +V 3 +hpk +K 8 +svn:date +V 27 +2005-01-07T23:55:39.223202Z +PROPS-END + +Node-path: anotherfile +Node-kind: file +Node-action: add +Prop-content-length: 10 +Text-content-length: 5 +Text-content-md5: 5d41402abc4b2a76b9719d911017c592 +Content-length: 15 + +PROPS-END +hello + +Revision-number: 3 +Prop-content-length: 106 +Content-length: 106 + +K 7 +svn:log +V 9 +third rev +K 10 +svn:author +V 3 +hpk +K 8 +svn:date +V 27 +2005-01-07T23:55:41.556642Z +PROPS-END + +Node-path: anotherfile +Node-kind: file +Node-action: change +Text-content-length: 5 +Text-content-md5: 7d793037a0760186574b0282f2f435e7 +Content-length: 5 + +world diff --git a/src/libvcs/pytest_plugin.py b/src/libvcs/pytest_plugin.py index 303171547..658879f10 100644 --- a/src/libvcs/pytest_plugin.py +++ b/src/libvcs/pytest_plugin.py @@ -276,7 +276,10 @@ def _create_svn_remote_repo( init_cmd_args = [] remote_repo_path = remote_repos_path / remote_repo_name - run(["svnadmin", "create", remote_repo_path, *init_cmd_args]) + run(["svnadmin", "create", str(remote_repo_path), *init_cmd_args]) + + assert remote_repo_path.exists() + assert remote_repo_path.is_dir() if remote_repo_post_init is not None and callable(remote_repo_post_init): remote_repo_post_init(remote_repo_path=remote_repo_path) @@ -284,6 +287,23 @@ def _create_svn_remote_repo( return remote_repo_path +def svn_remote_repo_single_commit_post_init(remote_repo_path: pathlib.Path) -> None: + assert remote_repo_path.exists() + repo_dumpfile = pathlib.Path(__file__).parent / "data" / "repotest.dump" + run( + " ".join( + [ + "svnadmin", + "load", + str(remote_repo_path), + "<", + str(repo_dumpfile), + ] + ), + shell=True, + ) + + @pytest.fixture @skip_if_svn_missing def create_svn_remote_repo( @@ -313,10 +333,9 @@ def fn( @skip_if_svn_missing def svn_remote_repo(remote_repos_path: pathlib.Path) -> pathlib.Path: """Pre-made. Local file:// based SVN server.""" - svn_repo_name = "svn_server_dir" remote_repo_path = _create_svn_remote_repo( remote_repos_path=remote_repos_path, - remote_repo_name=svn_repo_name, + remote_repo_name="svn_server_dir", remote_repo_post_init=None, ) @@ -456,7 +475,11 @@ def add_doctest_fixtures( doctest_namespace["create_git_remote_repo_bare"] = create_git_remote_repo doctest_namespace["git_local_clone"] = git_repo if shutil.which("svn"): - doctest_namespace["create_svn_remote_repo"] = create_svn_remote_repo + doctest_namespace["create_svn_remote_repo_bare"] = create_svn_remote_repo + doctest_namespace["create_svn_remote_repo"] = functools.partial( + create_svn_remote_repo, + remote_repo_post_init=svn_remote_repo_single_commit_post_init, + ) if shutil.which("hg"): doctest_namespace["create_hg_remote_repo_bare"] = create_hg_remote_repo doctest_namespace["create_hg_remote_repo"] = functools.partial( diff --git a/src/libvcs/sync/git.py b/src/libvcs/sync/git.py index 3c019e2ad..2571566c8 100644 --- a/src/libvcs/sync/git.py +++ b/src/libvcs/sync/git.py @@ -21,7 +21,8 @@ from typing import Any, Optional, Union from urllib import parse as urlparse -from libvcs._internal.types import StrOrBytesPath, StrPath +from libvcs._internal.types import StrPath +from libvcs.cmd.git import Git from libvcs.sync.base import ( BaseSync, VCSLocation, @@ -141,6 +142,7 @@ def convert_pip_url(pip_url: str) -> VCSLocation: class GitSync(BaseSync): bin_name = "git" schemes = ("git+http", "git+https", "git+file") + cmd: Git _remotes: GitSyncRemoteDict def __init__( @@ -231,6 +233,8 @@ def __init__( ) super().__init__(url=url, dir=dir, **kwargs) + self.cmd = Git(dir=dir, progress_callback=self.progress_callback) + origin = ( self._remotes.get("origin") if "origin" in self._remotes @@ -250,7 +254,7 @@ def from_pip_url(cls, pip_url: str, **kwargs: Any) -> "GitSync": def get_revision(self) -> str: """Return current revision. Initial repositories return 'initial'.""" try: - return self.run(["rev-parse", "--verify", "HEAD"]) + return self.cmd.rev_parse(verify=True, args="HEAD", check_returncode=True) except exc.CommandError: return "initial" @@ -300,20 +304,25 @@ def obtain(self, *args: Any, **kwargs: Any) -> None: url = self.url - cmd: list[StrOrBytesPath] = ["clone", "--progress"] - if self.git_shallow: - cmd.extend(["--depth", "1"]) - if self.tls_verify: - cmd.extend(["-c", "http.sslVerify=false"]) - cmd.extend([url, self.dir]) - self.log.info("Cloning.") - self.run(cmd, log_in_real_time=True) + # todo: log_in_real_time + self.cmd.clone( + url=url, + progress=True, + depth=1 if self.git_shallow else None, + config={"http.sslVerify": False} if self.tls_verify else None, + log_in_real_time=True, + ) self.log.info("Initializing submodules.") - self.run(["submodule", "init"], log_in_real_time=True) - cmd = ["submodule", "update", "--recursive", "--init"] - self.run(cmd, log_in_real_time=True) + self.cmd.submodule.init( + log_in_real_time=True, + ) + self.cmd.submodule.update( + init=True, + recursive=True, + log_in_real_time=True, + ) self.set_remotes(overwrite=True) @@ -333,7 +342,7 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N if not git_tag: self.log.debug("No git revision set, defaulting to origin/master") - symref = self.run(["symbolic-ref", "--short", "HEAD"]) + symref = self.cmd.symbolic_ref(name="HEAD", short=True) if symref: git_tag = symref.rstrip() else: @@ -344,7 +353,9 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N # Get head sha try: - head_sha = self.run(["rev-list", "--max-count=1", "HEAD"]) + head_sha = self.cmd.rev_list( + commit="HEAD", max_count=1, check_returncode=True + ) except exc.CommandError: self.log.error("Failed to get the hash for HEAD") return @@ -353,7 +364,7 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N # If a remote ref is asked for, which can possibly move around, # we must always do a fetch and checkout. - show_ref_output = self.run(["show-ref", git_tag], check_returncode=False) + show_ref_output = self.cmd.show_ref(pattern=git_tag, check_returncode=False) self.log.debug("show_ref_output: %s" % show_ref_output) is_remote_ref = "remotes" in show_ref_output self.log.debug("is_remote_ref: %s" % is_remote_ref) @@ -381,13 +392,11 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N # been fetched yet). try: error_code = 0 - tag_sha = self.run( - [ - "rev-list", - "--max-count=1", - git_remote_name + "/" + git_tag if is_remote_ref else git_tag, - ] + tag_sha = self.cmd.rev_list( + commit=git_remote_name + "/" + git_tag if is_remote_ref else git_tag, + max_count=1, ) + except exc.CommandError as e: error_code = e.returncode if e.returncode is not None else 0 tag_sha = "" @@ -400,7 +409,7 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N return try: - process = self.run(["fetch"], log_in_real_time=True) + process = self.cmd.fetch(log_in_real_time=True, check_returncode=True) except exc.CommandError: self.log.error("Failed to fetch repository '%s'" % url) return @@ -408,7 +417,7 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N if is_remote_ref: # Check if stash is needed try: - process = self.run(["status", "--porcelain", "--untracked-files=no"]) + process = self.cmd.status(porcelain=True, untracked_files="no") except exc.CommandError: self.log.error("Failed to get the status") return @@ -420,28 +429,28 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N # If Git < 1.7.6, uses --quiet --all git_stash_save_options = "--quiet" try: - process = self.run(["stash", "save", git_stash_save_options]) + process = self.cmd.stash.save(message=git_stash_save_options) except exc.CommandError: self.log.error("Failed to stash changes") # Checkout the remote branch try: - process = self.run(["checkout", git_tag]) + process = self.cmd.checkout(branch=git_tag) except exc.CommandError: self.log.error("Failed to checkout tag: '%s'" % git_tag) return # Rebase changes from the remote branch try: - process = self.run(["rebase", git_remote_name + "/" + git_tag]) + process = self.cmd.rebase(upstream=git_remote_name + "/" + git_tag) except exc.CommandError as e: if any(msg in str(e) for msg in ["invalid_upstream", "Aborting"]): self.log.error(e) else: # Rebase failed: Restore previous state. - self.run(["rebase", "--abort"]) + self.cmd.rebase(abort=True) if need_stash: - self.run(["stash", "pop", "--index", "--quiet"]) + self.cmd.stash.pop(index=True, quiet=True) self.log.error( "\nFailed to rebase in: '%s'.\n" @@ -451,16 +460,16 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N if need_stash: try: - process = self.run(["stash", "pop", "--index", "--quiet"]) + process = self.cmd.stash.pop(index=True, quiet=True) except exc.CommandError: # Stash pop --index failed: Try again dropping the index - self.run(["reset", "--hard", "--quiet"]) + self.cmd.reset(hard=True, quiet=True) try: - process = self.run(["stash", "pop", "--quiet"]) + process = self.cmd.stash.pop(quiet=True) except exc.CommandError: # Stash pop failed: Restore previous state. - self.run(["reset", "--hard", "--quiet", head_sha]) - self.run(["stash", "pop", "--index", "--quiet"]) + self.cmd.reset(pathspec=head_sha, hard=True, quiet=True) + self.cmd.stash.pop(index=True, quiet=True) self.log.error( "\nFailed to rebase in: '%s'.\n" "You will have to resolve the " @@ -470,13 +479,12 @@ def update_repo(self, set_remotes: bool = False, *args: Any, **kwargs: Any) -> N else: try: - process = self.run(["checkout", git_tag]) + process = self.cmd.checkout(branch=git_tag) except exc.CommandError: self.log.error("Failed to checkout tag: '%s'" % git_tag) return - cmd = ["submodule", "update", "--recursive", "--init"] - self.run(cmd, log_in_real_time=True) + self.cmd.submodule.update(recursive=True, init=True, log_in_real_time=True) def remotes(self) -> GitSyncRemoteDict: """Return remotes like git remote -v. @@ -492,7 +500,7 @@ def remotes(self) -> GitSyncRemoteDict: """ remotes = {} - cmd = self.run(["remote"]) + cmd = self.cmd.remote.run() ret: filter[str] = filter(None, cmd.split("\n")) for remote_name in ret: @@ -515,7 +523,9 @@ def remote(self, name: str, **kwargs: Any) -> Optional[GitRemote]: """ try: - ret = self.run(["remote", "show", "-n", name]) + ret = self.cmd.remote.show( + name=name, no_query_remotes=True, log_in_real_time=True + ) lines = ret.split("\n") remote_fetch_url = lines[1].replace("Fetch URL: ", "").strip() remote_push_url = lines[2].replace("Push URL: ", "").strip() @@ -545,9 +555,9 @@ def set_remote( url = self.chomp_protocol(url) if self.remote(name) and overwrite: - self.run(["remote", "set-url", name, url]) + self.cmd.remote.set_url(name=name, url=url, check_returncode=True) else: - self.run(["remote", "add", name, url]) + self.cmd.remote.add(name=name, url=url, check_returncode=True) remote = self.remote(name=name) if remote is None: @@ -590,7 +600,7 @@ def get_git_version(self) -> str: git version """ VERSION_PFX = "git version " - version = self.run(["version"]) + version = self.cmd.version() if version.startswith(VERSION_PFX): version = version[len(VERSION_PFX) :].split()[0] else: @@ -622,7 +632,9 @@ def status(self) -> GitStatus: branch_behind='0'\ ) """ - return GitStatus.from_stdout(self.run(["status", "-sb", "--porcelain=2"])) + return GitStatus.from_stdout( + self.cmd.status(short=True, branch=True, porcelain="2") + ) def get_current_remote_name(self) -> str: """Retrieve name of the remote / upstream of currently checked out branch. diff --git a/src/libvcs/sync/hg.py b/src/libvcs/sync/hg.py index 2be4b3b38..5878769be 100644 --- a/src/libvcs/sync/hg.py +++ b/src/libvcs/sync/hg.py @@ -12,6 +12,9 @@ import pathlib from typing import Any +from libvcs._internal.types import StrPath +from libvcs.cmd.hg import Hg + from .base import BaseSync logger = logging.getLogger(__name__) @@ -20,23 +23,44 @@ class HgSync(BaseSync): bin_name = "hg" schemes = ("hg", "hg+http", "hg+https", "hg+file") + cmd: Hg - def obtain(self, *args: Any, **kwargs: Any) -> None: - self.ensure_dir() + def __init__( + self, + *, + url: str, + dir: StrPath, + **kwargs: Any, + ) -> None: + """A hg repository. + + Parameters + ---------- + url : str + URL in subversion repository + """ + super().__init__(url=url, dir=dir, **kwargs) - # Double hyphens between [OPTION]... -- SOURCE [DEST] prevent command injections - # via aliases - self.run(["clone", "--noupdate", "-q", "--", self.url, str(self.dir)]) - self.run(["update", "-q"]) + self.cmd = Hg(dir=dir, progress_callback=self.progress_callback) + + def obtain(self, *args: Any, **kwargs: Any) -> None: + self.cmd.clone( + no_update=True, + quiet=True, + url=self.url, + ) + self.cmd.update( + quiet=True, + check_returncode=True, + ) def get_revision(self) -> str: return self.run(["parents", "--template={rev}"]) def update_repo(self, *args: Any, **kwargs: Any) -> None: - self.ensure_dir() if not pathlib.Path(self.dir / ".hg").exists(): self.obtain() self.update_repo() else: - self.run(["update"]) - self.run(["pull", "-u"]) + self.cmd.update() + self.cmd.pull(update=True) diff --git a/src/libvcs/sync/svn.py b/src/libvcs/sync/svn.py index 5ac270321..76b1a1091 100644 --- a/src/libvcs/sync/svn.py +++ b/src/libvcs/sync/svn.py @@ -9,39 +9,35 @@ The following are pypa/pip (MIT license): - - [`SvnSync.convert_pip_url`](libvcs.svn.SvnSync.convert_pip_url) - [`SvnSync.get_url`](libvcs.svn.SvnSync.get_url) - [`SvnSync.get_revision`](libvcs.svn.SvnSync.get_revision) - - [`get_rev_options`](libvcs.svn.get_rev_options) """ # NOQA: E5 import logging import os import pathlib import re from typing import Any, Optional -from urllib import parse as urlparse -from libvcs._internal.run import run -from libvcs._internal.types import StrOrBytesPath, StrPath +from libvcs._internal.types import StrPath +from libvcs.cmd.svn import Svn -from .base import BaseSync, VCSLocation, convert_pip_url as base_convert_pip_url +from .base import BaseSync logger = logging.getLogger(__name__) -def convert_pip_url(pip_url: str) -> VCSLocation: - # hotfix the URL scheme after removing svn+ from svn+ssh:// re-add it - url, rev = base_convert_pip_url(pip_url) - if url.startswith("ssh://"): - url = "svn+" + url - return VCSLocation(url=url, rev=rev) - - class SvnSync(BaseSync): bin_name = "svn" schemes = ("svn", "svn+ssh", "svn+http", "svn+https", "svn+svn") - - def __init__(self, *, url: str, dir: StrPath, **kwargs: Any) -> None: + cmd: Svn + + def __init__( + self, + *, + url: str, + dir: StrPath, + **kwargs: Any, + ) -> None: """A svn repository. Parameters @@ -49,21 +45,26 @@ def __init__(self, *, url: str, dir: StrPath, **kwargs: Any) -> None: url : str URL in subversion repository - svn_username : str, optional + username : str, optional username to use for checkout and update - svn_password : str, optional + password : str, optional password to use for checkout and update svn_trust_cert : bool trust the Subversion server site certificate, default False """ - if "svn_trust_cert" not in kwargs: - self.svn_trust_cert = False + self.svn_trust_cert = kwargs.pop("svn_trust_cert", False) + + self.username = kwargs.get("username") + self.password = kwargs.get("password") self.rev = kwargs.get("rev") + super().__init__(url=url, dir=dir, **kwargs) + self.cmd = Svn(dir=dir, progress_callback=self.progress_callback) + def _user_pw_args(self) -> list[Any]: args = [] for param_name in ["svn_username", "svn_password"]: @@ -72,23 +73,25 @@ def _user_pw_args(self) -> list[Any]: return args def obtain(self, quiet: Optional[bool] = None, *args: Any, **kwargs: Any) -> None: - self.ensure_dir() - url, rev = self.url, self.rev - cmd: list[StrOrBytesPath] = ["checkout", "-q", url, "--non-interactive"] + if rev is not None: + kwargs["revision"] = rev if self.svn_trust_cert: - cmd.append("--trust-server-cert") - cmd.extend(self._user_pw_args()) - cmd.extend(get_rev_options(url, rev)) - cmd.append(self.dir) - - self.run(cmd) + kwargs["trust_server_cert"] = True + self.cmd.checkout( + url=url, + username=self.username, + password=self.password, + non_interactive=True, + quiet=True, + check_returncode=True, + **kwargs, + ) def get_revision_file(self, location: str) -> int: """Return revision for a file.""" - - current_rev = self.run(["info", location]) + current_rev = self.cmd.info(location) _INI_RE = re.compile(r"^([^:]+):\s+(\S.*)$", re.M) @@ -133,17 +136,15 @@ def update_repo( ) -> None: self.ensure_dir() if pathlib.Path(self.dir / ".svn").exists(): - if dest is None: - dest = str(self.dir) - - url, rev = self.url, self.rev - - cmd = ["update"] - cmd.extend(self._user_pw_args()) - cmd.extend(get_rev_options(url, rev)) - cmd.append(dest) - - self.run(cmd) + self.cmd.checkout( + url=self.url, + username=self.username, + password=self.password, + non_interactive=True, + quiet=True, + check_returncode=True, + **kwargs, + ) else: self.obtain() self.update_repo() @@ -182,8 +183,8 @@ def _get_svn_url_rev(cls, location: str) -> tuple[Optional[str], int]: # We don't need to worry about making sure interactive mode # is being used to prompt for passwords, because passwords # are only potentially needed for remote server requests. - xml = run( - ["svn", "info", "--xml", location], + xml = Svn(dir=pathlib.Path(location).parent).info( + target=pathlib.Path(location), xml=True ) match = _svn_info_xml_url_re.search(xml) assert match is not None @@ -198,32 +199,3 @@ def _get_svn_url_rev(cls, location: str) -> tuple[Optional[str], int]: rev = 0 return url, rev - - -def get_rev_options(url: str, rev: None) -> list[Any]: - """Return revision options. From pip pip.vcs.subversion.""" - if rev: - rev_options = ["-r", rev] - else: - rev_options = [] - - r = urlparse.urlsplit(url) - if hasattr(r, "username"): - # >= Python-2.5 - username, password = r.username, r.password - else: - netloc = r[1] - if "@" in netloc: - auth = netloc.split("@")[0] - if ":" in auth: - username, password = auth.split(":", 1) - else: - username, password = auth, None - else: - username, password = None, None - - if username: - rev_options += ["--username", username] - if password: - rev_options += ["--password", password] - return rev_options diff --git a/tests/sync/test_git.py b/tests/sync/test_git.py index fe2597683..874298a5f 100644 --- a/tests/sync/test_git.py +++ b/tests/sync/test_git.py @@ -143,17 +143,17 @@ def test_repo_update_handle_cases( git_repo: GitSync = constructor(**lazy_constructor_options(**locals())) git_repo.obtain() # clone initial repo - mocka = mocker.spy(git_repo, "run") + cmd_mock = mocker.spy(git_repo.cmd, "symbolic_ref") git_repo.update_repo() - mocka.assert_any_call(["symbolic-ref", "--short", "HEAD"]) + cmd_mock.assert_any_call(name="HEAD", short=True) - mocka.reset_mock() + cmd_mock.reset_mock() # will only look up symbolic-ref if no rev specified for object git_repo.rev = "HEAD" git_repo.update_repo() - assert mocker.call(["symbolic-ref", "--short", "HEAD"]) not in mocka.mock_calls + assert mocker.call(name="HEAD", short=True) not in cmd_mock.mock_calls @pytest.mark.parametrize( @@ -213,10 +213,10 @@ def make_file(filename: str) -> pathlib.Path: some_file = make_file("some_stashed_file") git_repo.run(["add", some_file]) - mocka = mocker.spy(git_repo, "run") + cmd_mock = mocker.spy(git_repo.cmd, "symbolic_ref") git_repo.update_repo() - mocka.assert_any_call(["symbolic-ref", "--short", "HEAD"]) + cmd_mock.assert_any_call(name="HEAD", short=True) @pytest.mark.parametrize( @@ -258,8 +258,6 @@ def progress_callback_spy(output: str, timestamp: datetime.datetime) -> None: name="progress_callback_stub", side_effect=progress_callback_spy ) - run(["git", "rev-parse", "HEAD"], cwd=git_remote_repo) - # create a new repo with the repo as a remote git_repo: GitSync = constructor(**lazy_constructor_options(**locals())) git_repo.obtain()