Coverage for app/backend/src/tests/test_bugs.py: 99%

1import json 

2import uuid 

3from datetime import UTC, datetime, timedelta 

4from unittest.mock import patch 

5 

6import grpc 

7import pytest 

8from google.protobuf import empty_pb2, timestamp_pb2 

9from sqlalchemy import func, select 

10 

11from couchers.config import config 

12from couchers.crypto import random_hex 

13from couchers.db import session_scope 

14from couchers.models import NativeClientUser, OTAPackage, OTAPlatform 

15from couchers.models.logging import EventLog, EventSource, ExperimentExposure, ExposureSource 

16from couchers.proto import bugs_pb2 

17from couchers.proto.google.api import httpbody_pb2 

18from couchers.servicers.bugs import _fetch_signed_manifest 

19from tests.fixtures.db import generate_user 

20from tests.fixtures.sessions import bugs_session, real_bugs_session 

21 

22EAS_CLIENT_ID = uuid.UUID("11111111-1111-1111-1111-111111111111") 

23 

24 

25def test_bugs_disabled(): 

26 with bugs_session() as bugs, pytest.raises(grpc.RpcError) as e: 

27 bugs.ReportBug( 

28 bugs_pb2.ReportBugReq( 

29 subject="subject", 

30 description="description", 

31 results="results", 

32 frontend_version="frontend_version", 

33 user_agent="user_agent", 

34 page="page", 

35 ) 

36 ) 

37 assert e.value.code() == grpc.StatusCode.UNAVAILABLE 

38 

39 

40def test_bugs(db): 

41 with bugs_session() as bugs: 

42 

43 def dud_post(url, auth, json): 

44 assert url == "https://api.github.com/repos/org/repo/issues" 

45 assert auth == ("user", "token") 

46 

47 expected_body = f""" 

48# subject 

49## Description 

50description 

51 

52## Results 

53results 

54 

55## Diagnostics 

56**Backend version**: `{config.VERSION}` 

57**Frontend version**: `frontend_version` 

58**User Agent**: `user_agent` 

59**Locale**: `en` 

60**Screen resolution**: 1920x1080 

61**Page**: page 

62**User**: <not logged in> / `test_sofa_co`""".strip() 

63 

64 assert json == { 

65 "title": "subject", 

66 "body": expected_body, 

67 "labels": ["bug: triage needed"], 

68 } 

69 

70 class _PostReturn: 

71 status_code = 201 

72 

73 def json(self): 

74 return {"number": 11} 

75 

76 return _PostReturn() 

77 

78 config.BUG_TOOL_ENABLED = True 

79 

80 with patch("couchers.servicers.bugs.requests.post", dud_post): 

81 res = bugs.ReportBug( 

82 bugs_pb2.ReportBugReq( 

83 subject="subject", 

84 description="description", 

85 results="results", 

86 frontend_version="frontend_version", 

87 user_agent="user_agent", 

88 screen_resolution=bugs_pb2.ScreenResolution(width=1920, height=1080), 

89 page="page", 

90 ) 

91 ) 

92 

93 assert res.bug_id == "#11" 

94 assert res.bug_url == "https://github.com/org/repo/issues/11" 

95 

96 

97def test_bugs_with_user(db, frozen_timewarp): 

98 user, token = generate_user(username="testing_user") 

99 

100 with bugs_session(token) as bugs: 

101 

102 def dud_post(url, auth, json): 

103 assert url == "https://api.github.com/repos/org/repo/issues" 

104 assert auth == ("user", "token") 

105 

106 expected_body = f""" 

107# subject 

108## Description 

109description 

110 

111## Results 

112results 

113 

114## Diagnostics 

115**Backend version**: `{config.VERSION}` 

116**Frontend version**: `frontend_version` 

117**User Agent**: `user_agent` 

118**Locale**: `en` 

119**Screen resolution**: 390x844 

120**Page**: page 

121**User**: [@testing_user](http://localhost:3000/user/testing_user) (1) / `test_sofa_co` 

122**Sentry (this user)**: https://couchers.sentry.io/issues/?project=1234&query=user.id%3A1&start=2019-12-31T00%3A00%3A00&end=2020-01-01T01%3A00%3A00&utc=true""".strip() 

123 

124 assert json == { 

125 "title": "subject", 

126 "body": expected_body, 

127 "labels": ["bug: triage needed"], 

128 } 

129 

130 class _PostReturn: 

