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

1""" 

2Unit tests for simulator/workflows.py 

3 

4Tests for build_workflow_config, _video_gen_work, and the pre-built 

5workflow configs (PODCAST_WORKFLOW, SHORTS_WORKFLOW, etc.). 

6""" 

7 

8import math 

9import sys 

10import os 

11 

12import pytest 

13 

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

15 

16from tests.test_utils import temp_sys_path 

17 

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

19 from sim_types import WorkflowConfig, Model, QualityLevel, GPUType 

20 from constants import ( 

21 FPS, 

22 FRAMES_OPTIONS, 

23 FRAMES_PER_STEP_IDX, 

24 NUM_STEPS, 

25 SECONDS_IN_HOUR, 

26 SECONDS_IN_MINUTE, 

27 TOTAL_INPUT_TOKENS, 

28 ) 

29 from data_loading import load_latency_data 

30 from auto_model_allocator import AutoModelAllocator 

31 from model_provisioner.policies import STREAMWISE_POLICY, NAIVE_POLICY 

32 from workflows import ( 

33 MAX_FT_FRAMES, 

34 SUBSCENE_SECONDS, 

35 SUBSCENES_PER_SCENE, 

36 build_workflow_config, 

37 _get_num_subscenes, 

38 _get_num_scenes, 

39 _video_gen_work, 

40 WORKFLOWS, 

41 PODCAST_WORKFLOW, 

42 SHORTS_WORKFLOW, 

43 MOVIE_WORKFLOW, 

44 ANIMATED_STORY_WORKFLOW, 

45 LECTURE_WORKFLOW, 

46 DUBBING_WORKFLOW, 

47 EDITING_WORKFLOW, 

48 VIDEO_CHAT_WORKFLOW, 

49 ) 

50 

51 

52class TestModuleConstants: 

53 """Module-level constant tests.""" 

54 def test_max_ft_frames(self) -> None: 

55 assert MAX_FT_FRAMES == 1 + 80 

56 

57 def test_subscene_seconds(self) -> None: 

58 assert SUBSCENE_SECONDS == pytest.approx(MAX_FT_FRAMES / FPS[Model.FT]) 

59 

60 def test_subscenes_per_scene(self) -> None: 

61 assert SUBSCENES_PER_SCENE == 4 

62 

63 

64class TestHelperFunctions: 

65 """Helper function tests.""" 

66 def test_num_subscenes(self) -> None: 

67 secs = 2 * 60 # 2 minutes 

68 assert _get_num_subscenes(secs) == math.ceil(secs / SUBSCENE_SECONDS) 

69 

70 def test_num_scenes(self) -> None: 

71 secs = 2 * 60 # 2 minutes 

72 expected_subscenes = math.ceil(secs / SUBSCENE_SECONDS) 

73 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

74 assert _get_num_scenes(secs) == expected_scenes 

75 

76 

77class TestBuildWorkflowConfig: 

78 """Build workflow config tests.""" 

79 

80 def test_returns_workflow_config(self) -> None: 

81 cfg = build_workflow_config( 

82 total_video_seconds=60, 

83 input_tokens=1000, 

84 model_work={}, 

85 ) 

86 assert isinstance(cfg, WorkflowConfig) 

87 

88 def test_basic_fields(self) -> None: 

89 cfg = build_workflow_config( 

90 total_video_seconds=60, 

91 input_tokens=1000, 

92 model_work={}, 

93 ) 

94 num_ss = _get_num_subscenes(60) 

95 assert cfg.total_video_seconds == 60 

96 assert cfg.total_scenes == _get_num_scenes(60) 

97 assert cfg.total_subscenes == num_ss 

98 assert cfg.total_input_tokens == 1000 

99 assert cfg.target_resolution == QualityLevel.HIGH 

100 assert cfg.hf_frames == FRAMES_OPTIONS[Model.HF] 

101 assert cfg.ft_frames == FRAMES_OPTIONS[Model.FT] 

102 assert cfg.frames_per_step_idx == FRAMES_PER_STEP_IDX 

103 

104 def test_total_frames(self) -> None: 

105 cfg = build_workflow_config( 

106 total_video_seconds=60, 

107 input_tokens=0, 

108 model_work={}, 

109 ) 

110 assert cfg.total_frames[Model.HF] == 60 * FPS[Model.HF] 

111 assert cfg.total_frames[Model.FT] == 60 * FPS[Model.FT] 

112 

113 def test_per_subscene_frames(self) -> None: 

114 cfg = build_workflow_config( 

115 total_video_seconds=60, 

116 input_tokens=0, 

117 model_work={}, 

118 ) 

119 video_seconds = 60 

120 num_ss = _get_num_subscenes(video_seconds) 

121 assert cfg.per_subscene_frames[Model.HF] == math.ceil(video_seconds * FPS[Model.HF] / num_ss) 

122 assert cfg.per_subscene_frames[Model.FT] == math.ceil(video_seconds * FPS[Model.FT] / num_ss) 

