Coverage for backend / app / routers / badge.py: 100%

1import re 

2from typing import Annotated 

3 

4from fastapi import APIRouter, Depends, Request, Response 

5from fastapi.responses import RedirectResponse 

6from redis import RedisError 

7from redis.asyncio import Redis 

8 

9from app.constants import BADGE_CACHE_KEY 

10from app.dependencies.gh_client import get_github_client 

11from app.utils.github_client import GithubClient 

12from app.schemas import GhCommitStatus 

13from app.dependencies.redis_client import get_redis_client 

14 

15COV_RE = re.compile(r"([\d.]+)%") 

16 

17 

18router = APIRouter() 

19 

20BADGE_SVG = """\ 

21<svg xmlns="http://www.w3.org/2000/svg" width="110" height="20"> 

22 <title>coverage: {cov}</title> 

23 <defs> 

24 <linearGradient id="workflow-fill" x1="50%" y1="0%" x2="50%" y2="100%"> 

25 <stop stop-color="#444D56" offset="0%"/> 

26 <stop stop-color="#24292E" offset="100%"/> 

27 </linearGradient> 

28 <linearGradient id="state-fill" x1="50%" y1="0%" x2="50%" y2="100%"> 

29 <stop stop-color="{color_top}" offset="0%"/> 

30 <stop stop-color="{color_bot}" offset="100%"/> 

31 </linearGradient> 

32 </defs> 

33 <g font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="11"> 

34 <path d="M0,3 C0,1.3431 1.3552,0 3.02702703,0 L70,0 L70,20 L3.02702703,20 C1.3552,20 0,18.6569 0,17 L0,3 Z" fill="url(#workflow-fill)"/> 

35 <text fill="#010101" fill-opacity=".3"> 

36 <tspan x="10" y="15">coverage</tspan> 

37 </text> 

38 <text fill="#FFF"> 

39 <tspan x="10" y="14">coverage</tspan> 

40 </text> 

41 </g> 

42 <g transform="translate(70)" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="11"> 

43 <path d="M0 0h36.939C38.629 0 40 1.343 40 3v14c0 1.657-1.37 3-3.061 3H0V0z" fill="url(#state-fill)"/> 

44 <text text-anchor="middle" fill="#010101" fill-opacity=".3"> 

45 <tspan x="20" y="15">{cov}</tspan> 

46 </text> 

47 <text text-anchor="middle" fill="#FFF"> 

48 <tspan x="20" y="14">{cov}</tspan> 

49 </text> 

50 </g> 

51</svg> 

52""" 

53 

54 

55def _badge_color( 

56 coverage_percent: float | None, status: GhCommitStatus | None 

57) -> tuple[str, str]: 

58 if coverage_percent is None: 

59 return "#9F9F9F", "#8C8C8C" 

60 assert status is not None 

61 if status.state == "success": 

62 return "#34D058", "#28A745" 

63 return "#CB2431", "#B31D28" 

64 

65 

66async def _get_coverage( 

67 org: str, repo: str, gh_client: GithubClient 

68) -> tuple[float, GhCommitStatus] | tuple[None, None]: 

69 

70 # TODO: add caching 

71 context_pattern = re.compile("coverage", re.IGNORECASE) 

72 commits = await gh_client.get_latest_commits(owner=org, repo=repo, limit=5) 

73 for commit in commits: 

74 if "\n\n[skip ci]" in commit.message: 

75 continue 

76 statuses = await gh_client.get_commit_statuses( 

77 owner=org, repo=repo, sha=commit.sha 

78 ) 

79 for status in statuses: 

80 if context_pattern.search(status.context): 

81 m = COV_RE.search(status.description) 

82 if m: 

83 return float(m.group(1)), status 

84 return None, None 

85 return None, None 

86 

87 

88@router.get("/redirect/{org}/{repo}/") 

89async def redirect( 

90 *, 

91 org: str, 

92 repo: str, 

93 gh_client: Annotated[GithubClient, Depends(get_github_client)], 

94) -> Response: 

95 _, status = await _get_coverage(org, repo, gh_client) 

96 if status: 

97 return RedirectResponse(status.target_url) 

98 return Response(content="Status Not found", status_code=404) 

99 

100 

101def get_response(request: Request, content: str) -> Response: 

102 user_agent = request.headers.get("user-agent", "") 

103 headers = ( 

104 { 

105 "cache-control": "private, no-store", 

106 "cdn-cache-control": "no-store", 

107 } 

108 if user_agent.startswith("github-camo") 

109 else { 

110 "cache-control": "public, max-age=10", 

111 "cdn-cache-control": "max-age=10", 

112 } 

113 ) 

114 return Response( 

115 content=content, 

116 media_type="image/svg+xml", 

117 headers=headers, 

118 ) 

119 

120 

121@router.get("/{org}/{repo}.svg") 

122async def badge( 

123 *, 

124 org: str, 

125 repo: str, 

126 gh_client: Annotated[GithubClient, Depends(get_github_client)], 

127 request: Request, 

128 redis_client: Annotated[Redis, Depends(get_redis_client)], 

129) -> Response: 

130 

131 cache_key = BADGE_CACHE_KEY.format(org=org, repo=repo) 

132 try: 

133 cached = await redis_client.get(cache_key) 

134 except RedisError as e: 

135 print(f"Error accessing cache: {e}") 

136 cached = None 

137 

138 if cached: 

139 return get_response(request, cached.decode("utf-8")) 

140 

141 coverage, status = await _get_coverage(org, repo, gh_client) 

142 

143 color_top, color_bot = _badge_color(coverage, status) 

144 cov_text = f"{coverage:.0f}%" if coverage is not None else "??%" 

145 

146 svg = ( 

147 BADGE_SVG.replace("{cov}", cov_text) 

148 .replace("{color_top}", color_top) 

149 .replace("{color_bot}", color_bot) 

150 ) 

151 

152 try: 

153 await redis_client.set( 

154 cache_key, 

155 svg.encode("utf-8"), 

156 ex=60, 

157 ) 

158 except RedisError as e: 

159 print(f"Error caching data: {e}") 

160 

161 return get_response(request, svg)