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

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 

25@pytest.fixture(autouse=True) 

26def _(testconfig): 

27 pass 

28 

29 

30def test_bugs_disabled(): 

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

32 bugs.ReportBug( 

33 bugs_pb2.ReportBugReq( 

34 subject="subject", 

35 description="description", 

36 results="results", 

37 frontend_version="frontend_version", 

38 user_agent="user_agent", 

39 page="page", 

40 ) 

41 ) 

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

43 

44 

45def test_bugs(db): 

46 with bugs_session() as bugs: 

47 

48 def dud_post(url, auth, json): 

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

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

51 

52 expected_body = f""" 

53# subject 

54## Description 

55description 

56 

57## Results 

58results 

59 

60## Diagnostics 

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

62**Frontend version**: `frontend_version` 

63**User Agent**: `user_agent` 

64**Locale**: `en` 

65**Screen resolution**: 1920x1080 

66**Page**: page 

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

68 

69 assert json == { 

70 "title": "subject", 

71 "body": expected_body, 

72 "labels": ["bug tool", "bug: triage needed"], 

73 } 

74 

75 class _PostReturn: 

76 status_code = 201 

77 

78 def json(self): 

79 return {"number": 11} 

80 

81 return _PostReturn() 

82 

83 new_config = config.copy() 

84 new_config.BUG_TOOL_ENABLED = True 

85 

86 with patch("couchers.servicers.bugs.config", new_config): 

87 with patch("couchers.servicers.bugs.requests.post", dud_post): 87 ↛ anywhereline 87 didn't jump anywhere: it always raised an exception.

88 res = bugs.ReportBug( 

89 bugs_pb2.ReportBugReq( 

90 subject="subject", 

91 description="description", 

92 results="results", 

93 frontend_version="frontend_version", 

94 user_agent="user_agent", 

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

96 page="page", 

97 ) 

98 ) 

99 

100 assert res.bug_id == "#11" 

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

102 

103 

104def test_bugs_with_user(db): 

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

106 

107 with bugs_session(token) as bugs: 

108 

109 def dud_post(url, auth, json): 

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

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

112 

113 expected_body = f""" 

114# subject 

115## Description 

116description 

117 

118## Results 

119results 

120 

121## Diagnostics 

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

123**Frontend version**: `frontend_version` 

124**User Agent**: `user_agent` 

125**Locale**: `en` 

126**Screen resolution**: 390x844 

127**Page**: page 

