Coverage for app/backend/src/tests/test_native_updates.py: 96%

1import uuid 

2from datetime import UTC, datetime, timedelta 

3from typing import Any, cast 

4 

5from google.protobuf.timestamp_pb2 import Timestamp 

6 

7from couchers.context import CouchersContext 

8from couchers.native_updates import ( 

9 DEFAULT_OTA_BLOCK_DAYS, 

10 DEFAULT_OTA_WARN_DAYS, 

11 DEFAULT_STORE_BLOCK_DAYS, 

12 DEFAULT_STORE_WARN_DAYS, 

13 NativeClientInfo, 

14 Severity, 

15 UpdateAction, 

16 client_info_from_request, 

17 decide_native_update, 

18) 

19from couchers.proto import bugs_pb2 

20 

21NOW = datetime(2026, 5, 31, 12, 0, tzinfo=UTC) 

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

23 

24 

25class _FakeContext: 

26 def __init__(self, flags: dict[str, Any] | None = None) -> None: 

27 self._flags = flags or {} 

28 

29 def get_boolean_value(self, key: str, default: bool) -> bool: 

30 return bool(self._flags.get(key, default)) 

31 

32 def get_string_value(self, key: str, default: str) -> str: 

33 return str(self._flags.get(key, default)) 

34 

35 def get_integer_value(self, key: str, default: int) -> int: 

36 return int(self._flags.get(key, default)) 

37 

38 def get_float_value(self, key: str, default: float) -> float: 

39 return float(self._flags.get(key, default)) 

40 

41 def get_object_value(self, key: str, default: Any) -> Any: 

42 return self._flags.get(key, default) 

43 

44 

45def _days_ago(days: float) -> datetime: 

46 return NOW - timedelta(days=days) 

47 

48 

49def _decide( 

50 info: NativeClientInfo, 

51 flags: dict[str, Any] | None = None, 

52 *, 

53 banned: bool = False, 

54): 

55 context = cast(CouchersContext, _FakeContext(flags)) 

56 return decide_native_update(context, info, NOW, banned=banned) 

57 

58 

59def _ts(dt: datetime) -> Timestamp: 

60 out = Timestamp() 

61 out.FromDatetime(dt) 

62 return out 

63 

64 

65def test_client_info_from_full_proto(): 

66 req = bugs_pb2.CheckNativeStatusReq( 

67 eas_client_id=str(EAS_CLIENT_ID), 

68 platform="ios", 

69 runtime_version="ios-fingerprint", 

70 update_id="abc-123", 

71 launch_source="ota", 

72 created_at=_ts(datetime(2026, 5, 1, tzinfo=UTC)), 

73 embedded_created_at=_ts(datetime(2026, 1, 1, tzinfo=UTC)), 

74 ) 

75 info = client_info_from_request(req) 

76 assert info.platform == "ios" 

77 assert info.runtime_version == "ios-fingerprint" 

78 assert info.update_id == "abc-123" 

79 assert info.is_ota_launch is True 

80 assert info.bundle_created_at == datetime(2026, 5, 1, tzinfo=UTC) 

81 assert info.binary_created_at == datetime(2026, 1, 1, tzinfo=UTC) 

82 

83 

84def test_client_info_embedded_launch_source(): 

85 req = bugs_pb2.CheckNativeStatusReq( 

86 eas_client_id=str(EAS_CLIENT_ID), 

87 platform="android", 

88 launch_source="embedded", 

89 is_embedded_launch=True, 

90 update_id="none", 

91 ) 

92 info = client_info_from_request(req) 

93 assert info.is_ota_launch is False 

94 # "none" is the placeholder for absent updateId on the client. 

95 assert info.update_id is None 

96 

97 

98def test_client_info_defaults_when_request_empty(): 

99 info = client_info_from_request(bugs_pb2.CheckNativeStatusReq(eas_client_id=str(EAS_CLIENT_ID))) 

100 assert info == NativeClientInfo(eas_client_id=EAS_CLIENT_ID) 

101 

102 

103def test_no_timestamps_means_no_update(): 

104 decision = _decide(NativeClientInfo(eas_client_id=EAS_CLIENT_ID, platform="ios")) 

105 assert decision.action == UpdateAction.none 

106 assert decision.severity == Severity.none 

107 

108 

109def test_fresh_binary_no_update(): 

110 info = NativeClientInfo(eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(10)) 

111 assert _decide(info).severity == Severity.none 

112 

113 

114def test_binary_between_warn_and_block_is_store_warn(): 

115 info = NativeClientInfo( 

116 eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(DEFAULT_STORE_WARN_DAYS + 1) 

117 ) 

118 decision = _decide(info) 

119 assert decision.action == UpdateAction.store 

120 assert decision.severity == Severity.warn 

121 assert decision.act_by is not None and decision.act_by > NOW 

122 

123 

