Coverage for cli / tests / test_main.py: 100%

1""" 

2Tests for `_main` - orchestration of session creation, upload, status update, and cache 

3purge. 

4""" 

5 

6from pathlib import Path 

7from typing import Any, TypedDict 

8from unittest.mock import AsyncMock, MagicMock 

9 

10import httpx 

11import pytest 

12import typer 

13 

14from covered.cli import _main 

15 

16API_URL = "https://api.example.com" 

17API_KEY = "test_api_key" 

18REPO_OWNER = "owner" 

19REPO_NAME = "repo" 

20COMMIT_SHA = "abc123" 

21GH_TOKEN = "gh_xyz" 

22SITE_ID = "site-xyz" 

23 

24SESSION_RESP = { 

25 "site_id": SITE_ID, 

26 "bucket": "test-bucket", 

27 "region": "us-east-1", 

28 "access_key_id": "id", 

29 "secret_access_key": "secret", 

30 "session_token": "tok", 

31} 

32 

33CAMO_URL = "https://camo.githubusercontent.com/abc123def/xyz789ABC" 

34BADGE_HTML = ( 

35 f'<img alt="cov" src="{CAMO_URL}" ' 

36 f'data-canonical-src="{API_URL}/coverage/badge.svg">' 

37) 

38 

39 

40class PatchedMain(TypedDict): 

41 """ 

42 Handles to the three mocks installed by the `patched_main` fixture. 

43 """ 

44 

45 request: AsyncMock 

46 upload: AsyncMock 

47 coverage: MagicMock 

48 

49 

50def _main_kwargs(directory: Path, **overrides: Any) -> dict: 

51 """ 

52 Build a kwargs dict for `_main`; override only what each test cares about. 

53 """ 

54 return { 

55 "directory": directory, 

56 "api_url": API_URL, 

57 "api_key": API_KEY, 

58 "concurrency": 50, 

59 "repo_owner": REPO_OWNER, 

60 "repo_name": REPO_NAME, 

61 "commit_sha": COMMIT_SHA, 

62 "coverage_threshold": 90.0, 

63 "gh_token": GH_TOKEN, 

64 "purge_cache": False, 

65 **overrides, 

66 } 

67 

68 

69@pytest.fixture 

70def patched_main(monkeypatch: pytest.MonkeyPatch) -> PatchedMain: 

71 """ 

72 Patch `_main`'s three collaborators; defaults: 95% coverage, 5 files uploaded. 

73 """ 

74 request = AsyncMock() 

75 upload = AsyncMock(return_value=5) 

76 coverage = MagicMock(return_value=95.0) 

77 

78 monkeypatch.setattr("covered.cli._request", request) 

79 monkeypatch.setattr("covered.cli._upload_files", upload) 

80 monkeypatch.setattr("covered.cli._get_coverage_info", coverage) 

81 

82 return {"request": request, "upload": upload, "coverage": coverage} 

83 

84 

85async def test_main_creates_session_then_uploads_then_sets_status( 

86 patched_main: PatchedMain, tmp_path: Path 

87): 

88 """ 

89 Happy path: session is created, files uploaded, GitHub status posted. 

90 """ 

91 patched_main["request"].side_effect = [ 

92 httpx.Response(200, json=SESSION_RESP), # create-site 

93 httpx.Response(201), # github status 

94 ] 

95 

96 await _main(**_main_kwargs(tmp_path)) 

97 

98 calls = patched_main["request"].call_args_list 

99 assert len(calls) == 2 

100 assert calls[0].args == ("POST", f"{API_URL}/coverage/create-site/") 

101 patched_main["upload"].assert_awaited_once_with(tmp_path, SESSION_RESP, 50) 

102 assert calls[1].args == ( 

103 "POST", 

104 f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/statuses/{COMMIT_SHA}", 

105 ) 

106 

107 

108async def test_main_sets_success_status_when_coverage_meets_threshold( 

109 patched_main: PatchedMain, tmp_path: Path 

110): 

111 """ 

112 When coverage > threshold, the posted status `state` is `success`. 

113 """ 

