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

1from datetime import UTC, datetime, timedelta 

2from typing import Any, cast 

3 

4from google.protobuf.timestamp_pb2 import Timestamp 

5 

6from couchers.context import CouchersContext 

7from couchers.native_updates import ( 

8 DEFAULT_OTA_BLOCK_DAYS, 

9 DEFAULT_OTA_WARN_DAYS, 

10 DEFAULT_STORE_BLOCK_DAYS, 

11 DEFAULT_STORE_WARN_DAYS, 

12 NativeClientInfo, 

13 Severity, 

14 UpdateAction, 

15 client_info_from_request, 

16 decide_native_update, 

17) 

18from couchers.proto import bugs_pb2 

19 

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

21 

22 

23class _FakeContext: 

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

25 self._flags = flags or {} 

26 

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

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

29 

30 

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

32 return NOW - timedelta(days=days) 

33 

34 

35def _decide( 

36 info: NativeClientInfo, 

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

38 *, 

39 banned: bool = False, 

40): 

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

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

43 

44 

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

46 out = Timestamp() 

47 out.FromDatetime(dt) 

48 return out 

49 

50 

51def test_client_info_from_full_proto(): 

52 req = bugs_pb2.CheckNativeStatusReq( 

53 platform="ios", 

54 runtime_version="ios-fingerprint", 

55 update_id="abc-123", 

56 launch_source="ota", 

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

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

59 ) 

60 info = client_info_from_request(req) 

61 assert info.platform == "ios" 

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

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

64 assert info.is_ota_launch is True 

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

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

67 

68 

69def test_client_info_embedded_launch_source(): 

70 req = bugs_pb2.CheckNativeStatusReq( 

71 platform="android", launch_source="embedded", is_embedded_launch=True, update_id="none" 

72 ) 

73 info = client_info_from_request(req) 

74 assert info.is_ota_launch is False 

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

76 assert info.update_id is None 

77 

78 

79def test_client_info_defaults_when_request_empty(): 

80 info = client_info_from_request(bugs_pb2.CheckNativeStatusReq()) 

81 assert info == NativeClientInfo() 

82 

83 

84def test_no_timestamps_means_no_update(): 

85 decision = _decide(NativeClientInfo(platform="ios")) 

86 assert decision.action == UpdateAction.none 

87 assert decision.severity == Severity.none 

88 

89 

90def test_fresh_binary_no_update(): 

91 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(10)) 

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

93 

94 

95def test_binary_between_warn_and_block_is_store_warn(): 

96 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(DEFAULT_STORE_WARN_DAYS + 1)) 

97 decision = _decide(info) 

98 assert decision.action == UpdateAction.store 

99 assert decision.severity == Severity.warn 

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

101 

102 

103def test_binary_past_block_is_store_block(): 

104 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 5)) 

105 decision = _decide(info) 

106 assert decision.action == UpdateAction.store 

107 assert decision.severity == Severity.block 

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

109 

110 

111def test_ota_between_warn_and_block_is_ota_warn(): 

112 info = NativeClientInfo( 

113 platform="ios", 

114 is_ota_launch=True, 

115 binary_created_at=_days_ago(5), 

116 bundle_created_at=_days_ago(DEFAULT_OTA_WARN_DAYS + 1), 

117 ) 

118 decision = _decide(info) 

119 assert decision.action == UpdateAction.ota 

120 assert decision.severity == Severity.warn 

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

122 

123 

124def test_ota_past_block_is_ota_block(): 

125 info = NativeClientInfo( 

126 platform="ios", 

127 is_ota_launch=True, 

128 binary_created_at=_days_ago(5), 

129 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 2), 

130 ) 

131 decision = _decide(info) 

132 assert decision.action == UpdateAction.ota 

133 assert decision.severity == Severity.block 

134 

135 

136def test_ota_clock_ignored_when_not_running_ota(): 

137 info = NativeClientInfo( 

138 platform="ios", 

139 is_ota_launch=False, 

140 binary_created_at=_days_ago(5), 

141 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 100), 

142 ) 

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

144 

145 

146def test_binary_warn_but_ota_block_resolves_to_ota_block(): 

147 info = NativeClientInfo( 

148 platform="ios", 

149 is_ota_launch=True, 

150 binary_created_at=_days_ago(DEFAULT_STORE_WARN_DAYS + 1), 

151 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 2), 

152 ) 

153 decision = _decide(info) 

154 assert decision.action == UpdateAction.ota 

155 assert decision.severity == Severity.block 

156 

157 

158def test_store_precedence_when_severities_tie(): 

159 info = NativeClientInfo( 

160 platform="ios", 

161 is_ota_launch=True, 

162 binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 1), 

163 bundle_created_at=_days_ago(DEFAULT_OTA_BLOCK_DAYS + 1), 

164 ) 

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

166 

167 

168def test_binary_block_beats_ota_warn(): 

169 info = NativeClientInfo( 

170 platform="ios", 

171 is_ota_launch=True, 

172 binary_created_at=_days_ago(DEFAULT_STORE_BLOCK_DAYS + 1), 

173 bundle_created_at=_days_ago(DEFAULT_OTA_WARN_DAYS + 1), 

174 ) 

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

176 

177 

178def test_warn_days_drive_warn_threshold(): 

179 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(8)) 

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

181 assert decision.severity == Severity.warn 

182 

183 

184def test_block_days_drive_block_threshold(): 

185 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(10)) 

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

187 assert decision.action == UpdateAction.store 

188 assert decision.severity == Severity.block 

189 

190 

191def test_zero_block_disables_clock(): 

192 info = NativeClientInfo(platform="ios", binary_created_at=_days_ago(1000)) 

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

194 

195 

196def test_banned_bundle_on_ota_launch_forces_block(): 

197 info = NativeClientInfo( 

198 platform="ios", 

199 is_ota_launch=True, 

200 update_id="abc", 

201 binary_created_at=_days_ago(5), 

202 bundle_created_at=_days_ago(1), 

203 ) 

204 decision = _decide(info, banned=True) 

205 assert decision.action == UpdateAction.ota 

206 assert decision.severity == Severity.block 

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

208 assert decision.act_by is None 

209 

210 

211def test_banned_ignored_when_not_an_ota_launch(): 

212 info = NativeClientInfo(platform="ios", is_ota_launch=False, binary_created_at=_days_ago(5)) 

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