123 

124 def test_num_steps_default(self) -> None: 

125 cfg = build_workflow_config( 

126 total_video_seconds=60, 

127 input_tokens=0, 

128 model_work={}, 

129 ) 

130 assert cfg.num_steps[Model.FLUX] == NUM_STEPS[Model.FLUX] 

131 assert cfg.num_steps[Model.HF] == NUM_STEPS[Model.HF] 

132 assert cfg.num_steps[Model.FT] == NUM_STEPS[Model.FT] 

133 

134 def test_num_scenes_override(self) -> None: 

135 cfg = build_workflow_config( 

136 total_video_seconds=60, 

137 input_tokens=0, 

138 model_work={}, 

139 num_scenes_override=42, 

140 ) 

141 assert cfg.total_scenes == 42 

142 

143 def test_num_steps_override(self) -> None: 

144 cfg = build_workflow_config( 

145 total_video_seconds=60, 

146 input_tokens=0, 

147 model_work={}, 

148 num_steps_override={Model.HF: 99}, 

149 ) 

150 assert cfg.num_steps[Model.HF] == 99 

151 # Other steps unchanged 

152 assert cfg.num_steps[Model.FLUX] == NUM_STEPS[Model.FLUX] 

153 assert cfg.num_steps[Model.FT] == NUM_STEPS[Model.FT] 

154 

155 

156class TestPodcastWorkflow: 

157 """Podcast workflow config tests.""" 

158 

159 PODCAST_TOTAL_SECONDS = int(10 * SECONDS_IN_MINUTE) # 600 s 

160 

161 def test_total_video_seconds(self) -> None: 

162 assert PODCAST_WORKFLOW.total_video_seconds == self.PODCAST_TOTAL_SECONDS 

163 

164 def test_input_tokens(self) -> None: 

165 assert PODCAST_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

166 

167 def test_num_subscenes(self) -> None: 

168 expected = math.ceil(self.PODCAST_TOTAL_SECONDS / SUBSCENE_SECONDS) 

169 assert PODCAST_WORKFLOW.total_subscenes == expected 

170 

171 def test_num_scenes(self) -> None: 

172 expected_subscenes = math.ceil(self.PODCAST_TOTAL_SECONDS / SUBSCENE_SECONDS) 

173 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

174 assert PODCAST_WORKFLOW.total_scenes == expected_scenes 

175 

176 def test_model_work_keys(self) -> None: 

177 expected_models = { 

178 Model.GEMMA, Model.FLUX, 

179 Model.HF, Model.HF_VAE, 

180 Model.FT, Model.FT_VAE, 

181 Model.UPSCALER, Model.OTHERS, 

182 } 

183 assert set(PODCAST_WORKFLOW.model_work.keys()) == expected_models 

184 

185 def test_model_work_singleton_values(self) -> None: 

186 assert PODCAST_WORKFLOW.model_work[Model.GEMMA] == 1 

187 assert PODCAST_WORKFLOW.model_work[Model.FLUX] == 1 

188 assert PODCAST_WORKFLOW.model_work[Model.OTHERS] == 1 

189 

190 def test_model_work_subscene_values(self) -> None: 

191 assert PODCAST_WORKFLOW.model_work[Model.HF] == PODCAST_WORKFLOW.total_subscenes 

192 assert PODCAST_WORKFLOW.model_work[Model.FT] == PODCAST_WORKFLOW.total_subscenes 

193 

194 def test_model_work_frame_values(self) -> None: 

195 assert PODCAST_WORKFLOW.model_work[Model.HF_VAE] == self.PODCAST_TOTAL_SECONDS * FPS[Model.HF] 

196 assert PODCAST_WORKFLOW.model_work[Model.UPSCALER] == self.PODCAST_TOTAL_SECONDS * FPS[Model.FT] 

197 

198 def test_config_models(self) -> None: 

199 assert set(PODCAST_WORKFLOW.models) == { 

200 Model.GEMMA, Model.FLUX, 

201 Model.HF, Model.HF_VAE, 

202 Model.FT, Model.FT_VAE, 

203 Model.UPSCALER, Model.OTHERS, 

204 } 

205 

206 def test_parallelizable_models(self) -> None: 

207 # HF and FT have work == num_subscenes (>1), so parallelizable 

208 assert PODCAST_WORKFLOW.is_parallelizable(Model.HF) 

209 assert PODCAST_WORKFLOW.is_parallelizable(Model.FT) 

210 # VAE and UPSCALER have large work counts 

211 assert PODCAST_WORKFLOW.is_parallelizable(Model.HF_VAE) 

212 assert PODCAST_WORKFLOW.is_parallelizable(Model.UPSCALER) 

213 # Singleton models are not parallelizable 

214 assert not PODCAST_WORKFLOW.is_parallelizable(Model.GEMMA) 

215 assert not PODCAST_WORKFLOW.is_parallelizable(Model.FLUX) 

216 assert not PODCAST_WORKFLOW.is_parallelizable(Model.OTHERS) 

