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

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 if self.gh_cfg.access_token: 

25 self.log.debug("Using configured bearer token (%s chars) for GitHub API", len(self.gh_cfg.access_token)) 

26 

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

28 

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

30 return 

31 

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

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

34 if image.image_digest and image.index_name == REGISTRY_GHCR: 34 ↛ 49line 34 didn't jump to line 49 because the condition on line 34 was always true

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

36 if matched: 36 ↛ 41line 36 didn't jump to line 41 because the condition on line 36 was never true

37 tags: list[str] 

38 _release_url: str | None 

39 created_ts: str | None 

40 updated_ts: str | None 

41 tags, _release_url, created_ts, updated_ts = matched 

42 for t in tags: 

43 if re.fullmatch(VERSION_RE, t): 

44 self.log.debug( 

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

46 ) 

47 detail.version = t 

48 

49 if detail.version is not None: 

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

51 

52 api_response: Response | None = fetch_url( 

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

54 bearer_token=self.gh_cfg.access_token, 

55 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

56 allow_stale=True, # not assuming immutable tags 

57 ) 

58 if api_response and api_response.status_code == 404: 

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

60 alt_api_response: Response | None = fetch_url( 

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

62 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

63 bearer_token=self.gh_cfg.access_token, 

64 ) 

65 if alt_api_response and alt_api_response.is_success: 65 ↛ 80line 65 didn't jump to line 80 because the condition on line 65 was always true

66 alt_api_results = httpx_json_content(alt_api_response, {}) 

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

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

69 api_response = alt_api_response 

70 elif alt_api_results: 70 ↛ 80line 70 didn't jump to line 80 because the condition on line 70 was always true

71 self.log.debug( 

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

73 image.name, 

74 detail.version, 

75 alt_api_results.get("tag_name"), 

76 alt_api_results.get("name"), 

77 alt_api_results.get("published_at"), 

78 ) 

79 

80 if api_response and api_response.is_success: 

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

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

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

84 if reactions: 

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

86 return 

87 if api_response: 

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

89 self.log.debug( 

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

91 image.name, 

92 detail.version, 

93 api_response.status_code, 

94 api_results.get("errors"), 

95 ) 

96 else: 

97 self.log.debug( 

98 "Failed to fetch GitHub release info", 

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

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

101 ) 

102 

103 def match_packages( 

104 self, package: str, package_digest: str, source_repo_url: str 

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

106 

107 if not self.gh_cfg.access_token: 107 ↛ 111line 107 didn't jump to line 111 because the condition on line 107 was always true

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

109 return None 

110 

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

112 if not match: 

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

114 return None 

115 org_or_user: str = match.group(1) 

116 

117 api_result: Response | None = fetch_url( 

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

119 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

120 bearer_token=self.gh_cfg.access_token, 

121 ) 

122 if api_result and api_result.status_code == 404: 

123 api_result = fetch_url( 

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

125 cache_ttl=self.gh_cfg.mutable_cache_ttl, 

126 bearer_token=self.gh_cfg.access_token, 

127 ) 

128 if not api_result: 

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

130 return None 

131 if not api_result.is_success: 

132 self.log.warn( 

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

134 ) 

135 if api_result.status_code == 401: 

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

137 self.gh_cfg.access_token = None 

138 return None 

139 

140 pkg_releases = api_result.json() 

141 for release in pkg_releases: 

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

143 self.log.info( 

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

145 package, 

146 release.get("id"), 

147 release.get("metadata"), 

148 ) 

149 return ( 

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

151 release.get("html_url"), 

152 release.get("created_at"), 

153 release.get("updated_at"), 

154 ) 

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

156 return None