114 patched_main["coverage"].return_value = 95.0 

115 patched_main["request"].side_effect = [ 

116 httpx.Response(200, json=SESSION_RESP), # create-site 

117 httpx.Response(201), # github status 

118 ] 

119 

120 await _main(**_main_kwargs(tmp_path, coverage_threshold=90.0)) 

121 

122 assert ( 

123 patched_main["request"].call_args_list[1].kwargs["json"]["state"] == "success" 

124 ) 

125 

126 

127async def test_main_sets_failure_status_when_coverage_below_threshold( 

128 patched_main: PatchedMain, tmp_path: Path 

129): 

130 """ 

131 When coverage < threshold, the posted status `state` is `failure`. 

132 """ 

133 patched_main["coverage"].return_value = 80.0 

134 patched_main["request"].side_effect = [ 

135 httpx.Response(200, json=SESSION_RESP), # create-site 

136 httpx.Response(201), # github status 

137 ] 

138 

139 await _main(**_main_kwargs(tmp_path, coverage_threshold=90.0)) 

140 

141 assert ( 

142 patched_main["request"].call_args_list[1].kwargs["json"]["state"] == "failure" 

143 ) 

144 

145 

146async def test_main_status_at_threshold_is_success( 

147 patched_main: PatchedMain, tmp_path: Path 

148): 

149 """ 

150 Boundary: coverage exactly equal to threshold yields `success`, not `failure`. 

151 """ 

152 patched_main["coverage"].return_value = 90.0 

153 patched_main["request"].side_effect = [ 

154 httpx.Response(200, json=SESSION_RESP), # create-site 

155 httpx.Response(201), # github status 

156 ] 

157 

158 await _main(**_main_kwargs(tmp_path, coverage_threshold=90.0)) 

159 

160 assert ( 

161 patched_main["request"].call_args_list[1].kwargs["json"]["state"] == "success" 

162 ) 

163 

164 

165async def test_main_skips_status_update_when_coverage_missing( 

166 patched_main: PatchedMain, tmp_path: Path 

167): 

168 """ 

169 If `_get_coverage_info` returns None, no GitHub status call is made and `_main` 

170 returns early. 

171 """ 

172 patched_main["coverage"].return_value = None 

173 patched_main["request"].side_effect = [ 

174 httpx.Response(200, json=SESSION_RESP), # create-site 

175 ] 

176 

177 await _main(**_main_kwargs(tmp_path)) 

178 

179 assert patched_main["request"].call_count == 1 

180 

181 

182async def test_main_exits_with_error_when_status_post_fails( 

183 patched_main: PatchedMain, tmp_path: Path 

184): 

185 """ 

186 Non-201 response from the GitHub status endpoint raises `typer.Exit(1)`. 

187 """ 

188 patched_main["request"].side_effect = [ 

189 httpx.Response(200, json=SESSION_RESP), # create-site 

190 httpx.Response(500, text="oops"), # github status failed 

191 ] 

192 

193 with pytest.raises(typer.Exit) as exc_info: 

194 await _main(**_main_kwargs(tmp_path)) 

195 

196 assert exc_info.value.exit_code == 1 

197 

198 

199async def test_main_status_target_url_points_at_site( 

200 patched_main: PatchedMain, tmp_path: Path 

201): 

202 """ 

203 The status payload's `target_url` is `{api_url}/coverage/{site_id}/`. 

204 """ 

205 patched_main["request"].side_effect = [ 

206 httpx.Response(200, json=SESSION_RESP), # create-site 

207 httpx.Response(201), # github status 

208 ] 

209 

210 await _main(**_main_kwargs(tmp_path)) 

211 

212 payload = patched_main["request"].call_args_list[1].kwargs["json"] 

213 assert payload["target_url"] == f"{API_URL}/coverage/{SITE_ID}/" 

214 

215 

216async def test_main_status_request_headers(patched_main: PatchedMain, tmp_path: Path): 

217 """ 

218 The status request uses `Authorization: Bearer <gh_token>` and 

219 `Accept: application/vnd.github+json`. 

220 """ 