217 

218 def test_rebuild_matches(self) -> None: 

219 """Rebuilding the config with same params produces the same result.""" 

220 secs = self.PODCAST_TOTAL_SECONDS 

221 num_scenes = _get_num_scenes(secs) 

222 num_subscenes = _get_num_subscenes(secs) 

223 fresh = build_workflow_config( 

224 total_video_seconds=secs, 

225 input_tokens=TOTAL_INPUT_TOKENS, 

226 model_work=_video_gen_work( 

227 secs, 

228 num_scenes, 

229 num_subscenes, 

230 model_work_overrides={Model.FLUX: 1}, 

231 ), 

232 ) 

233 assert PODCAST_WORKFLOW == fresh 

234 

235 

236class TestShortsWorkflow: 

237 """Shorts workflow config tests.""" 

238 SHORTS_TOTAL_SECONDS = int(2 * SECONDS_IN_HOUR) # 7200 s 

239 

240 def test_total_video_seconds(self) -> None: 

241 assert SHORTS_WORKFLOW.total_video_seconds == self.SHORTS_TOTAL_SECONDS 

242 

243 def test_input_tokens(self) -> None: 

244 expected = int(2 * SECONDS_IN_HOUR * 500) 

245 assert SHORTS_WORKFLOW.total_input_tokens == expected 

246 

247 def test_num_scenes_override(self) -> None: 

248 """ShortsWorkflow overrides num_scenes to derive from input video.""" 

249 assert SHORTS_WORKFLOW.total_scenes == self.SHORTS_TOTAL_SECONDS // 10 

250 assert SHORTS_WORKFLOW.total_scenes == 720 

251 

252 def test_num_subscenes_uses_base(self) -> None: 

253 """num_subscenes still uses the base formula.""" 

254 assert SHORTS_WORKFLOW.total_subscenes == math.ceil(self.SHORTS_TOTAL_SECONDS / SUBSCENE_SECONDS) 

255 

256 def test_model_work_keys(self) -> None: 

257 assert set(SHORTS_WORKFLOW.model_work.keys()) == {Model.GEMMA, Model.OTHERS} 

258 

259 def test_model_work_values(self) -> None: 

260 assert SHORTS_WORKFLOW.model_work[Model.GEMMA] == SHORTS_WORKFLOW.total_scenes 

261 assert SHORTS_WORKFLOW.model_work[Model.OTHERS] == 1 

262 

263 def test_gemma_parallelizable(self) -> None: 

264 # 720 scenes → GEMMA work = 720, parallelizable 

265 assert SHORTS_WORKFLOW.is_parallelizable(Model.GEMMA) 

266 assert not SHORTS_WORKFLOW.is_parallelizable(Model.OTHERS) 

267 

268 def test_config_models_only_gemma_and_others(self) -> None: 

269 assert set(SHORTS_WORKFLOW.models) == {Model.GEMMA, Model.OTHERS} 

270 

271 def test_no_video_generation_models(self) -> None: 

272 """Shorts workflow does not include video generation models.""" 

273 for model in (Model.FLUX, Model.HF, Model.HF_VAE, Model.FT, Model.FT_VAE, Model.UPSCALER): 

274 assert model not in SHORTS_WORKFLOW.model_work 

275 

276 def test_rebuild_matches(self) -> None: 

277 """Rebuilding the config with same params produces the same result.""" 

278 secs = self.SHORTS_TOTAL_SECONDS 

279 scenes = secs // 10 

280 fresh = build_workflow_config( 

281 total_video_seconds=secs, 

282 input_tokens=int(2 * SECONDS_IN_HOUR * 500), 

283 model_work={Model.GEMMA: scenes, Model.OTHERS: 1}, 

284 num_scenes_override=scenes, 

285 ) 

286 assert SHORTS_WORKFLOW == fresh 

287 

288 

289class TestMovieWorkflow: 

290 """Movie workflow config tests.""" 

291 MOVIE_TOTAL_SECONDS = int(2 * SECONDS_IN_HOUR) # 7200 s 

292 

293 def test_total_video_seconds(self) -> None: 

294 assert MOVIE_WORKFLOW.total_video_seconds == self.MOVIE_TOTAL_SECONDS 

295 

296 def test_input_tokens(self) -> None: 

297 assert MOVIE_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

298 

299 def test_num_subscenes(self) -> None: 

300 expected = math.ceil(self.MOVIE_TOTAL_SECONDS / SUBSCENE_SECONDS) 

301 assert MOVIE_WORKFLOW.total_subscenes == expected 

302 

303 def test_num_scenes(self) -> None: 

304 expected_subscenes = math.ceil(self.MOVIE_TOTAL_SECONDS / SUBSCENE_SECONDS) 

305 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

306 assert MOVIE_WORKFLOW.total_scenes == expected_scenes 

307 

308 def test_model_work_keys(self) -> None: 

