Coverage for plugin/scripts/_issue_forge.py: 100%

50 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-10-26 04:21 +0000

1"""The forge half of ``issue_status.py``: ask `gh` or `glab`, and read back one shape. 

2 

3Split from ``issue_status.py`` for the same reason ``_issue_signals`` was: asking a forge 

4for a payload and deciding what that payload means are different instruments with 

5different failure modes. This half owns everything that knows a CLI exists — the argv, 

6the subprocess, and the per-platform key names — and hands the orchestrator issues in one 

7shared vocabulary. Nothing here reads an issue's prose or touches the working tree. 

8 

9The two CLIs disagree about the **shape** of an issue, not merely about flag names, and 

10there are five disagreements in a single object: 

11 

12 gh issue list --state open --json number,title,body,labels,author,createdAt,url 

13 -> {"number": 95, "body": "…", "labels": [{"name": "bug"}], 

14 "author": {"login": "tschm"}, "createdAt": "…", "url": "…"} 

15 

16 glab issue list --output json --per-page 20 

17 -> {"iid": 95, "description": "…", "labels": ["bug"], 

18 "author": {"username": "tschm"}, "created_at": "…", "web_url": "…"} 

19 

20Read with the other's keys, every one of those yields an empty body or a nameless issue 

21rather than an error — which is the failure mode `platform_cli.py` records having 

22shipped once already, when `/update` reached GitLab with a flag `glab` does not have. 

23""" 

24 

25from __future__ import annotations 

26 

27import json 

28import shutil 

29import subprocess # nosec B404 

30from pathlib import Path 

31from typing import Any 

32 

33# Spelled out rather than globbed: `gh` errors on an unknown field, so this list is 

34# itself a contract with the CLI and a typo here fails loudly instead of silently 

35# dropping a signal the triage depends on. 

36_GH_FIELDS = "number,title,body,labels,author,createdAt,updatedAt,url,state,stateReason,closedAt" 

37_GH_REQUEST_FIELDS = "number,title,body,url" 

38 

39# Which issues a listing asks for. GitHub takes the word; GitLab lists open issues by 

40# default and spells the other two as flags of their own. 

41STATES = ("open", "closed", "all") 

42_GLAB_STATE_FLAGS = {"open": [], "closed": ["--closed"], "all": ["--all"]} 

43 

44 

45class ForgeQueryError(Exception): 

46 """A platform CLI was absent, unauthenticated, or answered something unreadable.""" 

47 

48 

49def build_issue_command( 

50 platform: str, *, limit: int, labels: list[str], state: str = "open" 

51) -> list[str]: 

52 """Return the argv listing issues in *state*, optionally narrowed to *labels*. 

53 

54 >>> build_issue_command("github", limit=5, labels=[])[:5] 

55 ['gh', 'issue', 'list', '--state', 'open'] 

56 

57 GitLab's listing is open by default and takes its labels comma-joined, which is the 

58 second place these two CLIs stop lining up: 

59 

60 >>> build_issue_command("gitlab", limit=5, labels=["bug", "docs"])[-2:] 

61 ['--label', 'bug,docs'] 

62 

63 and it has no ``--state``, so the history is a flag of its own: 

64 

65 >>> build_issue_command("gitlab", limit=5, labels=[], state="all")[-1] 

66 '--all' 

67 """ 

68 if platform == "github": 

69 command = ["gh", "issue", "list", "--state", state, "--json", _GH_FIELDS] 

70 command += ["--limit", str(limit)] 

71 for label in labels: 

72 command += ["--label", label] 

73 return command 

74 command = ["glab", "issue", "list", "--output", "json", "--per-page", str(limit)] 

75 command += _GLAB_STATE_FLAGS[state] 

76 return [*command, "--label", ",".join(labels)] if labels else command 

77 

78 

79def build_request_command(platform: str, *, limit: int) -> list[str]: 

80 """Return the argv listing open requests, so an issue already being worked is seen. 

81 

82 >>> build_request_command("gitlab", limit=3) 

83 ['glab', 'mr', 'list', '--output', 'json', '--per-page', '3'] 

84 """ 

85 if platform == "github": 

86 return [ 

87 "gh", "pr", "list", "--state", "open", 

88 "--json", _GH_REQUEST_FIELDS, "--limit", str(limit), 

89 ] # fmt: skip 

90 return ["glab", "mr", "list", "--output", "json", "--per-page", str(limit)] 

91 

92 

93def build_reference_commands(platform: str, number: int) -> list[list[str]]: 

94 """Return the argv candidates that resolve reference *number*, in order. 

95 

96 On GitHub one number may name an issue or a pull request, and only trying tells 

97 you which — so both are returned and the first that answers wins. On GitLab a `#N` 

98 is always an issue, so there is exactly one candidate and a merge request reached 

99 this way is simply not found: 

100 

101 >>> build_reference_commands("gitlab", 75) 

102 [['glab', 'issue', 'view', '75', '--output', 'json']] 

103 >>> len(build_reference_commands("github", 75)) 

104 2 

105 """ 