221 patched_main["request"].side_effect = [ 

222 httpx.Response(200, json=SESSION_RESP), # create-site 

223 httpx.Response(201), # github status 

224 ] 

225 

226 await _main(**_main_kwargs(tmp_path)) 

227 

228 headers = patched_main["request"].call_args_list[1].kwargs["headers"] 

229 assert headers["Authorization"] == f"Bearer {GH_TOKEN}" 

230 assert headers["Accept"] == "application/vnd.github+json" 

231 

232 

233async def test_main_skips_cache_purge_when_purge_cache_false( 

234 patched_main: PatchedMain, tmp_path: Path 

235): 

236 """ 

237 When `purge_cache=False`, no invalidate-cache or README calls are made. 

238 """ 

239 patched_main["request"].side_effect = [ 

240 httpx.Response(200, json=SESSION_RESP), # create-site 

241 httpx.Response(201), # github status 

242 ] 

243 

244 await _main(**_main_kwargs(tmp_path, purge_cache=False)) 

245 

246 assert patched_main["request"].call_count == 2 

247 

248 

249async def test_main_invalidates_cache_when_purge_cache_true( 

250 patched_main: PatchedMain, tmp_path: Path 

251): 

252 """ 

253 When `purge_cache=True`, POST to /invalidate-cache/{owner}/{repo}/ with the token 

254 header. 

255 """ 

256 patched_main["request"].side_effect = [ 

257 httpx.Response(200, json=SESSION_RESP), # create-site 

258 httpx.Response(201), # github status 

259 httpx.Response(200), # invalidate-cache 

260 httpx.Response(200, text=""), # README (no badge match -> early return) 

261 ] 

262 

263 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

264 

265 invalidate = patched_main["request"].call_args_list[2] 

266 assert invalidate.args == ( 

267 "POST", 

268 f"{API_URL}/coverage/invalidate-cache/{REPO_OWNER}/{REPO_NAME}/", 

269 ) 

270 assert invalidate.kwargs["headers"] == {"token": API_KEY} 

271 

272 

273async def test_main_logs_failure_but_continues_when_invalidate_cache_fails( 

274 patched_main: PatchedMain, tmp_path: Path 

275): 

276 """ 

277 A non-200 invalidate-cache response is logged but the flow continues to Camo purge. 

278 """ 

279 patched_main["request"].side_effect = [ 

280 httpx.Response(200, json=SESSION_RESP), # create-site 

281 httpx.Response(201), # github status 

282 httpx.Response(500, text="oops"), # invalidate-cache failed 

283 httpx.Response(200, text=""), # README still attempted 

284 ] 

285 

286 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

287 

288 # 4 calls: create-site, status, invalidate (failed), README 

289 assert patched_main["request"].call_count == 4 

290 

291 

292async def test_main_fetches_readme_with_html_accept_header( 

293 patched_main: PatchedMain, tmp_path: Path 

294): 

295 """ 

296 The README request uses `Accept: application/vnd.github.html+json`. 

297 """ 

298 patched_main["request"].side_effect = [ 

299 httpx.Response(200, json=SESSION_RESP), # create-site 

300 httpx.Response(201), # github status 

301 httpx.Response(200), # invalidate-cache 

302 httpx.Response(200, text=""), # README (no badge match -> early return) 

303 ] 

304 

305 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

306 

307 readme = patched_main["request"].call_args_list[3] 

308 assert readme.args == ( 

309 "GET", 

310 f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/readme", 

311 ) 

312 assert readme.kwargs["headers"]["Accept"] == "application/vnd.github.html+json" 

313 assert readme.kwargs["headers"]["Authorization"] == f"Bearer {GH_TOKEN}" 

314 

315 

316async def test_main_purge_aborts_when_readme_fetch_fails( 

317 patched_main: PatchedMain, tmp_path: Path 

318): 

319 """ 

320 A non-200 README response is logged and `_main` returns without sending PURGE. 

321 """ 

