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

1"""GitHub transport selection and normalized operation capabilities.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6 

7from .runtime import CapabilityReport 

8 

9OPERATIONS: tuple[str, ...] = ( 

10 "issue_read", 

11 "issue_write", 

12 "pr_read", 

13 "pr_write", 

14 "pr_merge", 

15 "check_runs", 

16 "raw_actions_logs", 

17 "labels", 

18 "comments", 

19 "reviews", 

20 "files", 

21) 

22 

23NORMALIZED_FIELDS: tuple[str, ...] = ( 

24 "issue_labels", 

25 "pr_state", 

26 "draft_state", 

27 "mergeable_state", 

28 "check_runs", 

29 "comments", 

30 "reviews", 

31 "files", 

32 "merge_operations", 

33) 

34 

35 

36@dataclass(frozen=True) 

37class GitHubTransport: 

38 """Selected GitHub access path plus normalized operation support.""" 

39 

40 name: str 

41 available: bool 

42 capabilities: dict[str, bool] 

43 degraded: tuple[str, ...] = () 

44 reason: str = "" 

45 

46 def supports(self, operation: str) -> bool: 

47 return self.capabilities.get(operation, False) 

48 

49 def as_dict(self) -> dict: 

50 return { 

51 "transport": self.name, 

52 "available": self.available, 

53 "capabilities": dict(self.capabilities), 

54 "degraded": list(self.degraded), 

55 "normalized_fields": list(NORMALIZED_FIELDS), 

56 "reason": self.reason, 

57 } 

58 

59 def render(self) -> str: 

60 lines = [ 

61 "github transport:", 

62 f" selected: {self.name}", 

63 f" available: {'yes' if self.available else 'no'}", 

64 ] 

65 if self.reason: 

66 lines.append(f" reason: {self.reason}") 

67 if self.degraded: 

68 lines.append(f" degraded: {', '.join(self.degraded)}") 

69 return "\n".join(lines) 

70 

71 

72def resolve(report: CapabilityReport) -> GitHubTransport: 

73 """Resolve the preferred GitHub transport from runtime capabilities. 

74 

75 Local authenticated `gh` wins. When `gh` is unavailable, host-provided GitHub 

76 MCP/API access can still support read/comment/list operations, while operations with 

77 known gaps are surfaced as degraded capabilities. 

78 """ 

79 

80 if report.available("gh") and report.available("gh-auth"): 

81 return GitHubTransport( 

82 "gh", 

83 True, 

84 _caps(True), 

85 reason="authenticated GitHub CLI", 

86 ) 

87 if report.available("github-mcp"): 

88 capabilities = _caps(True) 

89 degraded = ("pr_merge", "check_runs", "raw_actions_logs") 

90 for name in degraded: 

91 capabilities[name] = False 

92 return GitHubTransport( 

93 "mcp", 

94 True, 

95 capabilities, 

96 degraded=degraded, 

97 reason="GitHub MCP/API capability reported by runtime", 

98 ) 

99 return GitHubTransport( 

100 "none", 

101 False, 

102 _caps(False), 

103 degraded=OPERATIONS, 

104 reason="no authenticated GitHub transport detected", 

105 ) 

106 

107 

108def _caps(value: bool) -> dict[str, bool]: 

109 return {name: value for name in OPERATIONS}