131 status_code = 201 

132 

133 def json(self): 

134 return {"number": 11} 

135 

136 return _PostReturn() 

137 

138 config.BUG_TOOL_ENABLED = True 

139 

140 with patch("couchers.servicers.bugs.requests.post", dud_post): 

141 res = bugs.ReportBug( 

142 bugs_pb2.ReportBugReq( 

143 subject="subject", 

144 description="description", 

145 results="results", 

146 frontend_version="frontend_version", 

147 user_agent="user_agent", 

148 screen_resolution=bugs_pb2.ScreenResolution(width=390, height=844), 

149 page="page", 

150 ) 

151 ) 

152 

153 assert res.bug_id == "#11" 

154 assert res.bug_url == "https://github.com/org/repo/issues/11" 

155 

156 

157def test_bugs_with_sentry_replay(db): 

158 with bugs_session() as bugs: 

159 

160 def dud_post(url, auth, json): 

161 expected_body = f""" 

162# subject 

163## Description 

164description 

165 

166## Results 

167results 

168 

169## Diagnostics 

170**Backend version**: `{config.VERSION}` 

171**Frontend version**: `frontend_version` 

172**User Agent**: `user_agent` 

173**Locale**: `en` 

174**Screen resolution**: 1920x1080 

175**Page**: page 

176**User**: <not logged in> / `test_sofa_co` 

177**Session replay**: https://couchers.sentry.io/replays/0123456789abcdef0123456789abcdef/?project=1234""".strip() 

178 

179 assert json["body"] == expected_body 

180 

181 class _PostReturn: 

182 status_code = 201 

183 

184 def json(self): 

185 return {"number": 11} 

186 

187 return _PostReturn() 

188 

189 config.BUG_TOOL_ENABLED = True 

190 

191 with patch("couchers.servicers.bugs.requests.post", dud_post): 

192 bugs.ReportBug( 

193 bugs_pb2.ReportBugReq( 

194 subject="subject", 

195 description="description", 

196 results="results", 

197 frontend_version="frontend_version", 

198 user_agent="user_agent", 

199 screen_resolution=bugs_pb2.ScreenResolution(width=1920, height=1080), 

200 page="page", 

201 sentry_replay_id="0123456789abcdef0123456789abcdef", 

202 ) 

203 ) 

204 

205 

206def test_bugs_invalid_sentry_replay_id_omitted(db): 

207 # A malformed/garbage replay id must not be interpolated into the issue at all. 

208 with bugs_session() as bugs: 

209 

210 def dud_post(url, auth, json): 

211 assert "Session replay" not in json["body"] 

212 

213 class _PostReturn: 

214 status_code = 201 

215 

216 def json(self): 

217 return {"number": 11} 

218 

219 return _PostReturn() 

220 

221 config.BUG_TOOL_ENABLED = True 

222 

223 with patch("couchers.servicers.bugs.requests.post", dud_post): 

224 bugs.ReportBug( 

225 bugs_pb2.ReportBugReq( 

226 subject="subject", 

227 description="description", 

228 results="results", 

229 sentry_replay_id="not-a-replay-id](https://evil.example)", 

230 ) 

231 ) 

232 

233 

234def test_bugs_fails_on_network_error(db): 

235 with bugs_session() as bugs: 

236 

237 def dud_post(url, auth, json): 

238 class _PostReturn: 

239 status_code = 400 

240 

241 return _PostReturn() 

242 

243 config.BUG_TOOL_ENABLED = True 

244 

245 with patch("couchers.servicers.bugs.requests.post", dud_post): 

246 with pytest.raises(grpc.RpcError) as e: 

247 res = bugs.ReportBug( 

248 bugs_pb2.ReportBugReq( 

249 subject="subject", 

250 description="description", 

251 results="results", 

252 frontend_version="frontend_version", 

253 user_agent="user_agent", 

254 page="page", 

255 ) 

256 ) 

257 assert e.value.code() == grpc.StatusCode.INTERNAL 

258 

259 

260def test_version(): 

261 with bugs_session() as bugs: 

262 res = bugs.Version(empty_pb2.Empty()) 

263 assert res.version == "testing_version" 

264 

265 

266def test_status(db): 

267 for _ in range(5): 

268 generate_user() 

269 

270 with bugs_session() as bugs: 

271 nonce = random_hex() 

272 res = bugs.Status(bugs_pb2.StatusReq(nonce=nonce)) 