106 if platform == "github": 

107 return [ 

108 ["gh", "issue", "view", str(number), "--json", "number,state,title"], 

109 ["gh", "pr", "view", str(number), "--json", "number,state,title"], 

110 ] 

111 return [["glab", "issue", "view", str(number), "--output", "json"]] 

112 

113 

114def run_json(command: list[str], target_dir: Path) -> Any: 

115 """Run *command* in *target_dir* and parse its stdout as JSON. 

116 

117 Raises: 

118 ForgeQueryError: the binary is missing, it failed, or stdout was not JSON. 

119 """ 

120 binary = shutil.which(command[0]) 

121 if binary is None: 

122 raise ForgeQueryError(f"{command[0]} is not installed") 

123 result = subprocess.run( # nosec B603 

124 [binary, *command[1:]], 

125 cwd=str(target_dir), 

126 capture_output=True, 

127 text=True, 

128 check=False, 

129 ) 

130 if result.returncode != 0: 

131 detail = (result.stderr or result.stdout).strip().splitlines() 

132 raise ForgeQueryError(detail[0] if detail else f"{command[0]} failed") 

133 try: 

134 return json.loads(result.stdout or "[]") 

135 except json.JSONDecodeError as exc: # pragma: no cover - defensive 

136 raise ForgeQueryError(f"{command[0]} did not answer JSON: {exc}") from exc 

137 

138 

139def _text(raw: dict[str, Any], key: str) -> str: 

140 """*raw*'s value at *key* as a string, with a missing or null value as ``''``.""" 

141 return str(raw.get(key) or "") 

142 

143 

144def _state(value: str) -> str: 

145 """One forge's issue state in the shared vocabulary: ``open``, ``closed`` or ``''``. 

146 

147 GitHub shouts (``OPEN``) and GitLab says ``opened``; neither is wrong, and a caller 

148 comparing against one spelling would silently miss the other: 

149 

150 >>> [_state(v) for v in ("OPEN", "opened", "CLOSED", "")] 

151 ['open', 'open', 'closed', ''] 

152 """ 

153 lowered = value.lower() 

154 return "open" if lowered == "opened" else lowered 

155 

156 

157def _normalize_github(raw: dict[str, Any]) -> dict[str, Any]: 

158 """One `gh issue list` object in the shared vocabulary.""" 

159 return { 

160 "id": raw.get("number"), 

161 "title": _text(raw, "title"), 

162 "body": _text(raw, "body"), 

163 "labels": [label.get("name", "") for label in raw.get("labels") or []], 

164 "author": (raw.get("author") or {}).get("login", ""), 

165 "created": _text(raw, "createdAt"), 

166 "url": _text(raw, "url"), 

167 "state": _state(_text(raw, "state")), 

168 "state_reason": _text(raw, "stateReason").lower(), 

169 "closed": _text(raw, "closedAt"), 

170 } 

171 

172 

173def _normalize_gitlab(raw: dict[str, Any]) -> dict[str, Any]: 

174 """One `glab issue list` object in the shared vocabulary. 

175 

176 Five keys differ from GitHub's and none of them errors when read with the wrong 

177 name — `iid`/`number`, `description`/`body`, label strings against label objects, 

178 snake_case against camelCase, and `web_url`/`url`. A sixth is missing outright: 

179 GitLab records no reason for a close, so ``state_reason`` is always empty here. 

180 """ 

181 return { 

182 "id": raw.get("iid"), 

183 "title": _text(raw, "title"), 

184 "body": _text(raw, "description"), 

185 "labels": list(raw.get("labels") or []), 

186 "author": (raw.get("author") or {}).get("username", ""), 

187 "created": _text(raw, "created_at"), 

188 "url": _text(raw, "web_url"), 

189 "state": _state(_text(raw, "state")), 

190 "state_reason": "", 

191 "closed": _text(raw, "closed_at"), 

192 } 

193 

194 

195def normalize(platform: str, raw: dict[str, Any]) -> dict[str, Any]: 

196 """Return one issue in the shared vocabulary, whichever CLI produced it. 

197 

198 The whole point is that both sides land on the same object: 

199 

200 >>> gh = {"number": 95, "title": "t", "body": "b", "labels": [{"name": "bug"}], 

201 ... "author": {"login": "tschm"}, "createdAt": "x", "url": "u"} 

202 >>> lab = {"iid": 95, "title": "t", "description": "b", "labels": ["bug"], 

203 ... "author": {"username": "tschm"}, "created_at": "x", "web_url": "u"} 

204 >>> normalize("github", gh) == normalize("gitlab", lab) 

205 True 

206 """ 

207 return _normalize_github(raw) if platform == "github" else _normalize_gitlab(raw)