Coverage for core/langgraph/_builtins_pr.py: 95%

1"""_builtins_pr.py — PR-review builtins (lazy-loaded, needs httpx).""" 

2 

3from __future__ import annotations 

4 

5import json 

6import logging 

7import os 

8from typing import Any 

9 

10import httpx 

11 

12log = logging.getLogger(__name__) 

13 

14 

15def _get_findings(state: dict, findings_key: str) -> list: 

16 """Hae findings state:sta dot-path-avaimella (esim. review.review_output). 

17 

18 Puuttuva polun osa tai ei-lista-arvo → tyhjä lista (manifestin policy). 

19 """ 

20 cur: Any = state 

21 for part in str(findings_key).split("."): 

22 if isinstance(cur, dict): 

23 cur = cur.get(part) 

24 else: 

25 return [] 

26 return cur if isinstance(cur, list) else [] 

27 

28 

29_WIP_PREFIXES = ("WIP: ",) 

30 

31 

32def _strip_wip_prefix(title: str) -> tuple[str, bool]: 

33 lower = title.lower() 

34 for prefix in _WIP_PREFIXES: 

35 if lower.startswith(prefix.lower()): 

36 return title[len(prefix):], True 

37 return title, False 

38 

39 

40def _is_success_status(status_code: int) -> bool: 

41 return 200 <= status_code < 300 

42 

43 

44def _wip_sync_failure_reason(status_code: int) -> str: 

45 if status_code == 401: 

46 return "authentication failed — check bot token" 

47 if status_code == 403: 

48 return "forbidden — bot lacks write access to the repository (read-only bot)" 

49 if status_code == 404: 

50 return "pull request not found (closed or deleted?)" 

51 if 500 <= status_code <= 599: 

52 return "Gitea server error" 

53 return "unexpected status" 

54 

55 

56def builtin_github_pr_comment(step: dict, state: dict) -> dict: 

57 """Post PR review (APPROVED / REQUEST_CHANGES) via Gitea/GitHub REST API. 

58 

59 Findings at or above `fail_on_severity` → event `REQUEST_CHANGES`, 

60 otherwise `APPROVED`. Non-2xx review POST / PR GET responses raise 

61 ValueError. After a successful review the WIP marker in the PR title 

62 is synced: REQUEST_CHANGES → prepend "WIP: "; APPROVED → strip the 

63 "WIP: " prefix. The WIP sync is best-effort: a non-2xx title PATCH 

64 emits an info-level log entry with a cause classified by status 

65 code (401 authentication, 403 missing repo write access, 404 PR 

66 gone, 5xx server error) and the run continues. 

67 No commit status is set; requiring 

68 human approval is a repo admin concern (branch protection). 

69 

70 Env vars used: 

71 RUN_PROFILE — must be "PR" (guard; anything else raises) 

72 GITEA_BASE_URL — Gitea instance URL 

73 GIT_TOKEN — personal access token 

74 PR_NUMBER — pull request number 

75 REPO_FULL_NAME — repository full name (owner/repo) 

76 """ 

77 run_profile = os.environ.get("RUN_PROFILE") 

78 if run_profile != "PR": 

79 raise ValueError( 

80 f"github-pr-comment requires RUN_PROFILE=PR (got {run_profile!r})" 

81 ) 

82 

83 params = step.get("params", {}) or {} 

84 findings_key = params.get("findings_key", "review_output") 

85 fail_on_severity = params.get("fail_on_severity", "Critical") 

86 

87 base_url = os.environ["GITEA_BASE_URL"].rstrip("/") 

88 token = os.environ["GIT_TOKEN"] 

89 pr_number = os.environ["PR_NUMBER"] 

90 repo_full_name = os.environ["REPO_FULL_NAME"] 

91 

92 findings = _get_findings(state, findings_key) 

93 headers = {"Authorization": f"token {token}", "Content-Type": "application/json"} 

94 

95 # PR review: standard Gitea/GitHub review, no commit status 

96 has_blocking = any(f.get("severity") == fail_on_severity for f in findings) 

97 event = "REQUEST_CHANGES" if has_blocking else "APPROVED" 

98 

99 body = _format_findings_as_markdown(findings) 

100 review_resp = httpx.post( 

101 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}/reviews", 

102 json={ 

103 "body": body, 

104 "event": event, 

105 }, 

106 headers=headers, 

107 ) 

108 if not _is_success_status(review_resp.status_code): 

109 raise ValueError( 

110 f"github-pr-comment: review POST failed ({review_resp.status_code}): {review_resp.text[:500]}" 

111 ) 

112 

113 pr_resp = httpx.get( 

114 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}", 

115 headers=headers, 

116 ) 

117 if not _is_success_status(pr_resp.status_code): 

118 raise ValueError( 

119 f"github-pr-comment: PR GET failed ({pr_resp.status_code}): {pr_resp.text[:500]}" 

120 ) 

121 try: 

122 pr_data = pr_resp.json() 

123 except json.JSONDecodeError: 

124 raise ValueError( 

125 f"github-pr-comment: PR GET returned invalid JSON: {pr_resp.text[:500]}" 

126 ) from None 

127 title = pr_data.get("title") if isinstance(pr_data, dict) else None 

128 if not isinstance(title, str): 

129 raise ValueError( 

130 f"github-pr-comment: PR GET missing title: {pr_resp.text[:500]}" 

131 ) 

132 

133 stripped, has_wip_prefix = _strip_wip_prefix(title) 

134 if event == "REQUEST_CHANGES": 

135 new_title = None if has_wip_prefix else f"WIP: {title}" 

136 else: 

137 new_title = stripped if has_wip_prefix else None 

138 

139 wip = "unchanged" 

140 wip_error = None 

141 if new_title is not None: 

142 patch_resp = httpx.patch( 

143 f"{base_url}/api/v1/repos/{repo_full_name}/pulls/{pr_number}", 

144 json={"title": new_title}, 

145 headers=headers, 

146 ) 

147 if not _is_success_status(patch_resp.status_code): 

148 log.info( 

149 "github-pr-comment: WIP title sync failed (%s): %s — %s", 

150 patch_resp.status_code, 

151 patch_resp.text[:500], 

152 _wip_sync_failure_reason(patch_resp.status_code), 

153 ) 

154 wip = "sync_failed" 

155 wip_error = f"{patch_resp.status_code}: {patch_resp.text[:500]}" 

156 else: 

157 wip = "set" if event == "REQUEST_CHANGES" else "cleared" 

158 

159 result = {**state, "review_posted": True, "findings_count": len(findings), "wip": wip} 

160 if wip_error is not None: 

161 result["wip_error"] = wip_error 

162 else: 

163 result.pop("wip_error", None) 

164 return result 

165 

166 

167def _format_findings_as_markdown(findings: list) -> str: 

168 if not findings: 

169 return "## PR Review\n\nNo findings." 

170 lines = ["## PR Review\n"] 

171 for f in findings: 

172 severity = f.get("severity", "Info") 

173 finding = f.get("finding", f.get("message", "")) 

174 evidence = f.get("evidence_ref", f.get("evidence", "")) 

175 lines.append(f"- **{severity}**: {finding}") 

176 if evidence: 

177 lines.append(f" - Evidence: {evidence}") 

178 lines.append(f"\n_{len(findings)} finding(s)_") 

179 return "\n".join(lines) 

180 

181 

182def register_all(executor: Any) -> None: 

183 executor.register("github-pr-comment", builtin_github_pr_comment)