273 assert res.nonce == nonce 

274 assert res.version == "testing_version" 

275 assert res.coucher_count == 5 

276 

277 

278def test_GetDescriptors(): 

279 with bugs_session() as bugs: 

280 res = bugs.GetDescriptors(empty_pb2.Empty()) 

281 # test we got something roughly binary back 

282 assert res.content_type == "application/octet-stream" 

283 assert len(res.data) > 2**12 

284 

285 

286def _get_events(session, event_type=None): 

287 stmt = select(EventLog).order_by(EventLog.id) 

288 if event_type: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true

289 stmt = stmt.where(EventLog.event_type == event_type) 

290 return session.execute(stmt).scalars().all() 

291 

292 

293def test_report_diagnostics_anonymous(db): 

294 with bugs_session() as bugs: 

295 bugs.ReportDiagnostics( 

296 bugs_pb2.ReportDiagnosticsReq( 

297 frontend_version="1.2.3", 

298 infos=[ 

299 bugs_pb2.DiagnosticInfo( 

300 tag="page.viewed", 

301 properties_json='{"path": "/"}', 

302 value=1, 

303 ), 

304 bugs_pb2.DiagnosticInfo( 

305 tag="session.started", 

306 properties_json='{"referrer": "google.com"}', 

307 value=1, 

308 ), 

309 ], 

310 ) 

311 ) 

312 

313 with session_scope() as session: 

314 events = _get_events(session) 

315 assert len(events) == 2 

316 

317 e0 = events[0] 

318 assert e0.event_type == "page.viewed" 

319 assert e0.properties == {"path": "/"} 

320 assert e0.user_id is None 

321 assert e0.source == EventSource.frontend 

322 assert e0.value == 1 

323 assert e0.version == "1.2.3" 

324 

325 e1 = events[1] 

326 assert e1.event_type == "session.started" 

327 assert e1.properties == {"referrer": "google.com"} 

328 assert e1.source == EventSource.frontend 

329 

330 

331def test_report_diagnostics_authenticated(db): 

332 user, token = generate_user() 

333 

334 with bugs_session(token) as bugs: 

335 bugs.ReportDiagnostics( 

336 bugs_pb2.ReportDiagnosticsReq( 

337 frontend_version="1.2.3", 

338 infos=[ 

339 bugs_pb2.DiagnosticInfo( 

340 tag="page.viewed", 

341 properties_json='{"path": "/search"}', 

342 value=1, 

343 ), 

344 ], 

345 ) 

346 ) 

347 

348 with session_scope() as session: 

349 events = _get_events(session) 

350 assert len(events) == 1 

351 assert events[0].user_id == user.id 

352 assert events[0].source == EventSource.frontend 

353 

354 

355def test_report_diagnostics_with_value(db): 

356 with bugs_session() as bugs: 

357 bugs.ReportDiagnostics( 

358 bugs_pb2.ReportDiagnosticsReq( 

359 frontend_version="1.2.3", 

360 infos=[ 

361 bugs_pb2.DiagnosticInfo( 

362 tag="search.result_hovered", 

363 properties_json='{"user_id": 5}', 

364 value=1500.5, 

365 ), 

366 ], 

367 ) 

368 ) 

369 

370 with session_scope() as session: 

371 events = _get_events(session) 

372 assert len(events) == 1 

373 assert events[0].value == pytest.approx(1500.5) 

374 

375 

376def test_report_diagnostics_with_occurred(db): 

377 ts = timestamp_pb2.Timestamp() 

378 ts.FromDatetime(datetime(2026, 1, 15, 10, 30, 0, tzinfo=UTC)) 

379 

380 with bugs_session() as bugs: 

381 bugs.ReportDiagnostics( 

382 bugs_pb2.ReportDiagnosticsReq( 

383 frontend_version="1.2.3", 

384 infos=[ 

385 bugs_pb2.DiagnosticInfo( 

386 tag="page.viewed", 

387 properties_json="{}", 

388 value=1, 

389 occurred=ts, 

390 ), 

391 ], 

392 ) 

393 ) 

394 

395 with session_scope() as session: 

396 events = _get_events(session) 

397 assert len(events) == 1 

398 assert events[0].occurred.year == 2026 

399 assert events[0].occurred.month == 1 

400 assert events[0].occurred.day == 15 

401 assert events[0].occurred.hour == 10 

402 assert events[0].occurred.minute == 30 

403 

404 