124def test_binary_past_block_is_store_block(): 

125 info = NativeClientInfo( 

126 eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 5) 

127 ) 

128 decision = _decide(info) 

129 assert decision.action == UpdateAction.store 

130 assert decision.severity == Severity.block 

131 assert decision.act_by is not None and decision.act_by <= NOW 

132 

133 

134def test_ota_between_warn_and_block_is_ota_warn(): 

135 info = NativeClientInfo( 

136 eas_client_id=EAS_CLIENT_ID, 

137 platform="ios", 

138 is_ota_launch=True, 

139 binary_created_at=_days_ago(5), 

140 bundle_created_at=_days_ago(DEFAULT_OTA_WARN_DAYS + 1), 

141 ) 

142 decision = _decide(info) 

143 assert decision.action == UpdateAction.ota 

144 assert decision.severity == Severity.warn 

145 assert decision.act_by is not None and decision.act_by > NOW 

146 

147 

148def test_ota_past_block_is_ota_block(): 

149 info = NativeClientInfo( 

150 eas_client_id=EAS_CLIENT_ID, 

151 platform="ios", 

152 is_ota_launch=True, 

153 binary_created_at=_days_ago(5), 

154 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 2), 

155 ) 

156 decision = _decide(info) 

157 assert decision.action == UpdateAction.ota 

158 assert decision.severity == Severity.block 

159 

160 

161def test_ota_clock_ignored_when_not_running_ota(): 

162 info = NativeClientInfo( 

163 eas_client_id=EAS_CLIENT_ID, 

164 platform="ios", 

165 is_ota_launch=False, 

166 binary_created_at=_days_ago(5), 

167 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 100), 

168 ) 

169 assert _decide(info).severity == Severity.none 

170 

171 

172def test_binary_warn_but_ota_block_resolves_to_ota_block(): 

173 info = NativeClientInfo( 

174 eas_client_id=EAS_CLIENT_ID, 

175 platform="ios", 

176 is_ota_launch=True, 

177 binary_created_at=_days_ago(DEFAULT_STORE_WARN_DAYS + 1), 

178 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 2), 

179 ) 

180 decision = _decide(info) 

181 assert decision.action == UpdateAction.ota 

182 assert decision.severity == Severity.block 

183 

184 

185def test_store_precedence_when_severities_tie(): 

186 info = NativeClientInfo( 

187 eas_client_id=EAS_CLIENT_ID, 

188 platform="ios", 

189 is_ota_launch=True, 

190 binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 1), 

191 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 1), 

192 ) 

193 assert _decide(info).action == UpdateAction.store 

194 

195 

196def test_binary_block_beats_ota_warn(): 

197 info = NativeClientInfo( 

198 eas_client_id=EAS_CLIENT_ID, 

199 platform="ios", 

200 is_ota_launch=True, 

201 binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 1), 

202 bundle_created_at=_days_ago(DEFAULT_OTA_WARN_DAYS + 1), 

203 ) 

204 assert _decide(info).action == UpdateAction.store 

205 

206 

207def test_warn_days_drive_warn_threshold(): 

208 info = NativeClientInfo(eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(8)) 

209 decision = _decide(info, flags={"native_store_warn_days": 7, "native_store_block_days": 30}) 

210 assert decision.severity == Severity.warn 

211 

212 

213def test_block_days_drive_block_threshold(): 

214 info = NativeClientInfo(eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(10)) 

215 decision = _decide(info, flags={"native_store_warn_days": 5, "native_store_block_days": 7}) 

216 assert decision.action == UpdateAction.store 

217 assert decision.severity == Severity.block 

218 

219 

220def test_zero_block_disables_clock(): 

221 info = NativeClientInfo(eas_client_id=EAS_CLIENT_ID, platform="ios", binary_created_at=_days_ago(1000)) 

222 assert _decide(info, flags={"native_store_block_days": 0}).severity == Severity.none 

223 

224 

225def test_banned_bundle_on_ota_launch_forces_block(): 

226 info = NativeClientInfo( 

227 eas_client_id=EAS_CLIENT_ID, 

228 platform="ios", 

229 is_ota_launch=True, 

230 update_id="abc", 

231 binary_created_at=_days_ago(5), 

232 bundle_created_at=_days_ago(1), 

233 ) 

234 decision = _decide(info, banned=True) 

235 assert decision.action == UpdateAction.ota 

236 assert decision.severity == Severity.block 

237 # Unset deadline = block-now per the proto contract; avoids clock-skew flipping into warn. 

238 assert decision.act_by is None 

239 

240 

241def test_banned_ignored_when_not_an_ota_launch(): 

242 info = NativeClientInfo( 

243 eas_client_id=EAS_CLIENT_ID, platform="ios", is_ota_launch=False, binary_created_at=_days_ago(5) 

244 ) 

245 assert _decide(info, banned=True).severity == Severity.none