Coverage for tests/simulator/test_simulator_multirequests.py: 100%

1import sys 

2import os 

3 

4# Add current path 

5sys.path.append(os.getcwd()) 

6 

7from tests.test_utils import assert_equals_approx 

8from tests.test_utils import temp_sys_path 

9 

10with temp_sys_path("simulator", "streamwise"): 

11 from multirequests import QPM_LIST 

12 from multirequests import get_replicas 

13 from multirequests import get_costs 

14 from multirequests import get_total_costs 

15 from multirequests import required_replicas 

16 from multirequests import aggregate_time_per_request_by_quality 

17 from multirequests import TIME_PER_REQ 

18 from multirequests import INIT_REPLICAS 

19 from multirequests import INIT_REPLICAS_BASELINE 

20 from multirequests import QUALITY_PORTIONS 

21 from multirequests import TIME_PER_REQ_ADAPTIVE 

22 from multirequests import get_time_per_request_baseline 

23 

24 from data_loading import load_latency_data 

25 from workflows import PODCAST_WORKFLOW 

26 

27 from constants import GPU_SPOT_COST 

28 from sim_types import GPUType 

29 from sim_types import Model 

30 

31 

32def test_multirequests() -> None: 

33 video_minutes = 10 

34 video_seconds = video_minutes * 60 

35 

36 replicas = get_replicas(video_seconds=video_seconds) 

37 assert len(replicas) 

38 

39 costs = get_costs( 

40 replicas=replicas, 

41 gpu_costs=GPU_SPOT_COST, 

42 ) 

43 

44 total_costs = get_total_costs(costs) 

45 

46 assert len(total_costs) == len(QPM_LIST) == 9 

47 

48 

49def test_required_replicas_scenes_partition() -> None: 

50 """Test required_replicas with scenes partition.""" 

51 result = required_replicas( 

52 name="test", 

53 video_seconds=600, 

54 ttff=10.0, 

55 per_sec=0.5, 

56 partition="scenes", 

57 req_per_min=1.0, 

58 ) 

59 # ttff_total = 10.0, per_sec_total = 600 * 0.5 = 300 

60 # total_time_per_request = 310, total_time_per_minute = 310 * 1.0 = 310 

61 # expected = 310 / 60 

62 assert_equals_approx(result, 310 / 60) 

63 

64 

65def test_required_replicas_frames_partition_vae() -> None: 

66 """Test required_replicas with frames partition for VAE.""" 

67 result = required_replicas( 

68 name="hf_vae", 

69 video_seconds=600, 

70 ttff=5.0, 

71 per_sec=0.1, 

72 partition="frames", 

73 req_per_min=2.0, 

74 ) 

75 # ttff_total = 5.0, per_sec_total = 600 * 0.1 = 60 

76 # total_time_per_request = 65, total_time_per_minute = 65 * 2.0 = 130 

77 # expected = 130 / 60 

78 assert_equals_approx(result, 130 / 60) 

79 

80 

81def test_required_replicas_frames_partition_upscaler() -> None: 

82 """Test required_replicas with frames partition for upscaler.""" 

83 result = required_replicas( 

84 name="upscaler", 

85 video_seconds=600, 

86 ttff=8.0, 

87 per_sec=0.2, 

88 partition="frames", 

89 req_per_min=0.5, 

90 ) 

91 # ttff_total = 8.0, per_sec_total = 600 * 0.2 = 120 

92 # total_time_per_request = 128, total_time_per_minute = 128 * 0.5 = 64 

93 # expected = 64 / 60 

94 assert_equals_approx(result, 64 / 60) 

95 

96 

97def test_required_replicas_subscenes_partition() -> None: 

98 """Test required_replicas with subscenes partition.""" 

99 result = required_replicas( 

100 name="test", 

101 video_seconds=300, 

102 ttff=15.0, 

103 per_sec=0.3, 

104 partition="subscenes", 

105 req_per_min=3.0, 

106 ) 

107 # ttff_total = 15.0, per_sec_total = 300 * 0.3 = 90 

