Coverage for src/ezcompiler/adapters/base_publisher.py: 100.00%

1# /////////////////////////////////////////////////////////////// 

2# BASE_PUBLISHER - Abstract base release publisher 

3# Project: ezcompiler 

4# /////////////////////////////////////////////////////////////// 

5 

6"""Shared CLI behavior for release publishers.""" 

7 

8from __future__ import annotations 

9 

10# /////////////////////////////////////////////////////////////// 

11# IMPORTS 

12# /////////////////////////////////////////////////////////////// 

13import shutil 

14 

15# CLI execution is isolated in _run_cli and always disables the shell. 

16import subprocess # noqa: S404 # nosec B404 

17from abc import ABC, abstractmethod 

18from pathlib import Path 

19from typing import Any 

20 

21from ..shared.exceptions import PublishAuthError, PublishCliError 

22 

23# /////////////////////////////////////////////////////////////// 

24# CONSTANTS 

25# /////////////////////////////////////////////////////////////// 

26 

27# Ceiling for each external CLI call. A `gh release create` uploading an 

28# installer behind a corporate proxy that stalls would otherwise hang the 

29# process indefinitely, with no output at all (capture_output=True). 

30CLI_TIMEOUT_SECONDS = 300 

31 

32# /////////////////////////////////////////////////////////////// 

33# CLASSES 

34# /////////////////////////////////////////////////////////////// 

35 

36 

37class BasePublisher(ABC): 

38 """Abstract publisher with one mockable seam for external CLI calls.""" 

39 

40 _cli_name: str = "" 

41 _install_hint: str = "" 

42 

43 def __init__(self, config: dict[str, Any] | None = None) -> None: 

44 self._config = config or {} 

45 

46 @property 

47 def config(self) -> dict[str, Any]: 

48 """Publisher configuration.""" 

49 return self._config 

50 

51 def preflight(self) -> None: 

52 """Raise unless the CLI is installed and authenticated. 

53 

54 Lets callers run every check before showing a recap, so that an 

55 unauthenticated CLI is reported as such rather than as an opaque 

56 failure of a later call. 

57 """ 

58 self._check_auth() 

59 

60 @abstractmethod 

61 def exists(self, tag: str) -> bool: 

62 """Whether a release exists for ``tag``.""" 

63 

64 @abstractmethod 

65 def publish( 

66 self, 

67 assets: list[Path], 

68 *, 

69 tag: str, 

70 title: str, 

71 notes: str | None = None, 

72 prerelease: bool = False, 

73 draft: bool = False, 

74 ) -> str: 

75 """Create a release with assets and return its URL.""" 

76 

77 @abstractmethod 

78 def get_publisher_name(self) -> str: 

79 """Human-readable publisher name.""" 

80 

81 def _resolve_cli(self) -> str: 

82 """Return the CLI binary path or raise with an installation hint.""" 

83 exe = shutil.which(self._cli_name) 

84 if exe is None: 

85 raise PublishCliError( 

86 f"'{self._cli_name}' not found in PATH. " 

87 f"Installer depuis {self._install_hint}, " 

88 f"puis lancer `{self._cli_name} auth login`." 

89 ) 

90 return exe 

91 

92 def _check_auth(self) -> None: 

93 """Raise when the external CLI reports no valid credentials.""" 

94 result = self._run_cli(["auth", "status"], check=False) 

95 if result.returncode != 0: 

96 raise PublishAuthError( 

97 f"'{self._cli_name}' is not authenticated. " 

98 f"Run `{self._cli_name} auth login`, or export the token " 

99 "into the environment." 

100 ) 

101 

102 def _run_cli( 

103 self, 

104 args: list[str], 

105 *, 

106 check: bool = True, 

107 timeout: float = CLI_TIMEOUT_SECONDS, 

108 ) -> subprocess.CompletedProcess[str]: 

109 """Run a CLI command without a shell and capture its output. 

110 

111 CLI stderr is retained in the returned result for callers using 

112 ``check=False``. It is never copied into an exception, because the 

113 external CLI can print credentials from outside this process. 

114 

115 Args: 

116 args: Argument vector appended to the resolved CLI binary. 

117 check: Raise ``PublishCliError`` on a non-zero return code. 

118 timeout: Hard ceiling in seconds; expiry is reported as a 

119 publication error rather than hanging the process. 

120 

121 Raises: 

122 PublishCliError: If the CLI cannot be executed, exceeds 

123 ``timeout``, or (with ``check=True``) returns non-zero. 

124 """ 

125 exe = self._resolve_cli() 

126 # Argument vector is passed directly; no shell interprets asset names. 

127 try: 

128 result = subprocess.run( # noqa: S603 # nosec B603 

129 [exe, *args], 

130 capture_output=True, 

131 text=True, 

132 check=False, 

133 shell=False, 

134 timeout=timeout, 

135 ) 

136 except subprocess.TimeoutExpired: 

137 raise PublishCliError( 

138 f"`{self._cli_name}` did not answer within {timeout:g} s and was " 

139 "interrupted. Check the network or the proxy, then review " 

140 "the state of the release before retrying." 

141 ) from None 

142 except OSError: 

143 raise PublishCliError( 

144 f"'{self._cli_name}' could not be executed." 

145 ) from None 

146 if check and result.returncode != 0: 

147 raise PublishCliError( 

148 f"`{self._cli_name}` failed (code {result.returncode})." 

149 ) 

150 return result