405def test_report_diagnostics_invalid_json(db): 

406 with bugs_session() as bugs, pytest.raises(grpc.RpcError) as e: 

407 bugs.ReportDiagnostics( 

408 bugs_pb2.ReportDiagnosticsReq( 

409 frontend_version="1.2.3", 

410 infos=[ 

411 bugs_pb2.DiagnosticInfo( 

412 tag="page.viewed", 

413 properties_json="not valid json{{{", 

414 value=1, 

415 ), 

416 ], 

417 ) 

418 ) 

419 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

420 

421 

422def test_report_diagnostics_empty_batch(db): 

423 with bugs_session() as bugs: 

424 bugs.ReportDiagnostics( 

425 bugs_pb2.ReportDiagnosticsReq( 

426 frontend_version="1.2.3", 

427 infos=[], 

428 ) 

429 ) 

430 

431 with session_scope() as session: 

432 events = _get_events(session) 

433 assert len(events) == 0 

434 

435 

436def test_report_diagnostics_too_many(db): 

437 infos = [bugs_pb2.DiagnosticInfo(tag=f"event.{i}", properties_json="{}", value=1) for i in range(101)] 

438 

439 with bugs_session() as bugs, pytest.raises(grpc.RpcError) as e: 

440 bugs.ReportDiagnostics( 

441 bugs_pb2.ReportDiagnosticsReq( 

442 frontend_version="1.2.3", 

443 infos=infos, 

444 ) 

445 ) 

446 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

447 

448 

449def test_report_diagnostics_frontend_version(db): 

450 with bugs_session() as bugs: 

451 bugs.ReportDiagnostics( 

452 bugs_pb2.ReportDiagnosticsReq( 

453 frontend_version="abc-def-123", 

454 infos=[ 

455 bugs_pb2.DiagnosticInfo( 

456 tag="page.viewed", 

457 properties_json="{}", 

458 value=1, 

459 ), 

460 ], 

461 ) 

462 ) 

463 

464 with session_scope() as session: 

465 events = _get_events(session) 

466 assert len(events) == 1 

467 assert events[0].version == "abc-def-123" 

468 

469 

470def test_check_native_status_anonymous(db): 

471 with bugs_session() as bugs: 

472 res = bugs.CheckNativeStatus( 

473 bugs_pb2.CheckNativeStatusReq( 

474 eas_client_id=str(EAS_CLIENT_ID), app_version="1.1.20", platform="ios", user_state="logged_out" 

475 ) 

476 ) 

477 

478 # No build timestamps reported -> no clock runs -> no update asked for. 

479 assert res.update_info.action == bugs_pb2.NATIVE_UPDATE_ACTION_NONE 

480 assert res.update_info.required is False 

481 

482 

483def test_check_native_status_authenticated(db): 

484 _, token = generate_user() 

485 

486 with bugs_session(token) as bugs: 

487 res = bugs.CheckNativeStatus( 

488 bugs_pb2.CheckNativeStatusReq( 

489 eas_client_id=str(EAS_CLIENT_ID), platform="android", user_state="authenticated" 

490 ) 

491 ) 

492 

493 assert res.update_info.action == bugs_pb2.NATIVE_UPDATE_ACTION_NONE 

494 assert res.update_info.required is False 

495 

496 

497def test_check_native_status_authenticated_records_mapping(db): 

498 user, token = generate_user() 

499 

500 with bugs_session(token) as bugs: 

501 bugs.CheckNativeStatus( 

502 bugs_pb2.CheckNativeStatusReq(eas_client_id=str(EAS_CLIENT_ID), platform="ios", user_state="authenticated") 

503 ) 

504 

505 with session_scope() as session: 

506 row = session.execute( 

507 select(NativeClientUser).where(NativeClientUser.eas_client_id == EAS_CLIENT_ID) 

508 ).scalar_one() 

509 assert row.user_id == user.id 

510 

511 

512def test_check_native_status_anonymous_does_not_record_mapping(db): 

513 with bugs_session() as bugs: 

514 bugs.CheckNativeStatus( 

515 bugs_pb2.CheckNativeStatusReq(eas_client_id=str(EAS_CLIENT_ID), platform="ios", user_state="logged_out") 

516 ) 

517 

518 with session_scope() as session: 

519 count = session.execute(select(func.count()).select_from(NativeClientUser)).scalar_one() 

520 assert count == 0 

521 

522 

523def test_check_native_status_append_only_log_of_sightings(db): 

