Coverage for src/keel/github.py: 100%

1"""Thin, fail-soft ``gh`` (GitHub CLI) wrappers (argv, no shell). 

2 

3Like :mod:`keel.git`, these build the exact ``gh`` command for each backbone 

4operation and run it via the injectable ``_run`` seam. Command construction is 

5unit-tested offline; live behaviour is opt-in. 

6""" 

7 

8from __future__ import annotations 

9 

10from .runner import CommandResult, run_argv 

11 

12 

13def open_pr( 

14 title: str, body: str, base: str, head: str, *, cwd: str | None = None, _run=None 

15) -> CommandResult: 

16 return run_argv( 

17 ["gh", "pr", "create", "--title", title, "--body", body, "--base", base, "--head", head], 

18 cwd=cwd, **_kw(_run), 

19 ) 

20 

21 

22def ci_conclusion(pr: int | str, *, cwd: str | None = None, _run=None) -> str | None: 

23 """Return the PR's check-rollup state (e.g. SUCCESS/FAILURE/PENDING), or ``None``.""" 

24 result = run_argv( 

25 ["gh", "pr", "view", str(pr), "--json", "statusCheckRollup", 

26 "--jq", "[.statusCheckRollup[].conclusion] | unique | join(\",\")"], 

27 cwd=cwd, **_kw(_run), 

28 ) 

29 if not result.ok: 

30 return None 

31 return result.output.strip() or None 

32 

33 

34def merged_prs( 

35 *, search: str | None = None, limit: int = 100, cwd: str | None = None, _run=None 

36) -> CommandResult: 

37 """List recently-merged PR numbers as a JSON array (``[{"number": N}, ...]``). 

38 

39 Thin I/O for ``capture-verify`` transport derivation: the authoritative 

40 merged-PR set is read from the host instead of trusting the agent's args. 

41 ``search`` narrows the set (e.g. ``"merged:>=2026-06-01"``). Fail-soft — 

42 the caller inspects ``result.ok`` and degrades gracefully when offline. 

43 """ 

44 argv = ["gh", "pr", "list", "--state", "merged", "--limit", str(limit), "--json", "number"] 

45 if search: 

46 argv += ["--search", search] 

47 return run_argv(argv, cwd=cwd, **_kw(_run)) 

48 

49 

50def list_prs( 

51 *, head: str | None = None, limit: int = 100, cwd: str | None = None, _run=None 

52) -> CommandResult: 

53 """List PRs (any state) as a JSON array (``[{"number": N, "headRefName": ...}, ...]``). 

54 

55 Thin I/O for dry-run integrity verification: the PRs that exist around a 

56 rehearsed run are read from the host. ``head`` narrows to a specific head 

57 branch. Fail-soft — the caller inspects ``result.ok`` and degrades to "no 

58 PRs observed" when offline. 

59 """ 

60 argv = [ 

61 "gh", "pr", "list", "--state", "all", "--limit", str(limit), 

62 "--json", "number,headRefName", 

63 ] 

64 if head: 

65 argv += ["--head", head] 

66 return run_argv(argv, cwd=cwd, **_kw(_run)) 

67 

68 

69def pr_merge_snapshot(pr: int | str, *, cwd: str | None = None, _run=None) -> CommandResult: 

70 return run_argv( 

71 [ 

72 "gh", "pr", "view", str(pr), 

73 "--json", "headRefOid,mergeStateStatus,statusCheckRollup", 

74 ], 

75 cwd=cwd, **_kw(_run), 

76 ) 

77 

78 

79def merge_pr( 

80 pr: int | str, *, method: str = "squash", cwd: str | None = None, _run=None 

81) -> CommandResult: 

82 return run_argv(["gh", "pr", "merge", str(pr), f"--{method}"], cwd=cwd, **_kw(_run)) 

83 

84 

85def comment(pr: int | str, body: str, *, cwd: str | None = None, _run=None) -> CommandResult: 

86 return run_argv(["gh", "pr", "comment", str(pr), "--body", body], cwd=cwd, **_kw(_run)) 

87 

88 

89def post_issue_comment( 

90 owner_repo: str, 

91 issue_or_pr: int | str, 

92 body: str, 

93 *, 

94 cwd: str | None = None, 

95 _run=None, 

96) -> CommandResult: 

97 return run_argv( 

98 [ 

99 "gh", 

100 "api", 

101 f"repos/{owner_repo}/issues/{issue_or_pr}/comments", 

102 "-X", 

103 "POST", 

104 "-f", 

105 f"body={body}", 

106 ], 

107 cwd=cwd, 

108 **_kw(_run), 

109 ) 

110 

111 

112def edit_issue_comment( 

113 owner_repo: str, 

114 comment_id: int | str, 

115 body: str, 

116 *, 

117 cwd: str | None = None, 

118 _run=None, 

119) -> CommandResult: 

120 return run_argv( 

121 [ 

122 "gh", 

123 "api", 

124 f"repos/{owner_repo}/issues/comments/{comment_id}", 

125 "-X", 

126 "PATCH", 

127 "-f", 

128 f"body={body}", 

129 ], 

130 cwd=cwd, 

131 **_kw(_run), 

132 ) 

133 

134 

135def close_issue(issue: int | str, *, cwd: str | None = None, _run=None) -> CommandResult: 

136 return run_argv(["gh", "issue", "close", str(issue)], cwd=cwd, **_kw(_run)) 

137 

138 

139def issue_facts(issue: int | str, *, cwd: str | None = None, _run=None) -> CommandResult: 

140 """Fetch an issue's ``title`` and ``labels`` as JSON for ``keel guard``. 

141 

142 Thin I/O for blocker evaluation: the issue facts are read from the host 

143 rather than trusting agent-supplied args. Fail-soft — the caller inspects 

144 ``result.ok`` and falls back to offline args when offline. 

145 """ 

146 return run_argv( 

147 ["gh", "issue", "view", str(issue), "--json", "title,labels"], 

148 cwd=cwd, **_kw(_run), 

149 ) 

150 

151 

152def _kw(_run): 

153 return {"_run": _run} if _run is not None else {}