309 expected_models = { 

310 Model.GEMMA, Model.FLUX, 

311 Model.HF, Model.HF_VAE, 

312 Model.FT, Model.FT_VAE, 

313 Model.UPSCALER, Model.OTHERS, 

314 } 

315 assert set(MOVIE_WORKFLOW.model_work.keys()) == expected_models 

316 

317 def test_model_work_singleton_values(self) -> None: 

318 assert MOVIE_WORKFLOW.model_work[Model.GEMMA] == 1 

319 assert MOVIE_WORKFLOW.model_work[Model.OTHERS] == 1 

320 

321 def test_model_work_flux_per_scene(self) -> None: 

322 """Movie generates one FLUX image per scene (unlike Podcast which uses 1).""" 

323 assert MOVIE_WORKFLOW.model_work[Model.FLUX] == MOVIE_WORKFLOW.total_scenes 

324 

325 def test_model_work_subscene_values(self) -> None: 

326 assert MOVIE_WORKFLOW.model_work[Model.HF] == MOVIE_WORKFLOW.total_subscenes 

327 assert MOVIE_WORKFLOW.model_work[Model.FT] == MOVIE_WORKFLOW.total_subscenes 

328 

329 def test_model_work_frame_values(self) -> None: 

330 assert MOVIE_WORKFLOW.model_work[Model.HF_VAE] == self.MOVIE_TOTAL_SECONDS * FPS[Model.HF] 

331 assert MOVIE_WORKFLOW.model_work[Model.UPSCALER] == self.MOVIE_TOTAL_SECONDS * FPS[Model.FT] 

332 

333 def test_config_models(self) -> None: 

334 assert set(MOVIE_WORKFLOW.models) == { 

335 Model.GEMMA, Model.FLUX, 

336 Model.HF, Model.HF_VAE, 

337 Model.FT, Model.FT_VAE, 

338 Model.UPSCALER, Model.OTHERS, 

339 } 

340 

341 def test_flux_parallelizable(self) -> None: 

342 """Movie has FLUX work == num_scenes (>1), so it's parallelizable.""" 

343 assert MOVIE_WORKFLOW.is_parallelizable(Model.FLUX) 

344 

345 def test_parallelizable_models(self) -> None: 

346 assert MOVIE_WORKFLOW.is_parallelizable(Model.HF) 

347 assert MOVIE_WORKFLOW.is_parallelizable(Model.FT) 

348 assert MOVIE_WORKFLOW.is_parallelizable(Model.HF_VAE) 

349 assert MOVIE_WORKFLOW.is_parallelizable(Model.UPSCALER) 

350 assert MOVIE_WORKFLOW.is_parallelizable(Model.FLUX) 

351 assert not MOVIE_WORKFLOW.is_parallelizable(Model.GEMMA) 

352 assert not MOVIE_WORKFLOW.is_parallelizable(Model.OTHERS) 

353 

354 def test_rebuild_matches(self) -> None: 

355 """Rebuilding the config with same params produces the same result.""" 

356 secs = self.MOVIE_TOTAL_SECONDS 

357 num_sc = _get_num_scenes(secs) 

358 num_ss = _get_num_subscenes(secs) 

359 fresh = build_workflow_config( 

360 total_video_seconds=secs, 

361 input_tokens=TOTAL_INPUT_TOKENS, 

362 model_work=_video_gen_work( 

363 secs, 

364 num_sc, 

365 num_ss, 

366 model_work_overrides={Model.FLUX: "num_scenes"}, 

367 ), 

368 ) 

369 assert MOVIE_WORKFLOW == fresh 

370 

371 

372class TestWorkflowComparisons: 

373 """Cross-workflow comparison tests.""" 

374 

375 def test_movie_more_subscenes_than_podcast(self) -> None: 

376 assert MOVIE_WORKFLOW.total_subscenes > PODCAST_WORKFLOW.total_subscenes 

377 

378 def test_movie_more_scenes_than_podcast(self) -> None: 

379 assert MOVIE_WORKFLOW.total_scenes > PODCAST_WORKFLOW.total_scenes 

380 

381 def test_shorts_has_most_scenes(self) -> None: 

382 """Shorts derives scenes from 2h input → 720 scenes, more than movie.""" 

383 assert SHORTS_WORKFLOW.total_scenes > MOVIE_WORKFLOW.total_scenes 

384 

385 def test_shorts_fewer_model_types_than_podcast(self) -> None: 

386 assert len(SHORTS_WORKFLOW.model_work) < len(PODCAST_WORKFLOW.model_work) 

387 

388 def test_movie_and_podcast_same_model_types(self) -> None: 

389 assert set(MOVIE_WORKFLOW.model_work.keys()) == set(PODCAST_WORKFLOW.model_work.keys()) 

390 

391 def test_movie_flux_work_exceeds_podcast(self) -> None: 

392 """Movie needs FLUX per scene; Podcast only needs 1.""" 

393 assert MOVIE_WORKFLOW.model_work[Model.FLUX] > PODCAST_WORKFLOW.model_work[Model.FLUX] 