524 # A shared install — same eas-client-id, two users — produces two rows; newest is user_b. 

525 user_a, token_a = generate_user() 

526 user_b, token_b = generate_user() 

527 

528 with bugs_session(token_a) as bugs: 

529 bugs.CheckNativeStatus(bugs_pb2.CheckNativeStatusReq(eas_client_id=str(EAS_CLIENT_ID), platform="ios")) 

530 with bugs_session(token_b) as bugs: 

531 bugs.CheckNativeStatus(bugs_pb2.CheckNativeStatusReq(eas_client_id=str(EAS_CLIENT_ID), platform="ios")) 

532 

533 with session_scope() as session: 

534 rows = ( 

535 session.execute( 

536 select(NativeClientUser) 

537 .where(NativeClientUser.eas_client_id == EAS_CLIENT_ID) 

538 .order_by(NativeClientUser.id) 

539 ) 

540 .scalars() 

541 .all() 

542 ) 

543 assert [r.user_id for r in rows] == [user_a.id, user_b.id] 

544 

545 

546def test_check_native_status_blocks_expired_binary(db): 

547 # A native binary older than the (default 91-day) store window -> required store update, blocking. 

548 embedded_created_at = timestamp_pb2.Timestamp() 

549 embedded_created_at.FromDatetime(datetime.now(UTC) - timedelta(days=120)) 

550 with bugs_session() as bugs: 

551 res = bugs.CheckNativeStatus( 

552 bugs_pb2.CheckNativeStatusReq( 

553 eas_client_id=str(EAS_CLIENT_ID), platform="ios", embedded_created_at=embedded_created_at 

554 ) 

555 ) 

556 

557 assert res.update_info.action == bugs_pb2.NATIVE_UPDATE_ACTION_STORE 

558 assert res.update_info.required is True 

559 assert res.update_info.act_by.ToDatetime(tzinfo=UTC) <= datetime.now(UTC) 

560 

561 

562def _multipart_part_json(body, name): 

563 """Extract and parse the JSON body of a named part from a multipart/mixed body.""" 

564 marker = f'name="{name}"' 

565 start = body.index("\r\n\r\n", body.index(marker)) + 4 

566 end = body.index("\r\n--", start) 

567 return json.loads(body[start:end]) 

568 

569 

570_OTA_CDN_ROOT = "https://cdn.testing.invalid/native/ota" 

571_CDN_CONTENT_TYPE = "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY" 

572 

573 

574class _FakeCDNResponse: 

575 # Echoes the requested URL back as the body so tests can assert which version was fetched and that 

576 # the bytes are served verbatim — standing in for the pre-signed manifest the CDN holds. 

577 def __init__(self, url): 

578 self.headers = {"content-type": _CDN_CONTENT_TYPE} 

579 self.content = url.encode() 

580 

581 def raise_for_status(self): 

582 pass 

583 

584 

585def _patch_cdn(): 

586 return patch("couchers.servicers.bugs.requests.get", side_effect=lambda url, timeout=None: _FakeCDNResponse(url)) 

587 

588 

589def _add_ota_package(*, platform, fingerprint, version, created_at, banned=False): 

590 with session_scope() as session: 

591 creator, _ = generate_user() 

592 package = OTAPackage( 

593 creator_user_id=creator.id, 

594 platform=platform, 

595 fingerprint=fingerprint, 

596 version=version, 

597 manifest_created_at=created_at, 

598 manifest_id=f"id-{version}", 

599 banned_at=created_at if banned else None, 

600 banned_by_user_id=creator.id if banned else None, 

601 banned_reason="test ban" if banned else None, 

602 ) 

603 session.add(package) 

604 session.flush() 

605 

606 

607def test_native_update_manifest_serves_matching_package(db, feature_flags): 

608 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

609 _fetch_signed_manifest.cache_clear() 

610 _add_ota_package( 

611 platform=OTAPlatform.ios, 

612 fingerprint="ios-fingerprint", 

613 version="v1.3.1.aaaa", 

614 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

615 ) 

616 with _patch_cdn(): 

617 with real_bugs_session() as (bugs, metadata_interceptor): 

618 res = bugs.GetNativeUpdateManifest( 

619 httpbody_pb2.HttpBody(), 

620 metadata=( 

621 ("eas-client-id", str(EAS_CLIENT_ID)), 

622 ("expo-platform", "ios"), 

623 ("expo-runtime-version", "ios-fingerprint"), 

624 ), 

625 ) 