108 # total_time_per_request = 105, total_time_per_minute = 105 * 3.0 = 315 

109 # expected = 315 / 60 

110 assert_equals_approx(result, 315 / 60) 

111 

112 

113def test_required_replicas_other_partition() -> None: 

114 """Test required_replicas with other/default partition.""" 

115 result = required_replicas( 

116 name="test", 

117 video_seconds=120, 

118 ttff=2.0, 

119 per_sec=0.1, 

120 partition="unknown", 

121 req_per_min=5.0, 

122 ) 

123 # ttff_total = 2.0, per_sec_total = 120 * 0.1 = 12 

124 # total_time_per_request = 14, total_time_per_minute = 14 * 5.0 = 70 

125 # expected = 70 / 60 

126 assert_equals_approx(result, 70 / 60) 

127 

128 

129def test_get_replicas_with_custom_parameters() -> None: 

130 """Test get_replicas with custom video length and QPM list.""" 

131 video_seconds = 300 

132 custom_qpm = [1, 5, 10] 

133 

134 replicas = get_replicas( 

135 video_seconds=video_seconds, 

136 requests_per_minute=0.5, 

137 time_per_req=TIME_PER_REQ, 

138 init_replicas=INIT_REPLICAS, 

139 qpms=custom_qpm, 

140 ) 

141 

142 # Check structure 

143 assert GPUType.A100 in replicas 

144 assert Model.FLUX in replicas[GPUType.A100] 

145 

146 flux_replicas = replicas[GPUType.A100][Model.FLUX] 

147 

148 # Should have one entry per QPM 

149 assert len(flux_replicas) == len(custom_qpm) 

150 

151 # Replicas should increase with higher QPM 

152 assert flux_replicas[0] <= flux_replicas[1] <= flux_replicas[2] 

153 

154 

155def test_get_replicas_returns_all_models() -> None: 

156 """Test that get_replicas().""" 

157 replicas = get_replicas(video_seconds=10 * 60) 

158 

159 assert len(replicas) > 0 

160 for gpu_type in replicas.keys(): 

161 assert gpu_type in replicas 

162 assert len(replicas[gpu_type]) > 0 

163 for model in replicas[gpu_type].keys(): 

164 assert model in replicas[gpu_type] 

165 replica_list = replicas[gpu_type][model] 

166 for replica_count in replica_list: 

167 assert replica_count > 0 

168 # Each successive QPM should have >= replicas than previous 

169 for i in range(len(replica_list) - 1): 

170 assert replica_list[i] <= replica_list[i + 1] 

171 

172 

173def test_get_costs_structure() -> None: 

174 """Test that get_costs returns correct structure.""" 

175 replicas = get_replicas(video_seconds=600) 

176 costs = get_costs(replicas=replicas, gpu_costs=GPU_SPOT_COST) 

177 

178 # Should match replicas structure 

179 assert set(costs.keys()) == set(replicas.keys()) 

180 

181 for gpu_type in costs: 

182 assert set(costs[gpu_type].keys()) == set(replicas[gpu_type].keys()) 

183 

184 for model in costs[gpu_type]: 

185 # Cost lists should have same length as replica lists 

186 assert len(costs[gpu_type][model]) == len(replicas[gpu_type][model]) 

187 

188 

189def test_costs() -> None: 

190 """Test that get_costs correctly calculates costs.""" 

191 replicas = get_replicas(video_seconds=600) 

192 costs = get_costs( 

193 replicas=replicas, 

194 gpu_costs=GPU_SPOT_COST) 

195 

196 # Verify cost calculations 

197 for gpu_type in replicas: 

198 for model in replicas[gpu_type]: 

199 for i, replica_count in enumerate(replicas[gpu_type][model]): 

200 expected_cost = replica_count * GPU_SPOT_COST[gpu_type] 

201 assert abs(costs[gpu_type][model][i] - expected_cost) < 0.01 

202 

203 for gpu_type in costs: 

204 for model in costs[gpu_type]: 

205 for cost in costs[gpu_type][model]: 