394 

395 def test_movie_vae_work_exceeds_podcast(self) -> None: 

396 assert MOVIE_WORKFLOW.model_work[Model.HF_VAE] > PODCAST_WORKFLOW.model_work[Model.HF_VAE] 

397 

398 

399class TestAnimatedStoryWorkflow: 

400 """Animated Story workflow config tests.""" 

401 ANIMATED_STORY_TOTAL_SECONDS = int(10 * SECONDS_IN_MINUTE) # 600 s 

402 

403 def test_total_video_seconds(self) -> None: 

404 assert ANIMATED_STORY_WORKFLOW.total_video_seconds == self.ANIMATED_STORY_TOTAL_SECONDS 

405 

406 def test_input_tokens(self) -> None: 

407 assert ANIMATED_STORY_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

408 

409 def test_num_subscenes(self) -> None: 

410 expected = math.ceil(self.ANIMATED_STORY_TOTAL_SECONDS / SUBSCENE_SECONDS) 

411 assert ANIMATED_STORY_WORKFLOW.total_subscenes == expected 

412 

413 def test_num_scenes(self) -> None: 

414 expected_subscenes = math.ceil(self.ANIMATED_STORY_TOTAL_SECONDS / SUBSCENE_SECONDS) 

415 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

416 assert ANIMATED_STORY_WORKFLOW.total_scenes == expected_scenes 

417 

418 def test_model_work_keys(self) -> None: 

419 expected_models = { 

420 Model.GEMMA, Model.FLUX, 

421 Model.HF, Model.HF_VAE, 

422 Model.FT, Model.FT_VAE, 

423 Model.UPSCALER, Model.OTHERS, 

424 } 

425 assert set(ANIMATED_STORY_WORKFLOW.model_work.keys()) == expected_models 

426 

427 def test_model_work_singleton_values(self) -> None: 

428 assert ANIMATED_STORY_WORKFLOW.model_work[Model.GEMMA] == 1 

429 assert ANIMATED_STORY_WORKFLOW.model_work[Model.FLUX] == 1 

430 assert ANIMATED_STORY_WORKFLOW.model_work[Model.OTHERS] == 1 

431 

432 def test_model_work_subscene_values(self) -> None: 

433 assert ANIMATED_STORY_WORKFLOW.model_work[Model.HF] == ANIMATED_STORY_WORKFLOW.total_subscenes 

434 assert ANIMATED_STORY_WORKFLOW.model_work[Model.FT] == ANIMATED_STORY_WORKFLOW.total_subscenes 

435 

436 def test_model_work_frame_values(self) -> None: 

437 assert ANIMATED_STORY_WORKFLOW.model_work[Model.HF_VAE] == self.ANIMATED_STORY_TOTAL_SECONDS * FPS[Model.HF] 

438 assert ANIMATED_STORY_WORKFLOW.model_work[Model.UPSCALER] == self.ANIMATED_STORY_TOTAL_SECONDS * FPS[Model.FT] 

439 

440 def test_model_work_matches_podcast(self) -> None: 

441 """AnimatedStory has identical model_work to Podcast.""" 

442 assert ANIMATED_STORY_WORKFLOW.model_work == PODCAST_WORKFLOW.model_work 

443 

444 def test_hf_num_steps_override(self) -> None: 

445 """HF denoising steps should be 5% higher than base for LoRA overhead.""" 

446 assert ANIMATED_STORY_WORKFLOW.num_steps[Model.HF] == int(NUM_STEPS[Model.HF] * 1.05) 

447 

448 def test_other_num_steps_unchanged(self) -> None: 

449 """Non-HF num_steps should remain at their default values.""" 

450 assert ANIMATED_STORY_WORKFLOW.num_steps[Model.FLUX] == NUM_STEPS[Model.FLUX] 

451 assert ANIMATED_STORY_WORKFLOW.num_steps[Model.FT] == NUM_STEPS[Model.FT] 

452 

453 def test_config_models(self) -> None: 

454 assert set(ANIMATED_STORY_WORKFLOW.models) == { 

455 Model.GEMMA, Model.FLUX, 

456 Model.HF, Model.HF_VAE, 

457 Model.FT, Model.FT_VAE, 

458 Model.UPSCALER, Model.OTHERS, 

459 } 

460 

461 def test_parallelizable_models(self) -> None: 

462 assert ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.HF) 

463 assert ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.FT) 

464 assert ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.HF_VAE) 

465 assert ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.UPSCALER) 

466 assert not ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.GEMMA) 

467 assert not ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.FLUX) 

468 assert not ANIMATED_STORY_WORKFLOW.is_parallelizable(Model.OTHERS) 

469 

470 def test_rebuild_matches(self) -> None: 

471 """Rebuilding the config with same params produces the same result.""" 

472 secs = self.ANIMATED_STORY_TOTAL_SECONDS 

473 num_scenes = _get_num_scenes(secs) 