626 

627 # the signed bytes are fetched from the CDN under the package's version and served verbatim 

628 assert res.content_type == _CDN_CONTENT_TYPE 

629 assert res.data.decode() == f"{_OTA_CDN_ROOT}/v1.3.1.aaaa/ios/manifest" 

630 # the client requires these response headers or it rejects the manifest 

631 assert metadata_interceptor.latest_headers["expo-protocol-version"] == "1" 

632 assert metadata_interceptor.latest_headers["expo-sfv-version"] == "0" 

633 

634 

635def test_native_update_manifest_resolves_per_platform(db, feature_flags): 

636 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

637 _fetch_signed_manifest.cache_clear() 

638 _add_ota_package( 

639 platform=OTAPlatform.ios, 

640 fingerprint="shared-fingerprint", 

641 version="v1.3.1.ios", 

642 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

643 ) 

644 _add_ota_package( 

645 platform=OTAPlatform.android, 

646 fingerprint="shared-fingerprint", 

647 version="v1.3.1.android", 

648 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

649 ) 

650 with _patch_cdn(): 

651 with real_bugs_session() as (bugs, _metadata_interceptor): 

652 res = bugs.GetNativeUpdateManifest( 

653 httpbody_pb2.HttpBody(), 

654 metadata=( 

655 ("eas-client-id", str(EAS_CLIENT_ID)), 

656 ("expo-platform", "android"), 

657 ("expo-runtime-version", "shared-fingerprint"), 

658 ), 

659 ) 

660 

661 assert res.data.decode() == f"{_OTA_CDN_ROOT}/v1.3.1.android/android/manifest" 

662 

663 

664def test_native_update_manifest_serves_newest_by_created_at(db, feature_flags): 

665 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

666 _fetch_signed_manifest.cache_clear() 

667 # the newer createdAt wins regardless of insertion order 

668 _add_ota_package( 

669 platform=OTAPlatform.ios, 

670 fingerprint="ios-fingerprint", 

671 version="v1.3.2.newer", 

672 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

673 ) 

674 _add_ota_package( 

675 platform=OTAPlatform.ios, 

676 fingerprint="ios-fingerprint", 

677 version="v1.3.1.older", 

678 created_at=datetime(2026, 5, 30, tzinfo=UTC), 

679 ) 

680 with _patch_cdn(): 

681 with real_bugs_session() as (bugs, _metadata_interceptor): 

682 res = bugs.GetNativeUpdateManifest( 

683 httpbody_pb2.HttpBody(), 

684 metadata=( 

685 ("eas-client-id", str(EAS_CLIENT_ID)), 

686 ("expo-platform", "ios"), 

687 ("expo-runtime-version", "ios-fingerprint"), 

688 ), 

689 ) 

690 

691 assert res.data.decode() == f"{_OTA_CDN_ROOT}/v1.3.2.newer/ios/manifest" 

692 

693 

694def test_native_update_manifest_banned_package_excluded(db, feature_flags): 

695 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

696 _fetch_signed_manifest.cache_clear() 

697 _add_ota_package( 

698 platform=OTAPlatform.ios, 

699 fingerprint="ios-fingerprint", 

700 version="v1.3.1.good", 

701 created_at=datetime(2026, 5, 30, tzinfo=UTC), 

702 ) 

703 _add_ota_package( 

704 platform=OTAPlatform.ios, 

705 fingerprint="ios-fingerprint", 

706 version="v1.3.2.bad", 

707 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

708 banned=True, 

709 ) 

710 with _patch_cdn(): 

711 with real_bugs_session() as (bugs, _metadata_interceptor): 

712 res = bugs.GetNativeUpdateManifest( 

713 httpbody_pb2.HttpBody(), 

714 metadata=( 

715 ("eas-client-id", str(EAS_CLIENT_ID)), 

716 ("expo-platform", "ios"), 

717 ("expo-runtime-version", "ios-fingerprint"), 

718 ), 

719 ) 

720 

721 # the newest is banned, so new check-ins get the previous one (a re-stamp would supersede it) 

722 assert res.data.decode() == f"{_OTA_CDN_ROOT}/v1.3.1.good/ios/manifest" 

723 

724 

725def test_native_update_manifest_runtime_mismatch_returns_directive(db): 

726 _add_ota_package( 

727 platform=OTAPlatform.ios, 

728 fingerprint="ios-fingerprint", 

729 version="v1.3.1.aaaa", 

730 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

731 ) 

