diff --git a/src/libvcs/_internal/run.py b/src/libvcs/_internal/run.py index 8b586ba8a..d68413ab8 100644 --- a/src/libvcs/_internal/run.py +++ b/src/libvcs/_internal/run.py @@ -15,7 +15,7 @@ import subprocess import sys import typing as t -from collections.abc import Iterable, Mapping, MutableMapping, Sequence +from collections.abc import Collection, Iterable, Mapping, MutableMapping, Sequence from libvcs import exc from libvcs._internal.types import StrPath @@ -75,7 +75,7 @@ def process( self, msg: str, kwargs: MutableMapping[str, t.Any], - ) -> tuple[t.Any, MutableMapping[str, t.Any]]: + ) -> tuple[str, MutableMapping[str, t.Any]]: """Add additional context information for loggers.""" prefixed_dict = {} prefixed_dict["bin_name"] = self.bin_name @@ -100,7 +100,7 @@ def __call__(self, output: str, timestamp: datetime.datetime) -> None: _ENV: t.TypeAlias = Mapping[bytes, StrPath] | Mapping[str, StrPath] _CMD = StrPath | Sequence[StrPath] -_FILE: t.TypeAlias = int | t.IO[t.Any] | None +_FILE: t.TypeAlias = int | t.IO[str] | t.IO[bytes] | None def run( @@ -110,7 +110,7 @@ def run( stdin: _FILE | None = None, stdout: _FILE | None = None, stderr: _FILE | None = None, - preexec_fn: t.Callable[[], t.Any] | None = None, + preexec_fn: t.Callable[[], object] | None = None, close_fds: bool = True, shell: bool = False, cwd: StrPath | None = None, @@ -119,7 +119,7 @@ def run( creationflags: int = 0, restore_signals: bool = True, start_new_session: bool = False, - pass_fds: t.Any = (), + pass_fds: Collection[int] = (), *, encoding: str | None = None, errors: str | None = None, diff --git a/src/libvcs/_internal/shortcuts.py b/src/libvcs/_internal/shortcuts.py index a3e401d33..57d81e56e 100644 --- a/src/libvcs/_internal/shortcuts.py +++ b/src/libvcs/_internal/shortcuts.py @@ -38,7 +38,7 @@ def create_project( path: StrPath, vcs: t.Literal["git"], progress_callback: ProgressCallbackProtocol | None = None, - **kwargs: dict[t.Any, t.Any], + **kwargs: dict[str, t.Any], ) -> GitSync: ... @@ -49,7 +49,7 @@ def create_project( path: StrPath, vcs: t.Literal["svn"], progress_callback: ProgressCallbackProtocol | None = None, - **kwargs: dict[t.Any, t.Any], + **kwargs: dict[str, t.Any], ) -> SvnSync: ... @@ -60,7 +60,7 @@ def create_project( path: StrPath, vcs: t.Literal["hg"], progress_callback: ProgressCallbackProtocol | None = ..., - **kwargs: dict[t.Any, t.Any], + **kwargs: dict[str, t.Any], ) -> HgSync: ... @@ -71,7 +71,7 @@ def create_project( path: StrPath, vcs: None = None, progress_callback: ProgressCallbackProtocol | None = None, - **kwargs: dict[t.Any, t.Any], + **kwargs: dict[str, t.Any], ) -> GitSync | HgSync | SvnSync: ... @@ -81,7 +81,7 @@ def create_project( path: StrPath, vcs: VCSLiteral | None = None, progress_callback: ProgressCallbackProtocol | None = None, - **kwargs: dict[t.Any, t.Any], + **kwargs: dict[str, t.Any], ) -> GitSync | HgSync | SvnSync: r"""Return an object representation of a VCS repository. @@ -127,7 +127,7 @@ def create_project( assert vcs_matches[0].vcs is not None - def is_vcs(val: t.Any) -> t.TypeGuard[VCSLiteral]: + def is_vcs(val: str) -> t.TypeGuard[VCSLiteral]: return isinstance(val, str) and val in {"git", "hg", "svn"} if is_vcs(vcs_matches[0].vcs): diff --git a/src/libvcs/_internal/subprocess.py b/src/libvcs/_internal/subprocess.py index 003de28f2..7e677dce8 100644 --- a/src/libvcs/_internal/subprocess.py +++ b/src/libvcs/_internal/subprocess.py @@ -45,7 +45,7 @@ import subprocess import sys import typing as t -from collections.abc import Mapping, Sequence +from collections.abc import Collection, Mapping, Sequence from libvcs._internal.types import StrOrBytesPath @@ -63,7 +63,7 @@ def __init__(self, output: str, *args: object) -> None: _ENV: t.TypeAlias = Mapping[str, str] else: _ENV: t.TypeAlias = Mapping[bytes, StrOrBytesPath] | Mapping[str, StrOrBytesPath] -_FILE: t.TypeAlias = None | int | t.IO[t.Any] +_FILE: t.TypeAlias = None | int | t.IO[str] | t.IO[bytes] _TXT: t.TypeAlias = bytes | str #: Command _CMD: t.TypeAlias = StrOrBytesPath | Sequence[StrOrBytesPath] @@ -96,7 +96,7 @@ class SubprocessCommand(SkipDefaultFieldsReprMixin): stdin: _FILE = None stdout: _FILE = None stderr: _FILE = None - preexec_fn: t.Callable[[], t.Any] | None = None + preexec_fn: t.Callable[[], object] | None = None close_fds: bool = True shell: bool = False cwd: StrOrBytesPath | None = None @@ -109,7 +109,7 @@ class SubprocessCommand(SkipDefaultFieldsReprMixin): # POSIX-only restore_signals: bool = True start_new_session: bool = False - pass_fds: t.Any = () + pass_fds: Collection[int] = () umask: int = -1 pipesize: int = -1 user: str | None = None diff --git a/src/libvcs/cmd/git.py b/src/libvcs/cmd/git.py index 23fd21d25..42cb41d1e 100644 --- a/src/libvcs/cmd/git.py +++ b/src/libvcs/cmd/git.py @@ -16,6 +16,19 @@ from libvcs._internal.types import StrOrBytesPath, StrPath _CMD = StrOrBytesPath | Sequence[StrOrBytesPath] +GitConfigValue: t.TypeAlias = bool | int | float | StrPath + + +class GitSubmoduleData(t.TypedDict): + """Structured submodule data returned by _ls().""" + + name: str + path: str + sha: str + url: str | None + branch: str | None + status_prefix: str + description: str class Git: @@ -141,7 +154,7 @@ def run( noglob_pathspecs: bool | None = None, icase_pathspecs: bool | None = None, no_optional_locks: bool | None = None, - config: dict[str, t.Any] | None = None, + config: dict[str, GitConfigValue] | None = None, config_env: str | None = None, # Pass-through to run() log_in_real_time: bool = False, @@ -241,7 +254,7 @@ def run( if config is not None: assert isinstance(config, dict) - def stringify(v: t.Any) -> str: + def stringify(v: GitConfigValue) -> str: if isinstance(v, bool): return "true" if v else "false" if not isinstance(v, str): @@ -316,7 +329,7 @@ def clone( verbose: bool | None = None, quiet: bool | None = None, # Pass-through to run - config: dict[str, t.Any] | None = None, + config: dict[str, GitConfigValue] | None = None, log_in_real_time: bool = False, # Special behavior check_returncode: bool | None = None, @@ -420,7 +433,7 @@ def clone( def fetch( self, *, - reftag: t.Any | None = None, + reftag: str | None = None, deepen: str | None = None, depth: str | None = None, upload_pack: str | None = None, @@ -788,7 +801,7 @@ def rebase( def pull( self, *, - reftag: t.Any | None = None, + reftag: str | None = None, repository: str | None = None, deepen: str | None = None, depth: str | None = None, @@ -3386,7 +3399,7 @@ def _ls( # Pass-through to run() log_in_real_time: bool = False, check_returncode: bool | None = None, - ) -> list[dict[str, t.Any]]: + ) -> list[GitSubmoduleData]: """Parse submodule status output into structured data. Parameters @@ -3398,7 +3411,7 @@ def _ls( Returns ------- - list[dict[str, Any]] + list[GitSubmoduleData] List of parsed submodule data. Examples @@ -3421,7 +3434,7 @@ def _ls( log_in_real_time=log_in_real_time, ) - submodules: list[dict[str, t.Any]] = [] + submodules: list[GitSubmoduleData] = [] for line in result.strip().split("\n"): if not line: diff --git a/tests/_internal/subprocess/test_SubprocessCommand.py b/tests/_internal/subprocess/test_SubprocessCommand.py index 345d2912e..7061febf8 100644 --- a/tests/_internal/subprocess/test_SubprocessCommand.py +++ b/tests/_internal/subprocess/test_SubprocessCommand.py @@ -8,12 +8,17 @@ import pytest from libvcs._internal.subprocess import SubprocessCommand +from libvcs._internal.types import StrOrBytesPath if t.TYPE_CHECKING: import pathlib -def idfn(val: t.Any) -> str: +CmdArgs = StrOrBytesPath | t.Sequence[StrOrBytesPath] +Kwargs = dict[str, t.Any] + + +def idfn(val: CmdArgs | Kwargs | SubprocessCommand | None) -> str: """Test ID naming function for SubprocessCommand py.test parametrize.""" if isinstance(val, list): if len(val): @@ -24,15 +29,19 @@ def idfn(val: t.Any) -> str: @pytest.mark.parametrize( - ("args", "kwargs", "expected_result"), + ("cmd_args", "kwargs", "expected_result"), [ - (["ls"], {}, SubprocessCommand("ls")), - ([["ls", "-l"]], {}, SubprocessCommand(["ls", "-l"])), - ([], {"args": ["ls", "-l"]}, SubprocessCommand(["ls", "-l"])), - (["ls -l"], {"shell": True}, SubprocessCommand("ls -l", shell=True)), - ([], {"args": "ls -l", "shell": True}, SubprocessCommand("ls -l", shell=True)), + ("ls", {}, SubprocessCommand("ls")), + (["ls", "-l"], {}, SubprocessCommand(["ls", "-l"])), + (None, {"args": ["ls", "-l"]}, SubprocessCommand(["ls", "-l"])), + ("ls -l", {"shell": True}, SubprocessCommand("ls -l", shell=True)), + ( + None, + {"args": "ls -l", "shell": True}, + SubprocessCommand("ls -l", shell=True), + ), ( - [], + None, {"args": ["ls", "-l"], "shell": True}, SubprocessCommand(["ls", "-l"], shell=True), ), @@ -40,12 +49,16 @@ def idfn(val: t.Any) -> str: ids=idfn, ) def test_init( - args: list[t.Any], - kwargs: dict[str, t.Any], - expected_result: t.Any, + cmd_args: CmdArgs | None, + kwargs: Kwargs, + expected_result: SubprocessCommand, ) -> None: """Test SubprocessCommand via list + kwargs, assert attributes.""" - cmd = SubprocessCommand(*args, **kwargs) + cmd = ( + SubprocessCommand(cmd_args, **kwargs) + if cmd_args is not None + else SubprocessCommand(**kwargs) + ) assert cmd == expected_result # Attributes in cmd should match what's passed in @@ -57,116 +70,116 @@ def test_init( assert proc.returncode == 0 -FIXTURES = [ - [["ls"], {}, SubprocessCommand("ls")], - [[["ls", "-l"]], {}, SubprocessCommand(["ls", "-l"])], +FIXTURES: list[tuple[CmdArgs, Kwargs, SubprocessCommand]] = [ + ("ls", {}, SubprocessCommand("ls")), + (["ls", "-l"], {}, SubprocessCommand(["ls", "-l"])), ] @pytest.mark.parametrize( - ("args", "kwargs", "expected_result"), + ("cmd_args", "kwargs", "expected_result"), FIXTURES, ids=idfn, ) def test_init_and_Popen( - args: list[t.Any], - kwargs: dict[str, t.Any], - expected_result: t.Any, + cmd_args: CmdArgs, + kwargs: Kwargs, + expected_result: SubprocessCommand, ) -> None: """Test SubprocessCommand with Popen.""" - cmd = SubprocessCommand(*args, **kwargs) + cmd = SubprocessCommand(cmd_args, **kwargs) assert cmd == expected_result cmd_proc = cmd.Popen() cmd_proc.communicate() assert cmd_proc.returncode == 0 - proc = subprocess.Popen(*args, **kwargs) + proc = subprocess.Popen(cmd_args, **kwargs) proc.communicate() assert proc.returncode == 0 @pytest.mark.parametrize( - ("args", "kwargs", "expected_result"), + ("cmd_args", "kwargs", "expected_result"), FIXTURES, ids=idfn, ) def test_init_and_Popen_run( - args: list[t.Any], - kwargs: dict[str, t.Any], - expected_result: t.Any, + cmd_args: CmdArgs, + kwargs: Kwargs, + expected_result: SubprocessCommand, ) -> None: """Test SubprocessCommand with run.""" - cmd = SubprocessCommand(*args, **kwargs) + cmd = SubprocessCommand(cmd_args, **kwargs) assert cmd == expected_result cmd_proc = cmd.Popen() cmd_proc.communicate() assert cmd_proc.returncode == 0 - proc = subprocess.run(*args, **kwargs, check=False) + proc = subprocess.run(cmd_args, **kwargs, check=False) assert proc.returncode == 0 @pytest.mark.parametrize( - ("args", "kwargs", "expected_result"), + ("cmd_args", "kwargs", "expected_result"), FIXTURES, ids=idfn, ) def test_init_and_check_call( - args: list[t.Any], - kwargs: dict[str, t.Any], - expected_result: t.Any, + cmd_args: CmdArgs, + kwargs: Kwargs, + expected_result: SubprocessCommand, ) -> None: """Test SubprocessCommand with Popen.check_call.""" - cmd = SubprocessCommand(*args, **kwargs) + cmd = SubprocessCommand(cmd_args, **kwargs) assert cmd == expected_result return_code = cmd.check_call() assert return_code == 0 - proc = subprocess.check_call(*args, **kwargs) + proc = subprocess.check_call(cmd_args, **kwargs) assert proc == return_code @pytest.mark.parametrize( - ("args", "kwargs", "expected_result"), + ("cmd_args", "kwargs", "expected_result"), FIXTURES, ) def test_init_and_check_output( - args: list[t.Any], - kwargs: dict[str, t.Any], - expected_result: t.Any, + cmd_args: CmdArgs, + kwargs: Kwargs, + expected_result: SubprocessCommand, ) -> None: """Test SubprocessCommand with Popen.check_output.""" - cmd = SubprocessCommand(*args, **kwargs) + cmd = SubprocessCommand(cmd_args, **kwargs) assert cmd == expected_result return_output = cmd.check_output() assert isinstance(return_output, bytes) - proc = subprocess.check_output(*args, **kwargs) + proc = subprocess.check_output(cmd_args, **kwargs) assert proc == return_output @pytest.mark.parametrize( - ("args", "kwargs", "run_kwargs"), + ("cmd_args", "kwargs", "run_kwargs"), [ - (["ls"], {}, {}), - ([["ls", "-l"]], {}, {}), - ([["ls", "-al"]], {}, {"stdout": subprocess.DEVNULL}), + ("ls", {}, {}), + (["ls", "-l"], {}, {}), + (["ls", "-al"], {}, {"stdout": subprocess.DEVNULL}), ], ids=idfn, ) def test_run( tmp_path: pathlib.Path, - args: list[t.Any], - kwargs: dict[str, t.Any], - run_kwargs: dict[str, t.Any], + cmd_args: CmdArgs, + kwargs: Kwargs, + run_kwargs: Kwargs, ) -> None: """Test SubprocessCommand.run().""" kwargs["cwd"] = tmp_path - cmd = SubprocessCommand(*args, **kwargs) + cmd = SubprocessCommand(cmd_args, **kwargs) response = cmd.run(**run_kwargs) assert response.returncode == 0 diff --git a/tests/cmd/test_git.py b/tests/cmd/test_git.py index 0dfe46baa..072279e07 100644 --- a/tests/cmd/test_git.py +++ b/tests/cmd/test_git.py @@ -18,7 +18,7 @@ @pytest.mark.parametrize("path_type", [str, pathlib.Path]) def test_git_constructor( - path_type: t.Callable[[str | pathlib.Path], t.Any], + path_type: t.Callable[[str | pathlib.Path], str | pathlib.Path], tmp_path: pathlib.Path, ) -> None: """Test Git constructor."""