Coverage for src / updates2mqtt / integrations / github_enrich.py: 100%

1import re 

2from typing import TYPE_CHECKING, Any 

3 

4import structlog 

5 

6from updates2mqtt.config import ( 

7 VERSION_RE, 

8 GitHubConfig, 

9) 

10from updates2mqtt.helpers import fetch_url, httpx_json_content 

11from updates2mqtt.integrations.docker_enrich import REGISTRY_GHCR, DockerImageInfo 

12from updates2mqtt.model import ReleaseDetail 

13 

14if TYPE_CHECKING: 

15 from httpx import Response 

16 

17log: Any = structlog.get_logger() 

18 

19 

20class GithubReleaseEnricher: 

21 def __init__(self, gh_cfg: GitHubConfig) -> None: 

22 self.log: Any = structlog.get_logger().bind(integration="github") 

23 self.gh_cfg: GitHubConfig = gh_cfg 

24 self.gh_token: str | None = self.gh_cfg.access_token 

25 if self.gh_token: 

26 self.log.debug("Using configured bearer token (%s chars) for GitHub API", len(self.gh_token)) 

27 

28 def enrich(self, image: DockerImageInfo, detail: ReleaseDetail) -> None: 

29 

30 if not detail.source_repo_url or not image.name: 

31 return 

32 

33 if not detail.version or not re.fullmatch(VERSION_RE, detail.version): 

34 self.log.debug("No valid version found for GitHub release %s: %s", image.name, detail.version) 

35 if image.image_digest and image.index_name == REGISTRY_GHCR: 

36 matched = self.match_packages(image.unqualified_name, image.image_digest, detail.source_repo_url) 

37 if matched: 

38 tags: list[str] 

39 _release_url: str | None 

40 created_ts: str | None 

41 updated_ts: str | None 

42 tags, _release_url, created_ts, updated_ts = matched 

43 for t in tags: 

44 if re.fullmatch(VERSION_RE, t): 

45 self.log.debug( 

46 "Matched %s version %s created on %s last updated %s", image.name, t, created_ts, updated_ts 

47 ) 

48 detail.version = t 

49 

50 if detail.version is not None: 

51 base_api = detail.source_repo_url.replace("https://github.com", "https://api.github.com/repos") 

52 

53 api_response: Response | None = fetch_url( 

54 f"{base_api}/releases/tags/{detail.version}", 

55 bearer_token=self.gh_token, 

56 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

57 allow_stale=True, # not assuming immutable tags 

58 ) 

59 if api_response and api_response.status_code == 404: 

60 # possible that source version doesn't match release gag 

61 alt_api_response: Response | None = fetch_url( 

62 f"{base_api}/releases/latest", 

63 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

64 bearer_token=self.gh_token, 

65 ) 

66 if alt_api_response and alt_api_response.is_success: 

67 alt_api_results = httpx_json_content(alt_api_response, {}) 

68 if alt_api_results and re.fullmatch(f"(V|v|r|R)?{detail.version}", alt_api_results.get("tag_name")): 

69 self.log.info(f"Matched {image.name} {detail.version} to latest release {alt_api_results['tag_name']}") 

70 api_response = alt_api_response 

71 elif alt_api_results: 

72 self.log.debug( 

73 "Failed to match %s release %s on GitHub, found tag %s for name %s published at %s", 

74 image.name, 

75 detail.version, 

76 alt_api_results.get("tag_name"), 

77 alt_api_results.get("name"), 

78 alt_api_results.get("published_at"), 

79 ) 

80 

81 if api_response and api_response.is_success: 

82 api_results: Any = httpx_json_content(api_response, {}) 

83 detail.summary = api_results.get("body") # ty:ignore[possibly-missing-attribute] 

84 reactions = api_results.get("reactions") # ty:ignore[possibly-missing-attribute] 

85 if reactions: 

86 detail.net_score = reactions.get("+1", 0) - reactions.get("-1", 0) 

87 return 

88 if api_response: 

89 api_results = httpx_json_content(api_response, default={}) 

90 self.log.debug( 

91 "Failed to find %s release %s on GitHub, status %s, errors; %s", 

92 image.name, 

93 detail.version, 

94 api_response.status_code, 

95 api_results.get("errors"), 

96 ) 

97 else: 

98 self.log.debug( 

99 "Failed to fetch GitHub release info", 

100 url=f"{base_api}/releases/tags/{detail.version}", 

101 status_code=(api_response and api_response.status_code) or None, 

102 ) 

103 

104 def match_packages( 

105 self, package: str, package_digest: str, source_repo_url: str 

106 ) -> tuple[list[str], str | None, str | None, str | None] | None: 

107 

108 if not self.gh_token: 

109 self.log.debug("No access token available for packages API") 

110 return None 

111 

112 match = re.match(r"https://github\.com/([^/]+)/(?:[^/]+?)(?:\.git)?$", source_repo_url) 

113 if not match: 

114 self.log.debug("Invalid source repo URL for GitHub API use: %s", source_repo_url) 

115 return None 

116 org_or_user: str = match.group(1) 

117 

118 api_result: Response | None = fetch_url( 

119 f"https://api.github.com/orgs/{org_or_user}/packages/container/{package}/versions", 

120 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

121 bearer_token=self.gh_token, 

122 ) 

123 if api_result and api_result.status_code == 404: 

124 api_result = fetch_url( 

125 f"https://api.github.com/users/{org_or_user}/packages/container/{package}/versions", 

126 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

127 bearer_token=self.gh_token, 

128 ) 

129 if not api_result: 

130 self.log.warn("Unable to retrieve GitHub packages API result, null response") 

131 return None 

132 if not api_result.is_success: 

133 self.log.warn( 

134 "Unable to retrieve GitHub packages API result, %s response: %s", api_result.status_code, api_result.text 

135 ) 

136 if api_result.status_code == 401: 

137 self.log.warn("Disabling Github access token for this session") 

138 self.gh_token = None 

139 return None 

140 

141 pkg_releases = api_result.json() 

142 for release in pkg_releases: 

143 if release.get("name") == package_digest: 

144 self.log.info( 

145 "Matched %s image digest using GitHub API for release %s, metadata %s", 

146 package, 

147 release.get("id"), 

148 release.get("metadata"), 

149 ) 

150 return ( 

151 release.get("metadata", {}).get("container", {}).get("tags", []), 

152 release.get("html_url"), 

153 release.get("created_at"), 

154 release.get("updated_at"), 

155 ) 

156 self.log.debug("No matching %s release found using GitHub API for %s", package, package_digest) 

157 return None