128**User**: [@testing_user](http://localhost:3000/user/testing_user) (1) / `test_sofa_co`""".strip() 

129 

130 assert json == { 

131 "title": "subject", 

132 "body": expected_body, 

133 "labels": ["bug tool", "bug: triage needed"], 

134 } 

135 

136 class _PostReturn: 

137 status_code = 201 

138 

139 def json(self): 

140 return {"number": 11} 

141 

142 return _PostReturn() 

143 

144 new_config = config.copy() 

145 new_config.BUG_TOOL_ENABLED = True 

146 

147 with patch("couchers.servicers.bugs.config", new_config): 

148 with patch("couchers.servicers.bugs.requests.post", dud_post): 148 ↛ anywhereline 148 didn't jump anywhere: it always raised an exception.

149 res = bugs.ReportBug( 

150 bugs_pb2.ReportBugReq( 

151 subject="subject", 

152 description="description", 

153 results="results", 

154 frontend_version="frontend_version", 

155 user_agent="user_agent", 

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

157 page="page", 

158 ) 

159 ) 

160 

161 assert res.bug_id == "#11" 

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

163 

164 

165def test_bugs_fails_on_network_error(db): 

166 with bugs_session() as bugs: 

167 

168 def dud_post(url, auth, json): 

169 class _PostReturn: 

170 status_code = 400 

171 

172 return _PostReturn() 

173 

174 new_config = config.copy() 

175 new_config.BUG_TOOL_ENABLED = True 

176 

177 with patch("couchers.servicers.bugs.config", new_config): 

178 with patch("couchers.servicers.bugs.requests.post", dud_post): 178 ↛ anywhereline 178 didn't jump anywhere: it always raised an exception.

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

180 res = bugs.ReportBug( 

181 bugs_pb2.ReportBugReq( 

182 subject="subject", 

183 description="description", 

184 results="results", 

185 frontend_version="frontend_version", 

186 user_agent="user_agent", 

187 page="page", 

188 ) 

189 ) 

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

191 

192 

193def test_version(): 

194 with bugs_session() as bugs: 

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

196 assert res.version == "testing_version" 

197 

198 

199def test_status(db): 

200 for _ in range(5): 

201 generate_user() 

202 

203 with bugs_session() as bugs: 

204 nonce = random_hex() 

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

206 assert res.nonce == nonce 

207 assert res.version == "testing_version" 

208 assert res.coucher_count == 5 

209 

210 

211def test_GetDescriptors(): 

212 with bugs_session() as bugs: 

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

214 # test we got something roughly binary back 

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

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

217 

218 

219def _get_events(session, event_type=None): 

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

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

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

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

224 

225 

226def test_report_diagnostics_anonymous(db): 

227 with bugs_session() as bugs: 

228 bugs.ReportDiagnostics( 

229 bugs_pb2.ReportDiagnosticsReq( 

230 frontend_version="1.2.3", 

231 infos=[ 

232 bugs_pb2.DiagnosticInfo( 

233 tag="page.viewed", 

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

235 value=1, 

236 ), 

237 bugs_pb2.DiagnosticInfo( 

238 tag="session.started", 

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

240 value=1, 

241 ), 

242 ], 

243 ) 

244 ) 

245 

246 with session_scope() as session: 

247 events = _get_events(session) 

248 assert len(events) == 2 

249 

250 e0 = events[0] 

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

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

253 assert e0.user_id is None 

254 assert e0.source == EventSource.frontend 

255 assert e0.value == 1 

256 assert e0.version == "1.2.3" 

257 

258 e1 = events[1] 

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

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

261 assert e1.source == EventSource.frontend 

262 

263 

264def test_report_diagnostics_authenticated(db): 

265 user, token = generate_user() 

266 

267 with bugs_session(token) as bugs: 

268 bugs.ReportDiagnostics( 

269 bugs_pb2.ReportDiagnosticsReq( 

270 frontend_version="1.2.3", 

271 infos=[ 

272 bugs_pb2.DiagnosticInfo( 

273 tag="page.viewed", 

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

275 value=1, 

276 ), 

277 ], 

278 ) 

279 ) 

280 

281 with session_scope() as session: 

282 events = _get_events(session) 

283 assert len(events) == 1 

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

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

286 

287 

288def test_report_diagnostics_with_value(db): 

289 with bugs_session() as bugs: 

290 bugs.ReportDiagnostics( 

291 bugs_pb2.ReportDiagnosticsReq( 

292 frontend_version="1.2.3", 

293 infos=[ 

294 bugs_pb2.DiagnosticInfo( 

295 tag="search.result_hovered", 

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

297 value=1500.5, 

298 ), 

299 ], 

300 ) 

301 ) 

302 

303 with session_scope() as session: 

304 events = _get_events(session) 

305 assert len(events) == 1 

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

307 

308 

309def test_report_diagnostics_with_occurred(db): 

310 ts = timestamp_pb2.Timestamp() 

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

312 

313 with bugs_session() as bugs: 

314 bugs.ReportDiagnostics( 

315 bugs_pb2.ReportDiagnosticsReq( 

316 frontend_version="1.2.3", 

317 infos=[ 

318 bugs_pb2.DiagnosticInfo( 

319 tag="page.viewed", 

320 properties_json="{}", 

321 value=1, 

322 occurred=ts, 

323 ), 

324 ], 

325 ) 

326 ) 

327 

328 with session_scope() as session: 

329 events = _get_events(session) 

330 assert len(events) == 1 

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

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

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

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

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

336 

337 

338def test_report_diagnostics_invalid_json(db): 

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

340 bugs.ReportDiagnostics( 

341 bugs_pb2.ReportDiagnosticsReq( 

342 frontend_version="1.2.3", 

343 infos=[ 

344 bugs_pb2.DiagnosticInfo( 

345 tag="page.viewed", 

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

347 value=1, 

348 ), 

349 ], 

350 ) 

351 ) 

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

353 

354 

355def test_report_diagnostics_empty_batch(db): 

356 with bugs_session() as bugs: 

357 bugs.ReportDiagnostics( 

358 bugs_pb2.ReportDiagnosticsReq( 

359 frontend_version="1.2.3", 

360 infos=[], 

361 ) 

362 ) 

363 

364 with session_scope() as session: 

365 events = _get_events(session) 

366 assert len(events) == 0 

367 

368 

369def test_report_diagnostics_too_many(db): 

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

371 

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

373 bugs.ReportDiagnostics( 

374 bugs_pb2.ReportDiagnosticsReq( 

375 frontend_version="1.2.3", 

376 infos=infos, 

377 ) 

378 ) 

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

380 

381 

382def test_report_diagnostics_frontend_version(db): 

383 with bugs_session() as bugs: 

384 bugs.ReportDiagnostics( 

385 bugs_pb2.ReportDiagnosticsReq( 

386 frontend_version="abc-def-123", 

387 infos=[ 

388 bugs_pb2.DiagnosticInfo( 

389 tag="page.viewed", 

390 properties_json="{}", 

391 value=1, 

392 ), 

393 ], 

394 ) 

395 ) 

396 

397 with session_scope() as session: 

398 events = _get_events(session) 

399 assert len(events) == 1 

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

401 

402 

403def test_check_native_status_anonymous(db): 

404 with bugs_session() as bugs: 

405 res = bugs.CheckNativeStatus( 

406 bugs_pb2.CheckNativeStatusReq( 

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

408 ) 

409 ) 

410 

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

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

413 assert res.update_info.required is False 

414 

415 

416def test_check_native_status_authenticated(db): 

417 _, token = generate_user() 

418 

419 with bugs_session(token) as bugs: 

420 res = bugs.CheckNativeStatus( 

421 bugs_pb2.CheckNativeStatusReq( 

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

423 ) 

424 ) 

425 

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

427 assert res.update_info.required is False 

428 

429 

430def test_check_native_status_authenticated_records_mapping(db): 

431 user, token = generate_user() 

432 

433 with bugs_session(token) as bugs: 

434 bugs.CheckNativeStatus( 

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

436 ) 

437 

438 with session_scope() as session: 

439 row = session.execute( 

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

441 ).scalar_one() 

442 assert row.user_id == user.id 

443 

444 

445def test_check_native_status_anonymous_does_not_record_mapping(db): 

446 with bugs_session() as bugs: 

447 bugs.CheckNativeStatus( 

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

449 ) 

450 

451 with session_scope() as session: 

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

453 assert count == 0 

454 

455 

456def test_check_native_status_append_only_log_of_sightings(db): 

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

458 user_a, token_a = generate_user() 

459 user_b, token_b = generate_user() 

460 

461 with bugs_session(token_a) as bugs: 

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

463 with bugs_session(token_b) as bugs: 

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

465 

466 with session_scope() as session: 

467 rows = ( 

468 session.execute( 

469 select(NativeClientUser) 

470 .where(NativeClientUser.eas_client_id == EAS_CLIENT_ID) 

471 .order_by(NativeClientUser.id) 

472 ) 

473 .scalars() 

474 .all() 

475 ) 

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

477 

478 

479def test_check_native_status_blocks_expired_binary(db): 

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

481 embedded_created_at = timestamp_pb2.Timestamp() 

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

483 with bugs_session() as bugs: 

484 res = bugs.CheckNativeStatus( 

485 bugs_pb2.CheckNativeStatusReq( 

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

487 ) 

488 ) 

489 

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

491 assert res.update_info.required is True 

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

493 

494 

495def _multipart_part_json(body, name): 

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

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

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

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

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

501 

502 

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

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

505 

506 

507class _FakeCDNResponse: 

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

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

510 def __init__(self, url): 

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

512 self.content = url.encode() 

513 

514 def raise_for_status(self): 

515 pass 

516 

517 

518def _patch_cdn(): 

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

520 

521 

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

523 with session_scope() as session: 

524 creator, _ = generate_user() 

525 package = OTAPackage( 

526 creator_user_id=creator.id, 

527 platform=platform, 

528 fingerprint=fingerprint, 

529 version=version, 

530 manifest_created_at=created_at, 

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

532 banned_at=created_at if banned else None, 

533 banned_by_user_id=creator.id if banned else None, 

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

535 ) 

536 session.add(package) 

537 session.flush() 

538 

539 

540def test_native_update_manifest_serves_matching_package(db, feature_flags): 

541 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

542 _fetch_signed_manifest.cache_clear() 

543 _add_ota_package( 

544 platform=OTAPlatform.ios, 

545 fingerprint="ios-fingerprint", 

546 version="v1.3.1.aaaa", 

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

548 ) 

549 with _patch_cdn(): 

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

551 res = bugs.GetNativeUpdateManifest( 

552 httpbody_pb2.HttpBody(), 

553 metadata=( 

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

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

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

557 ), 

558 ) 

559 

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

561 assert res.content_type == _CDN_CONTENT_TYPE 

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

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

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

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

566 

567 

568def test_native_update_manifest_resolves_per_platform(db, feature_flags): 

569 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

570 _fetch_signed_manifest.cache_clear() 

571 _add_ota_package( 

572 platform=OTAPlatform.ios, 

573 fingerprint="shared-fingerprint", 

574 version="v1.3.1.ios", 

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

576 ) 

577 _add_ota_package( 

578 platform=OTAPlatform.android, 

579 fingerprint="shared-fingerprint", 

580 version="v1.3.1.android", 

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

582 ) 

583 with _patch_cdn(): 

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

585 res = bugs.GetNativeUpdateManifest( 

586 httpbody_pb2.HttpBody(), 

587 metadata=( 

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

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

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

591 ), 

592 ) 

593 

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

595 

596 

597def test_native_update_manifest_serves_newest_by_created_at(db, feature_flags): 

598 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

599 _fetch_signed_manifest.cache_clear() 

600 # the newer createdAt wins regardless of insertion order 

601 _add_ota_package( 

602 platform=OTAPlatform.ios, 

603 fingerprint="ios-fingerprint", 

604 version="v1.3.2.newer", 

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

606 ) 

607 _add_ota_package( 

608 platform=OTAPlatform.ios, 

609 fingerprint="ios-fingerprint", 

610 version="v1.3.1.older", 

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

612 ) 

613 with _patch_cdn(): 

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

615 res = bugs.GetNativeUpdateManifest( 

616 httpbody_pb2.HttpBody(), 

617 metadata=( 

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

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

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

621 ), 

622 ) 

623 

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

625 

626 

627def test_native_update_manifest_banned_package_excluded(db, feature_flags): 

628 feature_flags.set("native_ota_cdn_root", _OTA_CDN_ROOT) 

629 _fetch_signed_manifest.cache_clear() 

630 _add_ota_package( 

631 platform=OTAPlatform.ios, 

632 fingerprint="ios-fingerprint", 

633 version="v1.3.1.good", 

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

635 ) 

636 _add_ota_package( 

637 platform=OTAPlatform.ios, 

638 fingerprint="ios-fingerprint", 

639 version="v1.3.2.bad", 

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

641 banned=True, 

642 ) 

643 with _patch_cdn(): 

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

645 res = bugs.GetNativeUpdateManifest( 

646 httpbody_pb2.HttpBody(), 

647 metadata=( 

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

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

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

651 ), 

652 ) 

653 

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

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

656 

657 

658def test_native_update_manifest_runtime_mismatch_returns_directive(db): 

659 _add_ota_package( 

660 platform=OTAPlatform.ios, 

661 fingerprint="ios-fingerprint", 

662 version="v1.3.1.aaaa", 

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

664 ) 

665 with _patch_cdn() as cdn_get: 

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

667 res = bugs.GetNativeUpdateManifest( 

668 httpbody_pb2.HttpBody(), 

669 metadata=( 

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

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

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

673 ), 

674 ) 

675 

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

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

678 cdn_get.assert_not_called() 

679 

680 

681def test_native_update_manifest_only_banned_package_returns_directive(db): 

682 _add_ota_package( 

683 platform=OTAPlatform.ios, 

684 fingerprint="ios-fingerprint", 

685 version="v1.3.1.aaaa", 

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

687 banned=True, 

688 ) 

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

690 res = bugs.GetNativeUpdateManifest( 

691 httpbody_pb2.HttpBody(), 

692 metadata=( 

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

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

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

696 ), 

697 ) 

698 

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

700 

701 

702def test_native_update_manifest_without_runtime_version_returns_directive(db): 

703 _add_ota_package( 

704 platform=OTAPlatform.ios, 

705 fingerprint="ios-fingerprint", 

706 version="v1.3.1.aaaa", 

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

708 ) 

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

710 res = bugs.GetNativeUpdateManifest( 

711 httpbody_pb2.HttpBody(), 

712 metadata=( 

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

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

715 ), 

716 ) 

717 

718 body = res.data.decode() 

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

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

721 

722 

723def test_native_update_manifest_no_package_returns_directive(db): 

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

725 res = bugs.GetNativeUpdateManifest( 

726 httpbody_pb2.HttpBody(), 

727 metadata=( 

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

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

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

731 ), 

732 ) 

733 

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

735 

736 

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

738 ts = timestamp_pb2.Timestamp() 

739 ts.FromDatetime(created_at) 

740 return bugs_pb2.CheckNativeStatusReq( 

741 eas_client_id=str(EAS_CLIENT_ID), 

742 platform="ios", 

743 runtime_version="ios-fingerprint", 

744 launch_source="ota", 

745 update_id=update_id, 

746 created_at=ts, 

747 ) 

748 

749 

750def test_check_native_status_ota_block_with_newer_bundle(db): 

751 _add_ota_package( 

752 platform=OTAPlatform.ios, 

753 fingerprint="ios-fingerprint", 

754 version="v1.3.2.newer", 

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

756 ) 

757 with bugs_session() as bugs: 

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

759 

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

761 assert res.update_info.required is True 

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

763 

764 

765def test_check_native_status_ota_block_without_target_raises(db): 

766 with bugs_session() as bugs, pytest.raises(Exception, match="no newer bundle to move to"): 

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

768 

769 

770def test_check_native_status_ota_block_only_older_target_raises(db): 

771 _add_ota_package( 

772 platform=OTAPlatform.ios, 

773 fingerprint="ios-fingerprint", 

774 version="v1.3.1.older", 

775 created_at=datetime.now(UTC) - timedelta(days=50), 

776 ) 

777 with bugs_session() as bugs, pytest.raises(Exception, match="no newer bundle to move to"): 

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

779 

780 

781def test_check_native_status_banned_ota_block_with_successor(db): 

782 _add_ota_package( 

783 platform=OTAPlatform.ios, 

784 fingerprint="ios-fingerprint", 

785 version="v1.bad", 

786 created_at=datetime.now(UTC) - timedelta(days=5), 

787 banned=True, 

788 ) 

789 _add_ota_package( 

790 platform=OTAPlatform.ios, 

791 fingerprint="ios-fingerprint", 

792 version="v1.good", 

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

794 ) 

795 with bugs_session() as bugs: 

796 res = bugs.CheckNativeStatus( 

797 _ota_check_req(created_at=datetime.now(UTC) - timedelta(days=5), update_id="id-v1.bad") 

798 ) 

799 

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

801 assert res.update_info.required is True 

802 assert res.update_info.cause == bugs_pb2.NATIVE_UPDATE_CAUSE_BANNED 

803 

804 

805def test_check_native_status_banned_ota_block_no_successor_raises(db): 

806 _add_ota_package( 

807 platform=OTAPlatform.ios,