| 215 | |
| 216 | |
| 217 | class VeoLiteFlowClientTests(unittest.IsolatedAsyncioTestCase): |
| 218 | async def asyncSetUp(self): |
| 219 | self.client = FlowClient(proxy_manager=None) |
| 220 | self.client._acquire_video_launch_gate = AsyncMock(return_value=(True, None, None)) |
| 221 | self.client._release_video_launch_gate = AsyncMock() |
| 222 | self.client._get_recaptcha_token = AsyncMock(return_value=("recaptcha-token", "browser-1")) |
| 223 | self.client._notify_browser_captcha_request_finished = AsyncMock() |
| 224 | |
| 225 | async def test_generate_video_text_uses_v2_payload_for_lite(self): |
| 226 | captured = {} |
| 227 | |
| 228 | async def fake_make_request(method, url, json_data, use_at, at_token, **kwargs): |
| 229 | captured["url"] = url |
| 230 | captured["json_data"] = json_data |
| 231 | return {"operations": [{"operation": {"name": "task-1"}}]} |
| 232 | |
| 233 | self.client._make_request = AsyncMock(side_effect=fake_make_request) |
| 234 | |
| 235 | await self.client.generate_video_text( |
| 236 | at="at-token", |
| 237 | project_id="project-1", |
| 238 | prompt="猫猫", |
| 239 | model_key="veo_3_1_t2v_lite", |
| 240 | aspect_ratio="VIDEO_ASPECT_RATIO_LANDSCAPE", |
| 241 | use_v2_model_config=True, |
| 242 | ) |
| 243 | |
| 244 | json_data = captured["json_data"] |
| 245 | request_data = json_data["requests"][0] |
| 246 | self.assertTrue(json_data["useV2ModelConfig"]) |
| 247 | self.assertIn("batchId", json_data["mediaGenerationContext"]) |
| 248 | self.assertEqual( |
| 249 | request_data["textInput"]["structuredPrompt"]["parts"][0]["text"], |
| 250 | "猫猫", |
| 251 | ) |
| 252 | self.assertNotIn("prompt", request_data["textInput"]) |
| 253 | self.assertEqual(request_data["videoModelKey"], "veo_3_1_t2v_lite") |
| 254 | self.assertEqual( |
| 255 | json_data["mediaGenerationContext"]["audioFailurePreference"], |
| 256 | "BLOCK_SILENCED_VIDEOS", |
| 257 | ) |
| 258 | |
| 259 | async def test_generate_video_text_normalizes_media_only_create_response(self): |
| 260 | captured = {} |
| 261 | |
| 262 | async def fake_make_request(method, url, json_data, use_at, at_token, **kwargs): |
| 263 | captured["json_data"] = json_data |
| 264 | return { |
| 265 | "remainingCredits": 30, |
| 266 | "workflows": [ |
| 267 | { |
| 268 | "name": "workflow-1", |
| 269 | "metadata": {"primaryMediaId": "media-1"}, |
| 270 | "projectId": "project-1", |
| 271 | } |
| 272 | ], |
| 273 | "media": [ |
| 274 | { |
nothing calls this directly
no outgoing calls
no test coverage detected