732 with _patch_cdn() as cdn_get: 

733 with real_bugs_session() as (bugs, _metadata_interceptor): 

734 res = bugs.GetNativeUpdateManifest( 

735 httpbody_pb2.HttpBody(), 

736 metadata=( 

737 ("eas-client-id", str(EAS_CLIENT_ID)), 

738 ("expo-platform", "ios"), 

739 ("expo-runtime-version", "some-other-fingerprint"), 

740 ), 

741 ) 

742 

743 assert _multipart_part_json(res.data.decode(), "directive") == {"type": "noUpdateAvailable"} 

744 # a mismatch must not even fetch — the manifest would be rejected on this build 

745 cdn_get.assert_not_called() 

746 

747 

748def test_native_update_manifest_only_banned_package_returns_directive(db): 

749 _add_ota_package( 

750 platform=OTAPlatform.ios, 

751 fingerprint="ios-fingerprint", 

752 version="v1.3.1.aaaa", 

753 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

754 banned=True, 

755 ) 

756 with real_bugs_session() as (bugs, _metadata_interceptor): 

757 res = bugs.GetNativeUpdateManifest( 

758 httpbody_pb2.HttpBody(), 

759 metadata=( 

760 ("eas-client-id", str(EAS_CLIENT_ID)), 

761 ("expo-platform", "ios"), 

762 ("expo-runtime-version", "ios-fingerprint"), 

763 ), 

764 ) 

765 

766 assert _multipart_part_json(res.data.decode(), "directive") == {"type": "noUpdateAvailable"} 

767 

768 

769def test_native_update_manifest_without_runtime_version_returns_directive(db): 

770 _add_ota_package( 

771 platform=OTAPlatform.ios, 

772 fingerprint="ios-fingerprint", 

773 version="v1.3.1.aaaa", 

774 created_at=datetime(2026, 5, 31, tzinfo=UTC), 

775 ) 

776 with real_bugs_session() as (bugs, metadata_interceptor): 

777 res = bugs.GetNativeUpdateManifest( 

778 httpbody_pb2.HttpBody(), 

779 metadata=( 

780 ("eas-client-id", str(EAS_CLIENT_ID)), 

781 ("expo-platform", "ios"), 

782 ), 

783 ) 

784 

785 body = res.data.decode() 

786 assert _multipart_part_json(body, "directive") == {"type": "noUpdateAvailable"} 

787 assert metadata_interceptor.latest_headers["expo-protocol-version"] == "1" 

788 

789 

790def test_native_update_manifest_no_package_returns_directive(db): 

791 with real_bugs_session() as (bugs, _metadata_interceptor): 

792 res = bugs.GetNativeUpdateManifest( 

793 httpbody_pb2.HttpBody(), 

794 metadata=( 

795 ("eas-client-id", str(EAS_CLIENT_ID)), 

796 ("expo-platform", "ios"), 

797 ("expo-runtime-version", "ios-fingerprint"), 

798 ), 

799 ) 

800 

801 assert _multipart_part_json(res.data.decode(), "directive") == {"type": "noUpdateAvailable"} 

802 

803 

804def _ota_check_req(*, created_at, update_id=""): 

805 ts = timestamp_pb2.Timestamp() 

806 ts.FromDatetime(created_at) 

807 return bugs_pb2.CheckNativeStatusReq( 

808 eas_client_id=str(EAS_CLIENT_ID), 

809 platform="ios", 

810 runtime_version="ios-fingerprint", 

811 launch_source="ota", 

812 update_id=update_id, 

813 created_at=ts, 

814 ) 

815 

816 

817def test_check_native_status_ota_block_with_newer_bundle(db): 

818 _add_ota_package( 

819 platform=OTAPlatform.ios, 

820 fingerprint="ios-fingerprint", 

821 version="v1.3.2.newer", 

822 created_at=datetime.now(UTC) - timedelta(days=1), 

823 ) 

824 with bugs_session() as bugs: 

825 res = bugs.CheckNativeStatus(_ota_check_req(created_at=datetime.now(UTC) - timedelta(days=40))) 

826 

827 assert res.update_info.action == bugs_pb2.NATIVE_UPDATE_ACTION_OTA 

828 assert res.update_info.required is True 

829 assert res.update_info.cause == bugs_pb2.NATIVE_UPDATE_CAUSE_AGE