206 assert cost > 0 

207 

208 # Total costs 

209 total_costs = get_total_costs(costs) 

210 assert len(total_costs) == len(QPM_LIST) 

211 

212 for i in range(len(total_costs)): 

213 expected_total = 0 

214 for gpu_type in costs: 

215 for model in costs[gpu_type]: 

216 expected_total += costs[gpu_type][model][i] 

217 assert_equals_approx(total_costs[i], expected_total) 

218 

219 # Costs should generally increase with QPM 

220 for i in range(len(total_costs) - 1): 

221 assert total_costs[i] <= total_costs[i + 1] 

222 

223 

224def test_get_total_costs_custom() -> None: 

225 """Test get_total_costs with custom QPM list.""" 

226 custom_qpm = [1, 2, 5] 

227 replicas = get_replicas(video_seconds=600, qpms=custom_qpm) 

228 costs = get_costs(replicas=replicas, gpu_costs=GPU_SPOT_COST) 

229 total_costs = get_total_costs(costs, qpms=custom_qpm) 

230 

231 assert len(total_costs) == len(custom_qpm) 

232 

233 

234def test_get_replicas_short_video() -> None: 

235 """Test get_replicas with very short video.""" 

236 replicas = get_replicas(video_seconds=10) # 10 seconds 

237 

238 assert len(replicas) > 0 

239 

240 for gpu_type in replicas: 

241 for model in replicas[gpu_type]: 

242 assert len(replicas[gpu_type][model]) > 0 

243 

244 

245def test_get_replicas_long_video() -> None: 

246 """Test get_replicas() with very long video.""" 

247 replicas = get_replicas(video_seconds=60 * 60) # 1 hour 

248 

249 assert len(replicas) > 0 

250 

251 # Replicas for long video should be higher than short video 

252 short_replicas = get_replicas(video_seconds=60) 

253 

254 for gpu_type in replicas: 

255 if gpu_type in short_replicas: 

256 for model in replicas[gpu_type]: 

257 if model in short_replicas[gpu_type]: 

258 # At least for QPM=1, long video should need more replicas 

259 assert replicas[gpu_type][model][0] >= short_replicas[gpu_type][model][0] 

260 

261 

262def test_aggregate_time_per_request_by_quality() -> None: 

263 """Test aggregate_time_per_request_by_quality().""" 

264 time_req_adaptive_agg = aggregate_time_per_request_by_quality( 

265 time_per_req=TIME_PER_REQ_ADAPTIVE, 

266 quality_portions=QUALITY_PORTIONS, 

267 ) 

268 assert GPUType.A100 in time_req_adaptive_agg 

269 assert GPUType.H100 in time_req_adaptive_agg 

270 

271 assert Model.GEMMA in time_req_adaptive_agg[GPUType.A100] 

272 assert Model.FLUX in time_req_adaptive_agg[GPUType.A100] 

273 assert Model.UPSCALER in time_req_adaptive_agg[GPUType.A100] 

274 assert Model.UPSCALER in time_req_adaptive_agg[GPUType.H100] 

275 

276 assert_equals_approx(time_req_adaptive_agg[GPUType.A100][Model.OTHERS], 25.80) 

277 assert_equals_approx(time_req_adaptive_agg[GPUType.H100][Model.UPSCALER], 48.12) 

278 

279 

280def test_get_time_per_request_baseline() -> None: 

281 latency_data = load_latency_data("simulator/data/") 

282 

283 time_per_req_baseline = get_time_per_request_baseline( 

284 workflow_config=PODCAST_WORKFLOW, 

285 latency_data=latency_data, 

286 init_replicas=INIT_REPLICAS_BASELINE, 

287 ) 

288 assert GPUType.A100 in time_per_req_baseline 

289 assert Model.GEMMA in time_per_req_baseline[GPUType.A100] 

290 assert_equals_approx(time_per_req_baseline[GPUType.A100][Model.FLUX], 9.75) 

291 assert_equals_approx(time_per_req_baseline[GPUType.A100][Model.FT], 24618.20)