474 num_subscenes = _get_num_subscenes(secs) 

475 fresh = build_workflow_config( 

476 total_video_seconds=secs, 

477 input_tokens=TOTAL_INPUT_TOKENS, 

478 model_work=_video_gen_work( 

479 secs, 

480 num_scenes, 

481 num_subscenes, 

482 model_work_overrides={Model.FLUX: 1} 

483 ), 

484 num_steps_override={Model.HF: int(NUM_STEPS[Model.HF] * 1.05)}, 

485 ) 

486 assert ANIMATED_STORY_WORKFLOW == fresh 

487 

488 

489class TestLectureWorkflow: 

490 """Lecture workflow config tests.""" 

491 LECTURE_TOTAL_SECONDS = int(5 * SECONDS_IN_MINUTE) # 300 s 

492 

493 def test_total_video_seconds(self) -> None: 

494 assert LECTURE_WORKFLOW.total_video_seconds == self.LECTURE_TOTAL_SECONDS 

495 

496 def test_input_tokens(self) -> None: 

497 assert LECTURE_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

498 

499 def test_num_subscenes(self) -> None: 

500 expected = math.ceil(self.LECTURE_TOTAL_SECONDS / SUBSCENE_SECONDS) 

501 assert LECTURE_WORKFLOW.total_subscenes == expected 

502 

503 def test_num_scenes(self) -> None: 

504 expected_subscenes = math.ceil(self.LECTURE_TOTAL_SECONDS / SUBSCENE_SECONDS) 

505 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

506 assert LECTURE_WORKFLOW.total_scenes == expected_scenes 

507 

508 def test_model_work_keys(self) -> None: 

509 expected_models = { 

510 Model.GEMMA, Model.FLUX, 

511 Model.HF, Model.HF_VAE, 

512 Model.FT, Model.FT_VAE, 

513 Model.UPSCALER, Model.OTHERS, 

514 } 

515 assert set(LECTURE_WORKFLOW.model_work.keys()) == expected_models 

516 

517 def test_model_work_singleton_values(self) -> None: 

518 assert LECTURE_WORKFLOW.model_work[Model.GEMMA] == 1 

519 assert LECTURE_WORKFLOW.model_work[Model.OTHERS] == 1 

520 

521 def test_model_work_flux_per_scene(self) -> None: 

522 """Lecture generates one FLUX image per scene (like Movie).""" 

523 assert LECTURE_WORKFLOW.model_work[Model.FLUX] == LECTURE_WORKFLOW.total_scenes 

524 

525 def test_model_work_subscene_values(self) -> None: 

526 assert LECTURE_WORKFLOW.model_work[Model.HF] == LECTURE_WORKFLOW.total_subscenes 

527 assert LECTURE_WORKFLOW.model_work[Model.FT] == LECTURE_WORKFLOW.total_subscenes 

528 

529 def test_model_work_frame_values(self) -> None: 

530 assert LECTURE_WORKFLOW.model_work[Model.HF_VAE] == self.LECTURE_TOTAL_SECONDS * FPS[Model.HF] 

531 assert LECTURE_WORKFLOW.model_work[Model.UPSCALER] == self.LECTURE_TOTAL_SECONDS * FPS[Model.FT] 

532 

533 def test_config_models(self) -> None: 

534 assert set(LECTURE_WORKFLOW.models) == { 

535 Model.GEMMA, Model.FLUX, 

536 Model.HF, Model.HF_VAE, 

537 Model.FT, Model.FT_VAE, 

538 Model.UPSCALER, Model.OTHERS, 

539 } 

540 

541 def test_flux_parallelizable(self) -> None: 

542 """Lecture has FLUX work == num_scenes (>1), so it's parallelizable.""" 

543 assert LECTURE_WORKFLOW.is_parallelizable(Model.FLUX) 

544 

545 def test_parallelizable_models(self) -> None: 

546 assert LECTURE_WORKFLOW.is_parallelizable(Model.HF) 

547 assert LECTURE_WORKFLOW.is_parallelizable(Model.FT) 

548 assert LECTURE_WORKFLOW.is_parallelizable(Model.HF_VAE) 

549 assert LECTURE_WORKFLOW.is_parallelizable(Model.UPSCALER) 

550 assert LECTURE_WORKFLOW.is_parallelizable(Model.FLUX) 

551 assert not LECTURE_WORKFLOW.is_parallelizable(Model.GEMMA) 

552 assert not LECTURE_WORKFLOW.is_parallelizable(Model.OTHERS) 

553 

554 def test_rebuild_matches(self) -> None: 

555 """Rebuilding the config with same params produces the same result.""" 

556 secs = self.LECTURE_TOTAL_SECONDS 

557 num_sc = _get_num_scenes(secs) 

558 num_ss = _get_num_subscenes(secs) 

559 fresh = build_workflow_config( 

560 total_video_seconds=secs, 

561 input_tokens=TOTAL_INPUT_TOKENS, 

562 model_work=_video_gen_work( 

563 secs, 

564 num_sc, 

565 num_ss, 

566 model_work_overrides={Model.FLUX: "num_scenes"}, 

567 ), 

568 ) 

