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

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

2 

3from __future__ import annotations 

4 

5import json 

6import os 

7from typing import Any 

8 

9import httpx 

10 

11 

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

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

14 

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

16 """ 

17 cur: Any = state 

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

19 if isinstance(cur, dict): 

20 cur = cur.get(part) 

21 else: 

22 return [] 

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

24 

25 

26_WIP_PREFIXES = ("WIP: ",) 

27 

28 

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

30 lower = title.lower() 

31 for prefix in _WIP_PREFIXES: 

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

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

34 return title, False 

35 

36 

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

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

39 

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

41 otherwise `APPROVED`. Non-2xx HTTP responses raise ValueError. After a 

42 successful review the WIP marker in the PR title is synced: 

43 REQUEST_CHANGES → prepend "WIP: "; APPROVED → strip the "WIP: " prefix. 

44 No commit status is set; requiring 

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

46 

47 Env vars used: 

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

49 GITEA_BASE_URL — Gitea instance URL 

50 GIT_TOKEN — personal access token 

51 PR_NUMBER — pull request number 

52 REPO_FULL_NAME — repository full name (owner/repo) 

53 """ 

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

55 if run_profile != "PR": 

56 raise ValueError( 

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

58 ) 

59 

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

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

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

63 

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

65 token = os.environ["GIT_TOKEN"] 

66 pr_number = os.environ["PR_NUMBER"] 

67 repo_full_name = os.environ["REPO_FULL_NAME"] 

68 

69 findings = _get_findings(state, findings_key) 

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

71 

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

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

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

75 

76 body = _format_findings_as_markdown(findings) 

77 review_resp = httpx.post( 

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

79 json={ 

80 "body": body, 

81 "event": event, 

82 }, 

83 headers=headers, 

84 ) 

85 if review_resp.status_code >= 400: 

86 raise ValueError( 

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

88 ) 

89 

90 pr_resp = httpx.get( 

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

92 headers=headers, 

93 ) 

94 if pr_resp.status_code >= 400: 

95 raise ValueError( 

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

97 ) 

98 try: 

99 pr_data = pr_resp.json() 

100 except json.JSONDecodeError: 

101 raise ValueError( 

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

103 ) from None 

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

105 if not isinstance(title, str): 

106 raise ValueError( 

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

108 ) 

109 

110 stripped, has_wip_prefix = _strip_wip_prefix(title) 

111 if event == "REQUEST_CHANGES": 

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

113 else: 

114 new_title = stripped if has_wip_prefix else None 

115 

116 wip = "unchanged" 

117 if new_title is not None: 

118 patch_resp = httpx.patch( 

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

120 json={"title": new_title}, 

121 headers=headers, 

122 ) 

123 if patch_resp.status_code >= 400: 

124 raise ValueError( 

125 f"github-pr-comment: title PATCH failed ({patch_resp.status_code}): {patch_resp.text[:500]}" 

126 ) 

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

128 

129 return {**state, "review_posted": True, "findings_count": len(findings), "wip": wip} 

130 

131 

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

133 if not findings: 

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

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

136 for f in findings: 

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

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

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

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

141 if evidence: 

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

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

144 return "\n".join(lines) 

145 

146 

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

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