Coverage for src/lpip/pipwrap.py: 100%

1"""Translate catalog entries to pip and run ``python -m pip`` (ADR-0001).""" 

2 

3from __future__ import annotations 

4 

5import subprocess 

6from collections.abc import Mapping 

7from dataclasses import replace 

8 

9from lpip.errors import InstallRevisionEqualsConflictError 

10from lpip.models import PackageSource, RenderedRequirement 

11from lpip.reqfile import parse_catalog_request 

12from lpip.validation import normalize_git_file_url 

13from lpip.versions import tag_for_version 

14 

15 

16def render_requirement(package: PackageSource) -> RenderedRequirement: 

17 """Build ``name @ url[@revision][#subdirectory=…]`` (AC-0006, ADR-0010).""" 

18 vcs = normalize_git_file_url(package.url) 

19 if package.revision: 

20 vcs = f"{vcs}@{package.revision}" 

21 if package.subdirectory: 

22 vcs = f"{vcs}#subdirectory={package.subdirectory}" 

23 return RenderedRequirement(package=package.name, text=f"{package.name} @ {vcs}") 

24 

25 

26def render_catalog_install( 

27 package: PackageSource, 

28 line: str, 

29 *, 

30 environ: Mapping[str, str] | None = None, 

31 revision: str | None = None, 

32) -> str: 

33 """Catalog URL; ``==`` tag (ADR-0018) or ``--revision`` override (ADR-0023).""" 

34 _name, version = parse_catalog_request(line) 

35 if revision is not None: 

36 if version is not None: 

37 raise InstallRevisionEqualsConflictError() 

38 return render_requirement(replace(package, revision=revision)).text 

39 if version is None: 

40 return render_requirement(package).text 

41 tag = tag_for_version(package, version, environ=environ) 

42 return render_requirement(replace(package, revision=tag)).text 

43 

44 

45def format_pip_install_command( 

46 python: str, 

47 requirement: str, 

48 *, 

49 upgrade: bool = False, 

50) -> str: 

51 """Shell-style dry-run line. Does not start pip (AC-0001).""" 

52 flag = " --upgrade" if upgrade else "" 

53 return f'{python} -m pip install{flag} "{requirement}"' 

54 

55 

56def pip_install_argv( 

57 python: str, 

58 requirement: str, 

59 *, 

60 upgrade: bool = False, 

61) -> list[str]: 

62 argv = [python, "-m", "pip", "install"] 

63 if upgrade: 

64 argv.append("--upgrade") 

65 argv.append(requirement) 

66 return argv 

67 

68 

69def run_pip_install( 

70 python: str, 

71 requirement: str, 

72 *, 

73 environ: Mapping[str, str] | None = None, 

74 upgrade: bool = False, 

75) -> int: 

76 """Run pip and forward stdout/stderr. Returns pip's exit code (AC-0015).""" 

77 completed = subprocess.run( # nosec B603 

78 pip_install_argv(python, requirement, upgrade=upgrade), 

79 env=dict(environ) if environ is not None else None, 

80 ) 

81 return completed.returncode