322 patched_main["request"].side_effect = [ 

323 httpx.Response(200, json=SESSION_RESP), # create-site 

324 httpx.Response(201), # github status 

325 httpx.Response(200), # invalidate-cache 

326 httpx.Response(404), # README fetch failed 

327 ] 

328 

329 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

330 

331 assert patched_main["request"].call_count == 4 

332 

333 

334async def test_main_purge_aborts_when_badge_not_found_in_readme( 

335 patched_main: PatchedMain, tmp_path: Path 

336): 

337 """ 

338 If the README has no matching `<img>`, `_main` logs and returns - no PURGE call. 

339 """ 

340 patched_main["request"].side_effect = [ 

341 httpx.Response(200, json=SESSION_RESP), # create-site 

342 httpx.Response(201), # github status 

343 httpx.Response(200), # invalidate-cache 

344 httpx.Response(200, text="<html>no badge here</html>"), # README (no badge) 

345 ] 

346 

347 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

348 

349 assert patched_main["request"].call_count == 4 

350 

351 

352async def test_main_purges_camo_url_extracted_from_readme( 

353 patched_main: PatchedMain, tmp_path: Path 

354): 

355 """ 

356 PURGE is sent to the Camo URL captured by the badge regex from the README. 

357 """ 

358 patched_main["request"].side_effect = [ 

359 httpx.Response(200, json=SESSION_RESP), # create-site 

360 httpx.Response(201), # github status 

361 httpx.Response(200), # invalidate-cache 

362 httpx.Response(200, text=BADGE_HTML), # README with matching badge 

363 httpx.Response(200), # PURGE 

364 ] 

365 

366 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

367 

368 purge = patched_main["request"].call_args_list[4] 

369 assert purge.args == ("PURGE", CAMO_URL) 

370 

371 

372async def test_main_badge_regex_only_matches_camo_with_matching_canonical_src( 

373 patched_main: PatchedMain, tmp_path: Path 

374): 

375 """ 

376 A Camo `<img>` whose `data-canonical-src` does not match `api_url` is ignored. 

377 """ 

378 other_html = ( 

379 f'<img src="{CAMO_URL}" data-canonical-src="https://other.example.com/badge">' 

380 ) 

381 patched_main["request"].side_effect = [ 

382 httpx.Response(200, json=SESSION_RESP), # create-site 

383 httpx.Response(201), # github status 

384 httpx.Response(200), # invalidate-cache 

385 httpx.Response(200, text=other_html), # README - badge points elsewhere 

386 ] 

387 

388 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

389 

390 assert patched_main["request"].call_count == 4 

391 

392 

393async def test_main_logs_failure_when_camo_purge_returns_non_200( 

394 patched_main: PatchedMain, tmp_path: Path 

395): 

396 """ 

397 A non-200 PURGE response is logged; `_main` does not raise or exit non-zero. 

398 """ 

399 patched_main["request"].side_effect = [ 

400 httpx.Response(200, json=SESSION_RESP), # create-site 

401 httpx.Response(201), # github status 

402 httpx.Response(200), # invalidate-cache 

403 httpx.Response(200, text=BADGE_HTML), # README with matching badge 

404 httpx.Response(500), # PURGE failed 

405 ] 

406 

407 await _main(**_main_kwargs(tmp_path, purge_cache=True)) 

408 

409 assert patched_main["request"].call_count == 5 

410 

411 

412async def test_main_create_site_uses_api_key_header_and_120s_timeout( 

413 patched_main: PatchedMain, tmp_path: Path 

414): 

415 """ 

416 The create-site request sends the `token` header and uses the longer 120s timeout. 

417 """ 

418 patched_main["request"].side_effect = [ 

419 httpx.Response(200, json=SESSION_RESP), # create-site 

420 httpx.Response(201), # github status 

421 ] 

422 

423 await _main(**_main_kwargs(tmp_path)) 

424 

425 create_call = patched_main["request"].call_args_list[0] 

426 assert create_call.kwargs["headers"] == {"token": API_KEY} 

427 assert create_call.kwargs["timeout"] == 120