Coverage for src/pdm_multirun/plugin.py: 79.69%

1"""PDM Multirun plugin.""" 

2 

3from __future__ import annotations 

4 

5import os 

6from typing import TYPE_CHECKING 

7 

8from pdm import termui 

9from pdm.cli.commands.run import Command as RunCommand 

10from pdm.cli.commands.run import Project 

11from pdm.cli.commands.use import Command as UseCommand 

12from pdm.cli.hooks import HookManager 

13 

14if TYPE_CHECKING: 

15 import argparse 

16 

17 from pdm.core import Core 

18 

19PYTHON_VERSIONS = os.getenv("PDM_MULTIRUN_VERSIONS", "").split() or [f"3.{minor}" for minor in range(8, 13)] 19 ↛ exitline 19 didn't run the list comprehension on line 19

20USE_VENVS = os.getenv("PDM_MULTIRUN_USE_VENVS", "") == "1" 

21 

22 

23def _comma_separated_list(value: str) -> list[str]: 

24 return value.split(",") 

25 

26 

27class MultirunCommand(RunCommand): 

28 """Run a command under multiple Python versions.""" 

29 

30 def add_arguments(self, parser: argparse.ArgumentParser) -> None: # noqa: D102 

31 super().add_arguments(parser) 

32 parser.add_argument( 

33 "-f", 

34 "--fail-fast", 

35 help="Exit as soon as an interpreter/venv cannot be found or used", 

36 action="store_true", 

37 default=False, 

38 ) 

39 parser.add_argument( 

40 "-i", 

41 "--interpreters", 

42 "--versions", 

43 "--names", 

44 help="Comma-separated list of Python versions or virtual environment names to run the command with", 

45 type=_comma_separated_list, 

46 ) 

47 parser.add_argument( 

48 "-e", 

49 "--venvs", 

50 action="store_true", 

51 default=USE_VENVS, 

52 help="Use virtual environments", 

53 ) 

54 

55 def handle(self, project: Project, options: argparse.Namespace) -> None: # noqa: D102 

56 os.environ["PDM_MULTIRUN"] = "1" 

57 old_python = str(project.environment.interpreter.path) 

58 project.core.ui.echo(f"Current interpreter: {old_python}", verbosity=termui.Verbosity.DETAIL) 

59 for selected in options.interpreters or PYTHON_VERSIONS: 

60 os.environ["PDM_MULTIRUN_CURRENT"] = selected 

61 use_kwargs = {"venv" if options.venvs else "python": selected} 

62 try: 

63 self._use(project, options, **use_kwargs) 

64 except Exception: # noqa: BLE001 

65 if options.fail_fast: 

66 raise 

67 project.core.ui.echo(f"Skipped interpreter/venv: {selected}", verbosity=termui.Verbosity.DETAIL) 

68 continue 

69 try: 

70 super().handle(project, options) 

71 except SystemExit as exit: 

72 if exit.code: 72 ↛ 73line 72 didn't jump to line 73, because the condition on line 72 was never true

73 self._use(project, options, old_python) 

74 raise 

75 os.environ.pop("PDM_MULTIRUN_CURRENT") 

76 

77 project.core.ui.echo(f"Restoring interpreter: {old_python}", verbosity=termui.Verbosity.DETAIL) 

78 self._use(project, options, old_python) 

79 os.environ.pop("PDM_MULTIRUN", None) 

80 

81 def _use(self, project: Project, options: argparse.Namespace, python: str = "", venv: str | None = None) -> None: 

82 old_echo = project.core.ui.echo 

83 if not options.verbose: 83 ↛ 86line 83 didn't jump to line 86, because the condition on line 83 was never false

84 project.core.ui.echo = lambda *args, **kwargs: None # type: ignore[method-assign] 

85 # unset cached environment 

86 project.environment = None # type: ignore[assignment] 

87 try: 

88 UseCommand().do_use( 

89 project, 

90 python=python, 

91 venv=venv, 

92 first=True, 

93 ignore_remembered=False, 

94 hooks=HookManager(project, skip=options.skip), 

95 ) 

96 finally: 

97 project.core.ui.echo = old_echo # type: ignore[method-assign] 

98 

99 

100def multirun(core: Core) -> None: # noqa: D103 

101 core.register_command(MultirunCommand, "multirun")