569 assert LECTURE_WORKFLOW == fresh 

570 

571 def test_shorter_than_podcast(self) -> None: 

572 """Lecture is 5 min vs Podcast's 10 min.""" 

573 assert LECTURE_WORKFLOW.total_video_seconds < PODCAST_WORKFLOW.total_video_seconds 

574 assert LECTURE_WORKFLOW.total_video_seconds == PODCAST_WORKFLOW.total_video_seconds // 2 

575 

576 def test_fewer_subscenes_than_podcast(self) -> None: 

577 assert LECTURE_WORKFLOW.total_subscenes < PODCAST_WORKFLOW.total_subscenes 

578 

579 def test_more_flux_work_than_podcast(self) -> None: 

580 """Lecture needs FLUX per scene; Podcast only needs 1.""" 

581 assert LECTURE_WORKFLOW.model_work[Model.FLUX] > PODCAST_WORKFLOW.model_work[Model.FLUX] 

582 

583 

584class TestDubbingWorkflow: 

585 """Dubbing workflow config tests.""" 

586 

587 DUBBING_TOTAL_SECONDS = int(10 * SECONDS_IN_MINUTE) # 600 s 

588 

589 def test_total_video_seconds(self) -> None: 

590 assert DUBBING_WORKFLOW.total_video_seconds == self.DUBBING_TOTAL_SECONDS 

591 

592 def test_input_tokens(self) -> None: 

593 assert DUBBING_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

594 

595 def test_num_subscenes(self) -> None: 

596 expected = math.ceil(self.DUBBING_TOTAL_SECONDS / SUBSCENE_SECONDS) 

597 assert DUBBING_WORKFLOW.total_subscenes == expected 

598 

599 def test_num_scenes(self) -> None: 

600 expected_subscenes = math.ceil(self.DUBBING_TOTAL_SECONDS / SUBSCENE_SECONDS) 

601 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

602 assert DUBBING_WORKFLOW.total_scenes == expected_scenes 

603 

604 def test_model_work_keys(self) -> None: 

605 expected_models = { 

606 Model.GEMMA, 

607 Model.HF, Model.HF_VAE, 

608 Model.FT, Model.FT_VAE, 

609 Model.UPSCALER, Model.OTHERS, 

610 } 

611 assert set(DUBBING_WORKFLOW.model_work.keys()) == expected_models 

612 

613 def test_no_flux(self) -> None: 

614 """Dubbing workflow should not include FLUX.""" 

615 assert Model.FLUX not in DUBBING_WORKFLOW.model_work 

616 

617 def test_model_work_singleton_values(self) -> None: 

618 assert DUBBING_WORKFLOW.model_work[Model.GEMMA] == 1 

619 assert DUBBING_WORKFLOW.model_work[Model.OTHERS] == 2 

620 

621 def test_model_work_subscene_values(self) -> None: 

622 assert DUBBING_WORKFLOW.model_work[Model.HF] == DUBBING_WORKFLOW.total_subscenes 

623 assert DUBBING_WORKFLOW.model_work[Model.FT] == DUBBING_WORKFLOW.total_subscenes 

624 

625 def test_model_work_frame_values(self) -> None: 

626 assert DUBBING_WORKFLOW.model_work[Model.HF_VAE] == self.DUBBING_TOTAL_SECONDS * FPS[Model.HF] 

627 assert DUBBING_WORKFLOW.model_work[Model.UPSCALER] == self.DUBBING_TOTAL_SECONDS * FPS[Model.FT] 

628 

629 def test_config_models(self) -> None: 

630 assert set(DUBBING_WORKFLOW.models) == { 

631 Model.GEMMA, 

632 Model.HF, Model.HF_VAE, 

633 Model.FT, Model.FT_VAE, 

634 Model.UPSCALER, Model.OTHERS, 

635 } 

636 

637 def test_parallelizable_models(self) -> None: 

638 assert DUBBING_WORKFLOW.is_parallelizable(Model.HF) 

639 assert DUBBING_WORKFLOW.is_parallelizable(Model.FT) 

640 assert DUBBING_WORKFLOW.is_parallelizable(Model.HF_VAE) 

641 assert DUBBING_WORKFLOW.is_parallelizable(Model.UPSCALER) 

642 assert DUBBING_WORKFLOW.is_parallelizable(Model.OTHERS) # work=2 → parallelizable 

643 assert not DUBBING_WORKFLOW.is_parallelizable(Model.GEMMA) 

644 

645 def test_others_work_is_two(self) -> None: 

646 """Dubbing has OTHERS work = 2, unlike Podcast which has 1.""" 

647 assert DUBBING_WORKFLOW.model_work[Model.OTHERS] == 2 

648 assert DUBBING_WORKFLOW.model_work[Model.OTHERS] > PODCAST_WORKFLOW.model_work[Model.OTHERS] 

649 

650 def test_rebuild_matches(self) -> None: 

651 """Rebuilding the config with same params produces the same result.""" 

652 secs = self.DUBBING_TOTAL_SECONDS 

653 num_scenes = _get_num_scenes(secs) 

654 num_ss = _get_num_subscenes(secs) 

655 work = _video_gen_work( 

656 secs, 

657 num_scenes, 

658 num_ss, 

659 model_work_overrides={ 

660 Model.FLUX: None, 

661 Model.OTHERS: 2, 

662 }, 

663 ) 

664 fresh = build_workflow_config( 

665 total_video_seconds=secs, 

666 input_tokens=TOTAL_INPUT_TOKENS, 

667 model_work=work, 

668 ) 

669 assert DUBBING_WORKFLOW == fresh 

670 

671 

672class TestEditingWorkflow: 

673 """Editing workflow config tests.""" 

674 

675 EDITING_TOTAL_SECONDS = int(10 * SECONDS_IN_MINUTE) # 600 s 

676 

677 def test_total_video_seconds(self) -> None: 

678 assert EDITING_WORKFLOW.total_video_seconds == self.EDITING_TOTAL_SECONDS 

679 

680 def test_input_tokens(self) -> None: 

681 assert EDITING_WORKFLOW.total_input_tokens == TOTAL_INPUT_TOKENS 

682 

683 def test_num_subscenes(self) -> None: 

684 expected = math.ceil(self.EDITING_TOTAL_SECONDS / SUBSCENE_SECONDS) 

685 assert EDITING_WORKFLOW.total_subscenes == expected 

686 

687 def test_num_scenes(self) -> None: 

688 expected_subscenes = math.ceil(self.EDITING_TOTAL_SECONDS / SUBSCENE_SECONDS) 

689 expected_scenes = math.ceil(expected_subscenes / SUBSCENES_PER_SCENE) 

690 assert EDITING_WORKFLOW.total_scenes == expected_scenes 

691 

692 def test_model_work_keys(self) -> None: 

693 expected_models = { 

694 Model.HF, Model.HF_VAE, 

695 Model.FT, Model.FT_VAE, 

696 Model.UPSCALER, 

697 } 

698 assert set(EDITING_WORKFLOW.model_work.keys()) == expected_models 

699 

700 def test_no_gemma(self) -> None: 

701 """Editing workflow should not include GEMMA.""" 

702 assert Model.GEMMA not in EDITING_WORKFLOW.model_work 

703 

704 def test_no_flux(self) -> None: 

705 """Editing workflow should not include FLUX.""" 

706 assert Model.FLUX not in EDITING_WORKFLOW.model_work 

707 

708 def test_no_others(self) -> None: 

709 """Editing workflow should not include OTHERS.""" 

710 assert Model.OTHERS not in EDITING_WORKFLOW.model_work 

711 

712 def test_model_work_subscene_values(self) -> None: 

713 assert EDITING_WORKFLOW.model_work[Model.HF] == EDITING_WORKFLOW.total_subscenes 

714 assert EDITING_WORKFLOW.model_work[Model.FT] == EDITING_WORKFLOW.total_subscenes 

715 

716 def test_model_work_frame_values(self) -> None: 

717 assert EDITING_WORKFLOW.model_work[Model.HF_VAE] == self.EDITING_TOTAL_SECONDS * FPS[Model.HF] 

718 assert EDITING_WORKFLOW.model_work[Model.UPSCALER] == self.EDITING_TOTAL_SECONDS * FPS[Model.FT] 

719 

720 def test_config_models(self) -> None: 

721 assert set(EDITING_WORKFLOW.models) == { 

722 Model.HF, Model.HF_VAE, 

723 Model.FT, Model.FT_VAE, 

724 Model.UPSCALER, 

725 } 

726 

727 def test_parallelizable_models(self) -> None: 

728 assert EDITING_WORKFLOW.is_parallelizable(Model.HF) 

729 assert EDITING_WORKFLOW.is_parallelizable(Model.HF_VAE) 

730 assert EDITING_WORKFLOW.is_parallelizable(Model.FT) 

731 assert EDITING_WORKFLOW.is_parallelizable(Model.FT_VAE) 

732 assert EDITING_WORKFLOW.is_parallelizable(Model.UPSCALER) 

733 

734 def test_fewer_models_than_podcast(self) -> None: 

735 """Editing has fewer model types than Podcast (no GEMMA, FLUX, OTHERS).""" 

736 assert len(EDITING_WORKFLOW.model_work) < len(PODCAST_WORKFLOW.model_work) 

737 

738 def test_rebuild_matches(self) -> None: 

739 """Rebuilding the config with same params produces the same result.""" 

740 secs = self.EDITING_TOTAL_SECONDS 

741 num_scenes = _get_num_scenes(secs) 

742 num_ss = _get_num_subscenes(secs) 

743 work = _video_gen_work( 

744 secs, 

745 num_scenes,