| 1 | import os |
| 2 | import sys |
| 3 | import tempfile |
| 4 | import types |
| 5 | import unittest |
| 6 | from pathlib import Path |
| 7 | from unittest.mock import patch |
| 8 | |
| 9 | from pydantic import ValidationError |
| 10 | |
| 11 | sys.path.insert(0, str(Path(__file__).parent.parent.parent)) |
| 12 | |
| 13 | from app.config import config |
| 14 | from app.models.schema import VideoScriptRequest, VideoSocialMetadataRequest |
| 15 | from app.services import llm |
| 16 | |
| 17 | RUN_INTEGRATION_TESTS = os.environ.get("MPT_RUN_INTEGRATION_TESTS", "").lower() in { |
| 18 | "1", |
| 19 | "true", |
| 20 | "yes", |
| 21 | } |
| 22 | |
| 23 | |
| 24 | class TestScriptPromptOptions(unittest.TestCase): |
| 25 | def test_normalize_text_response_removes_think_blocks(self): |
| 26 | """ |
| 27 | reasoning 模型可能返回 `<think>...</think>`。脚本生成链路必须只保留 |
| 28 | 最终正文，避免思考过程进入字幕和配音。 |
| 29 | """ |
| 30 | result = llm._normalize_text_response( |
| 31 | "<think>\nI should reason here.\n</think>\n测试成功", |
| 32 | "minimax", |
| 33 | ) |
| 34 | |
| 35 | self.assertEqual(result, "测试成功") |
| 36 | |
| 37 | def test_normalize_text_response_rejects_think_only_response(self): |
| 38 | """ |
| 39 | 如果模型只返回思考块而没有最终答案，应视为空内容，触发重试或明确错误。 |
| 40 | """ |
| 41 | with self.assertRaises(ValueError): |
| 42 | llm._normalize_text_response("<think>hidden reasoning</think>", "minimax") |
| 43 | |
| 44 | def test_normalize_text_response_removes_unclosed_think_block(self): |
| 45 | """ |
| 46 | 某些网关可能因为截断只返回未闭合的 `<think>`。这种内容同样不能 |
| 47 | 进入最终脚本；如果清理后没有正文，就应该按空响应处理。 |
| 48 | """ |
| 49 | with self.assertRaises(ValueError): |
| 50 | llm._normalize_text_response("<think>hidden reasoning", "minimax") |
| 51 | |
| 52 | def test_build_script_prompt_appends_advanced_requirements(self): |
| 53 | """ |
| 54 | 高级文案要求只作为附加约束，不替换默认系统提示词。 |
| 55 | 这样普通用户不配置时仍然走稳定默认规则，高级用户也能细化风格。 |
| 56 | """ |
| 57 | prompt = llm.build_script_prompt( |
| 58 | video_subject="咖啡", |
| 59 | language="zh-CN", |
| 60 | paragraph_number=3, |
| 61 | video_script_prompt="语气轻松，面向程序员", |
| 62 | ) |
| 63 | |
| 64 | self.assertIn("# Role: Video Script Generator", prompt) |
| 65 | self.assertIn("- video subject: 咖啡", prompt) |
| 66 | self.assertIn("- number of paragraphs: 3", prompt) |
| 67 | self.assertIn("- language: zh-CN", prompt) |
| 68 | self.assertIn("# Additional User Requirements:", prompt) |
| 69 | self.assertIn("语气轻松，面向程序员", prompt) |
| 70 | |
| 71 | def test_custom_system_prompt_keeps_runtime_context(self): |
| 72 | """ |
| 73 | 自定义 system prompt 会替换默认脚本规则，但视频主题、语言、段落数 |
| 74 | 仍由服务层统一追加，避免高级用户漏写必要上下文。 |
| 75 | """ |
| 76 | prompt = llm.build_script_prompt( |
| 77 | video_subject="露营", |
| 78 | language="en", |
| 79 | paragraph_number=2, |
| 80 | custom_system_prompt="Only write cinematic narration.", |
| 81 | ) |
| 82 | |
| 83 | self.assertNotIn("# Role: Video Script Generator", prompt) |
| 84 | self.assertIn("Only write cinematic narration.", prompt) |
| 85 | self.assertIn("- video subject: 露营", prompt) |
| 86 | self.assertIn("- number of paragraphs: 2", prompt) |
| 87 | self.assertIn("- language: en", prompt) |
| 88 | |
| 89 | def test_generate_script_sends_custom_prompt_to_llm(self): |
| 90 | captured = {} |
| 91 | |
| 92 | def fake_generate_response(prompt): |
| 93 | captured["prompt"] = prompt |
| 94 | return "第一段。\n\n第二段。" |
| 95 | |
| 96 | with patch.object(llm, "_generate_response", side_effect=fake_generate_response): |
| 97 | result = llm.generate_script( |
| 98 | video_subject="咖啡", |
| 99 | language="zh-CN", |
| 100 | paragraph_number=2, |
| 101 | video_script_prompt="开头更有悬念", |
| 102 | ) |
| 103 | |
| 104 | self.assertEqual(result, "第一段。\n\n第二段。") |
| 105 | self.assertIn("- number of paragraphs: 2", captured["prompt"]) |
| 106 | self.assertIn("开头更有悬念", captured["prompt"]) |
| 107 | |
| 108 | def test_generate_terms_can_request_script_ordered_keywords(self): |
| 109 | """ |
| 110 | 按文案顺序匹配素材依赖 LLM 返回有序关键词。这里不调用真实模型， |
| 111 | 只验证服务层会把“按脚本叙事顺序输出”的约束写入 prompt，避免 |
| 112 | 后续素材下载虽然顺序化，但关键词仍然是全局无序主题词。 |
| 113 | """ |
| 114 | captured = {} |
| 115 | |
| 116 | def fake_generate_response(prompt): |
| 117 | captured["prompt"] = prompt |
| 118 | return '["opening city", "middle office", "final sunset"]' |
| 119 | |
| 120 | with patch.object(llm, "_generate_response", side_effect=fake_generate_response): |
| 121 | result = llm.generate_terms( |
| 122 | video_subject="startup story", |
| 123 | video_script="First city. Then office. Finally sunset.", |
| 124 | amount=3, |
| 125 | match_script_order=True, |
| 126 | ) |
| 127 | |
| 128 | self.assertEqual(result, ["opening city", "middle office", "final sunset"]) |
| 129 | self.assertIn("chronological stock-video search terms", captured["prompt"]) |
| 130 | self.assertIn("same order as the script narration", captured["prompt"]) |
| 131 | |
| 132 | def test_video_script_request_rejects_invalid_advanced_options(self): |
| 133 | """ |
| 134 | API 请求模型需要限制高级 prompt 参数，避免外部调用绕过 WebUI |
| 135 | 传入异常段落数或超长提示词，导致模型成本和结果不可控。 |
| 136 | """ |
| 137 | with self.assertRaises(ValidationError): |
| 138 | VideoScriptRequest(video_subject="咖啡", paragraph_number=0) |
| 139 | |
| 140 | with self.assertRaises(ValidationError): |
| 141 | VideoScriptRequest( |
| 142 | video_subject="咖啡", |
| 143 | video_script_prompt="x" * (llm.MAX_SCRIPT_PROMPT_LENGTH + 1), |
| 144 | ) |
| 145 | |
| 146 | |
| 147 | class TestLiteLLMProvider(unittest.TestCase): |
| 148 | def setUp(self): |
| 149 | self.original_app_config = dict(config.app) |
| 150 | |
| 151 | def tearDown(self): |
| 152 | config.app.clear() |
| 153 | config.app.update(self.original_app_config) |
| 154 | |
| 155 | def _use_litellm_provider(self, model_name="openai/gpt-4o-mini"): |
| 156 | config.app["llm_provider"] = "litellm" |
| 157 | config.app["litellm_model_name"] = model_name |
| 158 | |
| 159 | def test_litellm_provider_returns_normalized_text(self): |
| 160 | """ |
| 161 | 验证 LiteLLM provider 的主路径不依赖真实网络和私有 API key。 |
| 162 | |
| 163 | 这里用 fake module 注入 `sys.modules`，直接覆盖动态 import 的 |
| 164 | `litellm.completion()`，确保测试稳定覆盖 `_generate_response()` 里的 |
| 165 | litellm 分支。 |
| 166 | """ |
| 167 | self._use_litellm_provider() |
| 168 | |
| 169 | fake_litellm = types.SimpleNamespace() |
| 170 | |
| 171 | def _completion(**kwargs): |
| 172 | self.assertEqual(kwargs["model"], "openai/gpt-4o-mini") |
| 173 | self.assertEqual( |
| 174 | kwargs["messages"], [{"role": "user", "content": "Say hello"}] |
| 175 | ) |
| 176 | self.assertTrue(kwargs["drop_params"]) |
| 177 | message = types.SimpleNamespace(content="hello\nworld") |
| 178 | choice = types.SimpleNamespace(message=message) |
| 179 | return types.SimpleNamespace(choices=[choice]) |
| 180 | |
| 181 | fake_litellm.completion = _completion |
| 182 | |
| 183 | with patch.dict(sys.modules, {"litellm": fake_litellm}): |
| 184 | result = llm._generate_response("Say hello") |
| 185 | |
| 186 | self.assertEqual(result, "helloworld") |
| 187 | |
| 188 | def test_litellm_provider_requires_model_name(self): |
| 189 | self._use_litellm_provider(model_name="") |
| 190 | |
| 191 | result = llm._generate_response("test") |
| 192 | |
| 193 | self.assertIn("Error:", result) |
| 194 | self.assertIn("model_name is not set", result) |
| 195 | |
| 196 | def test_litellm_provider_handles_empty_response(self): |
| 197 | self._use_litellm_provider() |
| 198 | |
| 199 | fake_litellm = types.SimpleNamespace( |
| 200 | completion=lambda **kwargs: types.SimpleNamespace(choices=[]) |
| 201 | ) |
| 202 | |
| 203 | with patch.dict(sys.modules, {"litellm": fake_litellm}): |
| 204 | result = llm._generate_response("test") |
| 205 | |
| 206 | self.assertIn("Error:", result) |
| 207 | self.assertIn("returned empty response", result) |
| 208 | |
| 209 | def test_litellm_provider_handles_empty_message(self): |
| 210 | """ |
| 211 | 某些 OpenAI-compatible 网关在内容过滤或安全拦截时会返回 |
| 212 | HTTP 200，但 `choices[0].message` 为 None。这里必须返回 |
| 213 | 可诊断的错误，而不是抛出 AttributeError。 |
| 214 | """ |
| 215 | self._use_litellm_provider() |
| 216 | |
| 217 | fake_litellm = types.SimpleNamespace( |
| 218 | completion=lambda **kwargs: types.SimpleNamespace( |
| 219 | choices=[types.SimpleNamespace(message=None)] |
| 220 | ) |
| 221 | ) |
| 222 | |
| 223 | with patch.dict(sys.modules, {"litellm": fake_litellm}): |
| 224 | result = llm._generate_response("test") |
| 225 | |
| 226 | self.assertIn("Error:", result) |
| 227 | self.assertIn("returned empty message", result) |
| 228 | |
| 229 | def test_sanitize_error_message_redacts_url_credentials_and_query_tokens(self): |
| 230 | message = ( |
| 231 | "request failed for " |
| 232 | "https://myuser:mypassword@proxy.example.com/v1/chat" |
| 233 | "?api_key=secret-key&token=secret-token&safe=value" |
| 234 | ) |
| 235 | |
| 236 | result = llm._sanitize_error_message(message) |
| 237 | |
| 238 | self.assertIn("https://***:***@proxy.example.com", result) |
| 239 | self.assertIn("api_key=***", result) |
| 240 | self.assertIn("token=***", result) |
| 241 | self.assertIn("safe=value", result) |
| 242 | self.assertNotIn("myuser", result) |
| 243 | self.assertNotIn("mypassword", result) |
| 244 | self.assertNotIn("secret-key", result) |
| 245 | self.assertNotIn("secret-token", result) |
| 246 | |
| 247 | def test_openai_provider_error_redacts_embedded_base_url_credentials(self): |
| 248 | """ |
| 249 | 自定义 OpenAI-compatible base_url 可能包含代理网关的 user:pass。 |
| 250 | SDK 抛错时常会把 URL 带回异常信息，这里验证最终返回给 WebUI/API 的 |
| 251 | `Error:` 文案不会泄露这些凭据。 |
| 252 | """ |
| 253 | config.app["llm_provider"] = "groq" |
| 254 | config.app["groq_api_key"] = "groq-key" |
| 255 | config.app["groq_model_name"] = "llama-3.3-70b-versatile" |
| 256 | config.app["groq_base_url"] = "https://myuser:mypassword@proxy.example.com/openai/v1" |
| 257 | |
| 258 | class FakeCompletions: |
| 259 | def create(self, **kwargs): |
| 260 | raise RuntimeError( |
| 261 | "connection failed: " |
| 262 | "https://myuser:mypassword@proxy.example.com/openai/v1" |
| 263 | "?access_token=secret-token" |
| 264 | ) |
| 265 | |
| 266 | fake_client = types.SimpleNamespace( |
| 267 | chat=types.SimpleNamespace(completions=FakeCompletions()) |
| 268 | ) |
| 269 | |
| 270 | with patch.object(llm, "OpenAI", return_value=fake_client): |
| 271 | result = llm._generate_response("test") |
| 272 | |
| 273 | self.assertIn("Error:", result) |
| 274 | self.assertIn("https://***:***@proxy.example.com", result) |
| 275 | self.assertIn("access_token=***", result) |
| 276 | self.assertNotIn("myuser", result) |
| 277 | self.assertNotIn("mypassword", result) |
| 278 | self.assertNotIn("secret-token", result) |
| 279 | |
| 280 | def test_openai_provider_still_uses_existing_path(self): |
| 281 | config.app["llm_provider"] = "openai" |
| 282 | config.app["openai_api_key"] = "" |
| 283 | config.app["openai_base_url"] = "https://api.openai.com/v1" |
| 284 | config.app["openai_model_name"] = "gpt-4o-mini" |
| 285 | |
| 286 | result = llm._generate_response("test") |
| 287 | |
| 288 | self.assertIn("Error:", result) |
| 289 | self.assertIn("api_key is not set", result) |
| 290 | self.assertNotIn("litellm", result.lower()) |
| 291 | |
| 292 | def _use_qwen_provider(self): |
| 293 | config.app["llm_provider"] = "qwen" |
| 294 | config.app["qwen_api_key"] = "qwen-key" |
| 295 | config.app["qwen_model_name"] = "qwen-max" |
| 296 | |
| 297 | def _patch_dashscope_generation(self, response): |
| 298 | class FakeGenerationResponse(dict): |
| 299 | pass |
| 300 | |
| 301 | fake_response = FakeGenerationResponse(response) |
| 302 | fake_response.status_code = response.get("status_code", 200) |
| 303 | fake_dashscope = types.SimpleNamespace( |
| 304 | api_key="", |
| 305 | Generation=types.SimpleNamespace(call=lambda **kwargs: fake_response), |
| 306 | ) |
| 307 | fake_dashscope_response = types.SimpleNamespace( |
| 308 | GenerationResponse=FakeGenerationResponse |
| 309 | ) |
| 310 | |
| 311 | return patch.dict( |
| 312 | sys.modules, |
| 313 | { |
| 314 | "dashscope": fake_dashscope, |
| 315 | "dashscope.api_entities": types.SimpleNamespace(), |
| 316 | "dashscope.api_entities.dashscope_response": fake_dashscope_response, |
| 317 | }, |
| 318 | ) |
| 319 | |
| 320 | def test_qwen_provider_reads_chat_choices_content(self): |
| 321 | """ |
| 322 | DashScope chat 模式会把文本放在 `output.choices[0].message.content`。 |
| 323 | 这里覆盖 issue #966 报告的 `output.text is None` 场景，避免再次触发 |
| 324 | `'NoneType' object has no attribute 'replace'`。 |
| 325 | """ |
| 326 | self._use_qwen_provider() |
| 327 | response = { |
| 328 | "output": { |
| 329 | "text": None, |
| 330 | "choices": [{"message": {"content": "你好\n世界"}}], |
| 331 | } |
| 332 | } |
| 333 | |
| 334 | with self._patch_dashscope_generation(response): |
| 335 | result = llm._generate_response("Say hello") |
| 336 | |
| 337 | self.assertEqual(result, "你好世界") |
| 338 | |
| 339 | def test_qwen_provider_falls_back_to_output_text(self): |
| 340 | """保留旧 DashScope completion 响应结构的兼容路径。""" |
| 341 | self._use_qwen_provider() |
| 342 | response = {"output": {"text": "旧格式\n响应"}} |
| 343 | |
| 344 | with self._patch_dashscope_generation(response): |
| 345 | result = llm._generate_response("Say hello") |
| 346 | |
| 347 | self.assertEqual(result, "旧格式响应") |
| 348 | |
| 349 | def test_qwen_provider_reports_empty_text(self): |
| 350 | """Qwen 空响应应返回可诊断错误，而不是底层 AttributeError。""" |
| 351 | self._use_qwen_provider() |
| 352 | response = {"output": {"text": None, "choices": [{"message": {"content": None}}]}} |
| 353 | |
| 354 | with self._patch_dashscope_generation(response): |
| 355 | result = llm._generate_response("Say hello") |
| 356 | |
| 357 | self.assertIn("Error:", result) |
| 358 | self.assertIn("returned empty text content", result) |
| 359 | self.assertNotIn("NoneType", result) |
| 360 | |
| 361 | def test_qwen_provider_reports_empty_choices(self): |
| 362 | """Qwen chat 响应 choices 为空时应返回明确错误。""" |
| 363 | self._use_qwen_provider() |
| 364 | response = {"output": {"text": None, "choices": []}} |
| 365 | |
| 366 | with self._patch_dashscope_generation(response): |
| 367 | result = llm._generate_response("Say hello") |
| 368 | |
| 369 | self.assertIn("Error:", result) |
| 370 | self.assertIn("returned empty choices", result) |
| 371 | self.assertNotIn("NoneType", result) |
| 372 | |
| 373 | def test_aihubmix_provider_uses_openai_compatible_client(self): |
| 374 | """ |
| 375 | AIHubMix 是 OpenAI-compatible 网关。这里用 fake OpenAI client |
| 376 | 验证独立 provider 会使用合作方默认地址和推荐模型，避免真实网络 |
| 377 | 或私有 API Key 影响测试稳定性。 |
| 378 | """ |
| 379 | config.app["llm_provider"] = "aihubmix" |
| 380 | config.app["aihubmix_api_key"] = "aihubmix-key" |
| 381 | config.app["aihubmix_base_url"] = "" |
| 382 | config.app["aihubmix_model_name"] = "" |
| 383 | |
| 384 | class FakeCompletions: |
| 385 | def create(self, **kwargs): |
| 386 | self.kwargs = kwargs |
| 387 | message = types.SimpleNamespace(content="hello\naihubmix") |
| 388 | choice = types.SimpleNamespace(message=message) |
| 389 | return types.SimpleNamespace(choices=[choice]) |
| 390 | |
| 391 | fake_completions = FakeCompletions() |
| 392 | fake_client = types.SimpleNamespace( |
| 393 | chat=types.SimpleNamespace(completions=fake_completions) |
| 394 | ) |
| 395 | |
| 396 | with ( |
| 397 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 398 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 399 | ): |
| 400 | result = llm._generate_response("Say hello") |
| 401 | |
| 402 | openai_client.assert_called_once_with( |
| 403 | api_key="aihubmix-key", |
| 404 | base_url="https://aihubmix.com/v1", |
| 405 | ) |
| 406 | self.assertEqual( |
| 407 | fake_completions.kwargs, |
| 408 | { |
| 409 | "model": "gpt-5.4-mini", |
| 410 | "messages": [{"role": "user", "content": "Say hello"}], |
| 411 | }, |
| 412 | ) |
| 413 | self.assertEqual(result, "helloaihubmix") |
| 414 | |
| 415 | def test_aimlapi_provider_uses_openai_compatible_client(self): |
| 416 | config.app["llm_provider"] = "aimlapi" |
| 417 | config.app["aimlapi_api_key"] = "aimlapi-key" |
| 418 | config.app["aimlapi_base_url"] = "" |
| 419 | config.app["aimlapi_model_name"] = "" |
| 420 | |
| 421 | class FakeCompletions: |
| 422 | def create(self, **kwargs): |
| 423 | self.kwargs = kwargs |
| 424 | message = types.SimpleNamespace(content="hello\naimlapi") |
| 425 | choice = types.SimpleNamespace(message=message) |
| 426 | return types.SimpleNamespace(choices=[choice]) |
| 427 | |
| 428 | fake_completions = FakeCompletions() |
| 429 | fake_client = types.SimpleNamespace( |
| 430 | chat=types.SimpleNamespace(completions=fake_completions) |
| 431 | ) |
| 432 | |
| 433 | with ( |
| 434 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 435 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 436 | ): |
| 437 | result = llm._generate_response("Say hello") |
| 438 | |
| 439 | openai_client.assert_called_once_with( |
| 440 | api_key="aimlapi-key", |
| 441 | base_url="https://api.aimlapi.com/v1", |
| 442 | ) |
| 443 | self.assertEqual( |
| 444 | fake_completions.kwargs, |
| 445 | { |
| 446 | "model": "openai/gpt-4o-mini", |
| 447 | "messages": [{"role": "user", "content": "Say hello"}], |
| 448 | }, |
| 449 | ) |
| 450 | self.assertEqual(result, "helloaimlapi") |
| 451 | |
| 452 | def test_evolink_provider_uses_openai_compatible_client(self): |
| 453 | """ |
| 454 | EvoLink exposes OpenAI-compatible Chat Completions at direct.evolink.ai. |
| 455 | The provider should keep its own default endpoint and model instead of |
| 456 | requiring users to overload the generic OpenAI settings. |
| 457 | """ |
| 458 | config.app["llm_provider"] = "evolink" |
| 459 | config.app["evolink_api_key"] = "evolink-key" |
| 460 | config.app["evolink_base_url"] = "" |
| 461 | config.app["evolink_model_name"] = "" |
| 462 | |
| 463 | class FakeCompletions: |
| 464 | def create(self, **kwargs): |
| 465 | self.kwargs = kwargs |
| 466 | message = types.SimpleNamespace(content="hello\nevolink") |
| 467 | choice = types.SimpleNamespace(message=message) |
| 468 | return types.SimpleNamespace(choices=[choice]) |
| 469 | |
| 470 | fake_completions = FakeCompletions() |
| 471 | fake_client = types.SimpleNamespace( |
| 472 | chat=types.SimpleNamespace(completions=fake_completions) |
| 473 | ) |
| 474 | |
| 475 | with ( |
| 476 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 477 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 478 | ): |
| 479 | result = llm._generate_response("Say hello") |
| 480 | |
| 481 | openai_client.assert_called_once_with( |
| 482 | api_key="evolink-key", |
| 483 | base_url="https://direct.evolink.ai/v1", |
| 484 | ) |
| 485 | self.assertEqual( |
| 486 | fake_completions.kwargs, |
| 487 | { |
| 488 | "model": "gpt-5.5", |
| 489 | "messages": [{"role": "user", "content": "Say hello"}], |
| 490 | }, |
| 491 | ) |
| 492 | self.assertEqual(result, "helloevolink") |
| 493 | |
| 494 | def test_volcengine_provider_uses_openai_compatible_client(self): |
| 495 | """ |
| 496 | VolcEngine Ark 暴露 OpenAI-compatible Chat Completions。 |
| 497 | 这里用 fake OpenAI client 覆盖 provider 默认地址和默认模型， |
| 498 | 避免真实网络或私有 API key 影响测试稳定性。 |
| 499 | """ |
| 500 | config.app["llm_provider"] = "volcengine" |
| 501 | config.app["volcengine_api_key"] = "volcengine-key" |
| 502 | config.app["volcengine_base_url"] = "" |
| 503 | config.app["volcengine_model_name"] = "" |
| 504 | |
| 505 | class FakeCompletions: |
| 506 | def create(self, **kwargs): |
| 507 | self.kwargs = kwargs |
| 508 | message = types.SimpleNamespace(content="hello\nvolcengine") |
| 509 | choice = types.SimpleNamespace(message=message) |
| 510 | return types.SimpleNamespace(choices=[choice]) |
| 511 | |
| 512 | fake_completions = FakeCompletions() |
| 513 | fake_client = types.SimpleNamespace( |
| 514 | chat=types.SimpleNamespace(completions=fake_completions) |
| 515 | ) |
| 516 | |
| 517 | with ( |
| 518 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 519 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 520 | ): |
| 521 | result = llm._generate_response("Say hello") |
| 522 | |
| 523 | openai_client.assert_called_once_with( |
| 524 | api_key="volcengine-key", |
| 525 | base_url="https://ark.cn-beijing.volces.com/api/v3", |
| 526 | ) |
| 527 | self.assertEqual( |
| 528 | fake_completions.kwargs, |
| 529 | { |
| 530 | "model": "doubao-seed-2-1-turbo-260628", |
| 531 | "messages": [{"role": "user", "content": "Say hello"}], |
| 532 | }, |
| 533 | ) |
| 534 | self.assertEqual(result, "hellovolcengine") |
| 535 | |
| 536 | def test_grok_provider_still_uses_existing_path(self): |
| 537 | config.app["llm_provider"] = "grok" |
| 538 | config.app["grok_api_key"] = "" |
| 539 | config.app["grok_base_url"] = "https://api.x.ai/v1" |
| 540 | config.app["grok_model_name"] = "grok-4.3" |
| 541 | |
| 542 | result = llm._generate_response("test") |
| 543 | |
| 544 | self.assertIn("Error:", result) |
| 545 | self.assertIn("api_key is not set", result) |
| 546 | self.assertNotIn("litellm", result.lower()) |
| 547 | |
| 548 | def test_groq_provider_requires_api_key(self): |
| 549 | config.app["llm_provider"] = "groq" |
| 550 | config.app["groq_api_key"] = "" |
| 551 | config.app["groq_base_url"] = "https://api.groq.com/openai/v1" |
| 552 | config.app["groq_model_name"] = "llama-3.3-70b-versatile" |
| 553 | |
| 554 | result = llm._generate_response("test") |
| 555 | |
| 556 | self.assertIn("Error:", result) |
| 557 | self.assertIn("api_key is not set", result) |
| 558 | self.assertNotIn("litellm", result.lower()) |
| 559 | |
| 560 | def test_groq_provider_uses_default_base_url(self): |
| 561 | config.app["llm_provider"] = "groq" |
| 562 | config.app["groq_api_key"] = "groq-test-key" |
| 563 | config.app["groq_base_url"] = "" |
| 564 | config.app["groq_model_name"] = "llama-3.3-70b-versatile" |
| 565 | |
| 566 | fake_response = types.SimpleNamespace( |
| 567 | choices=[ |
| 568 | types.SimpleNamespace( |
| 569 | message=types.SimpleNamespace(content="hello\ngroq") |
| 570 | ) |
| 571 | ] |
| 572 | ) |
| 573 | fake_client = types.SimpleNamespace( |
| 574 | chat=types.SimpleNamespace( |
| 575 | completions=types.SimpleNamespace(create=lambda **kwargs: fake_response) |
| 576 | ) |
| 577 | ) |
| 578 | |
| 579 | with ( |
| 580 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 581 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 582 | ): |
| 583 | result = llm._generate_response("Say hello") |
| 584 | |
| 585 | openai_client.assert_called_once_with( |
| 586 | api_key="groq-test-key", |
| 587 | base_url="https://api.groq.com/openai/v1", |
| 588 | ) |
| 589 | self.assertEqual(result, "hellogroq") |
| 590 | |
| 591 | def _use_ollama_provider(self, base_url=""): |
| 592 | config.app["llm_provider"] = "ollama" |
| 593 | config.app["ollama_api_key"] = "" |
| 594 | config.app["ollama_base_url"] = base_url |
| 595 | config.app["ollama_model_name"] = "llama3" |
| 596 | |
| 597 | def _assert_ollama_base_url(self, expected_base_url: str): |
| 598 | class FakeCompletions: |
| 599 | def create(self, **kwargs): |
| 600 | self.kwargs = kwargs |
| 601 | message = types.SimpleNamespace(content="hello\nollama") |
| 602 | choice = types.SimpleNamespace(message=message) |
| 603 | return types.SimpleNamespace(choices=[choice]) |
| 604 | |
| 605 | fake_completions = FakeCompletions() |
| 606 | fake_client = types.SimpleNamespace( |
| 607 | chat=types.SimpleNamespace(completions=fake_completions) |
| 608 | ) |
| 609 | |
| 610 | with ( |
| 611 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 612 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 613 | ): |
| 614 | result = llm._generate_response("Say hello") |
| 615 | |
| 616 | openai_client.assert_called_once_with( |
| 617 | api_key="ollama", |
| 618 | base_url=expected_base_url, |
| 619 | ) |
| 620 | self.assertEqual( |
| 621 | fake_completions.kwargs, |
| 622 | { |
| 623 | "model": "llama3", |
| 624 | "messages": [{"role": "user", "content": "Say hello"}], |
| 625 | }, |
| 626 | ) |
| 627 | self.assertEqual(result, "helloollama") |
| 628 | |
| 629 | def test_ollama_default_base_url_uses_localhost_outside_container(self): |
| 630 | """ |
| 631 | 普通本机运行时，Ollama 默认仍然使用 localhost，避免影响已有用户。 |
| 632 | """ |
| 633 | self._use_ollama_provider() |
| 634 | |
| 635 | with patch.object(config, "is_running_in_container", return_value=False): |
| 636 | self._assert_ollama_base_url("http://localhost:11434/v1") |
| 637 | |
| 638 | def test_ollama_default_base_url_uses_host_gateway_inside_container(self): |
| 639 | """ |
| 640 | 容器内运行时，localhost 指向容器自身；默认改为 host.docker.internal， |
| 641 | 方便 Docker Desktop 用户访问宿主机上的 Ollama。 |
| 642 | """ |
| 643 | self._use_ollama_provider() |
| 644 | |
| 645 | with ( |
| 646 | patch.object(config, "is_running_in_container", return_value=True), |
| 647 | patch.object(config, "_can_resolve_hostname", return_value=True), |
| 648 | ): |
| 649 | self._assert_ollama_base_url("http://host.docker.internal:11434/v1") |
| 650 | |
| 651 | def test_ollama_default_base_url_falls_back_to_container_gateway(self): |
| 652 | """ |
| 653 | 原生 Linux Docker 里不一定能解析 host.docker.internal。此时使用容器 |
| 654 | 默认网关作为兜底地址，比直接返回不可解析的 hostname 更稳。 |
| 655 | """ |
| 656 | self._use_ollama_provider() |
| 657 | |
| 658 | with ( |
| 659 | patch.object(config, "is_running_in_container", return_value=True), |
| 660 | patch.object(config, "_can_resolve_hostname", return_value=False), |
| 661 | patch.object(config, "get_container_default_gateway_ip", return_value="172.17.0.1"), |
| 662 | ): |
| 663 | self._assert_ollama_base_url("http://172.17.0.1:11434/v1") |
| 664 | |
| 665 | def test_ollama_explicit_base_url_takes_precedence(self): |
| 666 | """ |
| 667 | 用户手动配置的 ollama_base_url 优先级最高，不受容器检测影响。 |
| 668 | """ |
| 669 | self._use_ollama_provider(base_url="http://ollama:11434/v1") |
| 670 | |
| 671 | with patch.object(config, "is_running_in_container", return_value=True): |
| 672 | self._assert_ollama_base_url("http://ollama:11434/v1") |
| 673 | |
| 674 | def test_mimo_provider_uses_openai_compatible_client(self): |
| 675 | """ |
| 676 | MiMo 官方接口兼容 OpenAI Chat Completions 协议。这里用 fake OpenAI |
| 677 | client 验证 provider 会使用 MiMo 独立配置和默认 base_url，不依赖 |
| 678 | 真实网络或私有 API Key。 |
| 679 | """ |
| 680 | config.app["llm_provider"] = "mimo" |
| 681 | config.app["mimo_api_key"] = "mimo-key" |
| 682 | config.app["mimo_base_url"] = "" |
| 683 | config.app["mimo_model_name"] = "" |
| 684 | |
| 685 | class FakeCompletions: |
| 686 | def create(self, **kwargs): |
| 687 | self.kwargs = kwargs |
| 688 | message = types.SimpleNamespace(content="hello\nmimo") |
| 689 | choice = types.SimpleNamespace(message=message) |
| 690 | return types.SimpleNamespace(choices=[choice]) |
| 691 | |
| 692 | fake_completions = FakeCompletions() |
| 693 | fake_client = types.SimpleNamespace( |
| 694 | chat=types.SimpleNamespace(completions=fake_completions) |
| 695 | ) |
| 696 | |
| 697 | with ( |
| 698 | patch.object(llm, "OpenAI", return_value=fake_client) as openai_client, |
| 699 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 700 | ): |
| 701 | result = llm._generate_response("Say hello") |
| 702 | |
| 703 | openai_client.assert_called_once_with( |
| 704 | api_key="mimo-key", |
| 705 | base_url="https://api.xiaomimimo.com/v1", |
| 706 | ) |
| 707 | self.assertEqual( |
| 708 | fake_completions.kwargs, |
| 709 | { |
| 710 | "model": "mimo-v2.5-pro", |
| 711 | "messages": [{"role": "user", "content": "Say hello"}], |
| 712 | }, |
| 713 | ) |
| 714 | self.assertEqual(result, "hellomimo") |
| 715 | |
| 716 | def test_azure_provider_uses_azure_client_directly(self): |
| 717 | """ |
| 718 | Azure OpenAI 的鉴权、endpoint 和 api-version 都由 AzureOpenAI 客户端处理。 |
| 719 | 这个测试覆盖 issue #892：azure 分支必须直接调用 AzureOpenAI 创建的客户端， |
| 720 | 不能继续落入普通 OpenAI-compatible 分支，否则会丢失 Azure 专用请求配置。 |
| 721 | """ |
| 722 | config.app["llm_provider"] = "azure" |
| 723 | config.app["azure_api_key"] = "azure-key" |
| 724 | config.app["azure_base_url"] = "https://example.openai.azure.com" |
| 725 | config.app["azure_model_name"] = "gpt-4o-mini" |
| 726 | config.app["azure_api_version"] = "2024-02-15-preview" |
| 727 | |
| 728 | class FakeCompletions: |
| 729 | def create(self, **kwargs): |
| 730 | self.kwargs = kwargs |
| 731 | message = types.SimpleNamespace(content="hello\nazure") |
| 732 | choice = types.SimpleNamespace(message=message) |
| 733 | return types.SimpleNamespace(choices=[choice]) |
| 734 | |
| 735 | fake_completions = FakeCompletions() |
| 736 | fake_client = types.SimpleNamespace( |
| 737 | chat=types.SimpleNamespace(completions=fake_completions) |
| 738 | ) |
| 739 | |
| 740 | with ( |
| 741 | patch.object(llm, "AzureOpenAI", return_value=fake_client) as azure_client, |
| 742 | patch.object(llm, "OpenAI") as openai_client, |
| 743 | patch.object(llm, "ChatCompletion", types.SimpleNamespace), |
| 744 | ): |
| 745 | result = llm._generate_response("Say hello") |
| 746 | |
| 747 | azure_client.assert_called_once_with( |
| 748 | api_key="azure-key", |
| 749 | api_version="2024-02-15-preview", |
| 750 | azure_endpoint="https://example.openai.azure.com", |
| 751 | ) |
| 752 | openai_client.assert_not_called() |
| 753 | self.assertEqual( |
| 754 | fake_completions.kwargs, |
| 755 | { |
| 756 | "model": "gpt-4o-mini", |
| 757 | "messages": [{"role": "user", "content": "Say hello"}], |
| 758 | }, |
| 759 | ) |
| 760 | self.assertEqual(result, "helloazure") |
| 761 | |
| 762 | def test_g4f_provider_requires_explicit_opt_in(self): |
| 763 | """ |
| 764 | g4f 存在供应链和稳定性风险，不能因为用户把 provider 写成 g4f |
| 765 | 就默认加载第三方包并访问逆向接口，必须显式启用。 |
| 766 | """ |
| 767 | config.app["llm_provider"] = "g4f" |
| 768 | config.app["enable_g4f"] = False |
| 769 | |
| 770 | result = llm._generate_response("test") |
| 771 | |
| 772 | self.assertIn("Error:", result) |
| 773 | self.assertIn("g4f provider is disabled", result) |
| 774 | |
| 775 | def test_g4f_provider_uses_lazy_import_after_opt_in(self): |
| 776 | config.app["llm_provider"] = "g4f" |
| 777 | config.app["enable_g4f"] = True |
| 778 | config.app["g4f_model_name"] = "gpt-3.5-turbo" |
| 779 | |
| 780 | fake_g4f = types.SimpleNamespace() |
| 781 | fake_g4f.ChatCompletion = types.SimpleNamespace( |
| 782 | create=lambda **kwargs: "hello from g4f" |
| 783 | ) |
| 784 | |
| 785 | with patch.dict(sys.modules, {"g4f": fake_g4f}): |
| 786 | result = llm._generate_response("test") |
| 787 | |
| 788 | self.assertEqual(result, "hello from g4f") |
| 789 | |
| 790 | def test_g4f_provider_reports_missing_optional_dependency(self): |
| 791 | config.app["llm_provider"] = "g4f" |
| 792 | config.app["enable_g4f"] = True |
| 793 | config.app["g4f_model_name"] = "gpt-3.5-turbo" |
| 794 | |
| 795 | with patch.dict(sys.modules, {"g4f": None}): |
| 796 | result = llm._generate_response("test") |
| 797 | |
| 798 | self.assertIn("Error:", result) |
| 799 | self.assertIn("g4f package is not installed by default", result) |
| 800 | |
| 801 | |
| 802 | class TestRuntimeEnvironmentDetection(unittest.TestCase): |
| 803 | def test_container_detection_ignores_plain_linux_cgroup_file(self): |
| 804 | """ |
| 805 | 普通 Linux 也有 /proc/1/cgroup，不能因为文件存在就判定为容器。 |
| 806 | """ |
| 807 | with tempfile.TemporaryDirectory() as tmp_dir: |
| 808 | cgroup_path = Path(tmp_dir) / "cgroup" |
| 809 | cgroup_path.write_text("0::/init.scope\n", encoding="utf-8") |
| 810 | |
| 811 | self.assertFalse( |
| 812 | config.is_running_in_container( |
| 813 | dockerenv_path=str(Path(tmp_dir) / "missing-dockerenv"), |
| 814 | containerenv_path=str(Path(tmp_dir) / "missing-containerenv"), |
| 815 | cgroup_path=str(cgroup_path), |
| 816 | ) |
| 817 | ) |
| 818 | |
| 819 | def test_container_detection_accepts_dockerenv_marker(self): |
| 820 | with tempfile.TemporaryDirectory() as tmp_dir: |
| 821 | dockerenv_path = Path(tmp_dir) / ".dockerenv" |
| 822 | dockerenv_path.write_text("", encoding="utf-8") |
| 823 | |
| 824 | self.assertTrue( |
| 825 | config.is_running_in_container( |
| 826 | dockerenv_path=str(dockerenv_path), |
| 827 | containerenv_path=str(Path(tmp_dir) / "missing-containerenv"), |
| 828 | cgroup_path=str(Path(tmp_dir) / "missing-cgroup"), |
| 829 | ) |
| 830 | ) |
| 831 | |
| 832 | def test_container_detection_accepts_cgroup_container_marker(self): |
| 833 | with tempfile.TemporaryDirectory() as tmp_dir: |
| 834 | cgroup_path = Path(tmp_dir) / "cgroup" |
| 835 | cgroup_path.write_text( |
| 836 | "0::/system.slice/docker-abcdef.scope\n", |
| 837 | encoding="utf-8", |
| 838 | ) |
| 839 | |
| 840 | self.assertTrue( |
| 841 | config.is_running_in_container( |
| 842 | dockerenv_path=str(Path(tmp_dir) / "missing-dockerenv"), |
| 843 | containerenv_path=str(Path(tmp_dir) / "missing-containerenv"), |
| 844 | cgroup_path=str(cgroup_path), |
| 845 | ) |
| 846 | ) |
| 847 | |
| 848 | def test_container_gateway_ip_decodes_default_route(self): |
| 849 | with tempfile.TemporaryDirectory() as tmp_dir: |
| 850 | route_path = Path(tmp_dir) / "route" |
| 851 | route_path.write_text( |
| 852 | "Iface\tDestination\tGateway\tFlags\tRefCnt\tUse\tMetric\tMask\tMTU\tWindow\tIRTT\n" |
| 853 | "eth0\t00000000\t010011AC\t0003\t0\t0\t0\t00000000\t0\t0\t0\n", |
| 854 | encoding="utf-8", |
| 855 | ) |
| 856 | |
| 857 | self.assertEqual( |
| 858 | config.get_container_default_gateway_ip(str(route_path)), |
| 859 | "172.17.0.1", |
| 860 | ) |
| 861 | |
| 862 | def test_container_gateway_ip_ignores_missing_default_route(self): |
| 863 | with tempfile.TemporaryDirectory() as tmp_dir: |
| 864 | route_path = Path(tmp_dir) / "route" |
| 865 | route_path.write_text( |
| 866 | "Iface\tDestination\tGateway\tFlags\tRefCnt\tUse\tMetric\tMask\tMTU\tWindow\tIRTT\n" |
| 867 | "eth0\t0011AC0A\t00000000\t0001\t0\t0\t0\t00FFFFFF\t0\t0\t0\n", |
| 868 | encoding="utf-8", |
| 869 | ) |
| 870 | |
| 871 | self.assertEqual(config.get_container_default_gateway_ip(str(route_path)), "") |
| 872 | |
| 873 | |
| 874 | class TestSocialMetadata(unittest.TestCase): |
| 875 | """通用短视频发布文案元数据生成。""" |
| 876 | |
| 877 | def test_build_prompt_auto_language_uses_source_language(self): |
| 878 | """ |
| 879 | language 默认 auto 时，不应该固定成某个国家或语种，而是让模型 |
| 880 | 跟随视频主题和脚本的语言，扩大 API 适用范围。 |
| 881 | """ |
| 882 | prompt = llm.build_social_metadata_prompt( |
| 883 | video_subject="上海一日游", |
| 884 | video_script="今天带你快速看完上海经典路线。", |
| 885 | language="auto", |
| 886 | platform="tiktok", |
| 887 | ) |
| 888 | |
| 889 | self.assertIn("TikTok", prompt) |
| 890 | self.assertIn("Use the same language as the video subject and script", prompt) |
| 891 | self.assertIn("上海一日游", prompt) |
| 892 | self.assertIn("array of exactly 5 strings", prompt) |
| 893 | |
| 894 | def test_build_prompt_accepts_explicit_language(self): |
| 895 | prompt = llm.build_social_metadata_prompt( |
| 896 | video_subject="Coffee tips", |
| 897 | language="en-US", |
| 898 | platform="youtube_shorts", |
| 899 | ) |
| 900 | |
| 901 | self.assertIn("YouTube Shorts", prompt) |
| 902 | self.assertIn('Write "title" and "caption" in this language: en-US', prompt) |
| 903 | self.assertIn("array of exactly 3 strings", prompt) |
| 904 | |
| 905 | def test_unknown_platform_falls_back_to_tiktok(self): |
| 906 | prompt = llm.build_social_metadata_prompt( |
| 907 | video_subject="x", |
| 908 | platform="unsupported-platform", |
| 909 | ) |
| 910 | |
| 911 | self.assertIn("TikTok", prompt) |
| 912 | |
| 913 | def test_normalize_hashtags_from_string_dedupes_and_clamps(self): |
| 914 | tags = llm._normalize_hashtags("#fyp fyp, trending #Trending viral", count=2) |
| 915 | |
| 916 | self.assertEqual(tags, ["#fyp", "#trending"]) |
| 917 | |
| 918 | def test_normalize_hashtags_from_list_keeps_unicode_letters(self): |
| 919 | tags = llm._normalize_hashtags( |
| 920 | ["上海 旅行", "#việt nam", " ", "@bad!chars"], count=5 |
| 921 | ) |
| 922 | |
| 923 | self.assertEqual(tags, ["#上海旅行", "#việtnam", "#badchars"]) |
| 924 | |
| 925 | def test_parse_social_metadata_recovers_embedded_json(self): |
| 926 | raw = 'Sure: {"title":"T","caption":"C","hashtags":["#x"]} thanks' |
| 927 | result = llm._parse_social_metadata(raw, "tiktok") |
| 928 | |
| 929 | self.assertEqual(result["title"], "T") |
| 930 | self.assertEqual(result["caption"], "C") |
| 931 | self.assertEqual(result["hashtags"], ["#x"]) |
| 932 | |
| 933 | def test_parse_social_metadata_requires_title_or_caption(self): |
| 934 | with self.assertRaises(ValueError): |
| 935 | llm._parse_social_metadata('{"hashtags":["#x"]}', "tiktok") |
| 936 | |
| 937 | def test_generate_social_metadata_uses_llm_response(self): |
| 938 | payload = ( |
| 939 | '{"title":"上海一日游","caption":"收藏这条路线，下次直接出发！",' |
| 940 | '"hashtags":["#上海","#旅行","#shorts"]}' |
| 941 | ) |
| 942 | with patch.object(llm, "_generate_response", return_value=payload): |
| 943 | result = llm.generate_social_metadata( |
| 944 | video_subject="上海一日游", |
| 945 | video_script="今天带你快速看完上海经典路线。", |
| 946 | language="zh-CN", |
| 947 | platform="tiktok", |
| 948 | ) |
| 949 | |
| 950 | self.assertEqual(result["title"], "上海一日游") |
| 951 | self.assertEqual(result["caption"], "收藏这条路线，下次直接出发！") |
| 952 | self.assertEqual(result["hashtags"], ["#上海", "#旅行", "#shorts"]) |
| 953 | |
| 954 | def test_generate_social_metadata_falls_back_to_generic_hashtags(self): |
| 955 | with patch.object( |
| 956 | llm, "_generate_response", return_value="Error: api_key is not set" |
| 957 | ): |
| 958 | result = llm.generate_social_metadata( |
| 959 | video_subject="Coffee tips", |
| 960 | video_script="Save these three coffee tips.", |
| 961 | platform="instagram_reels", |
| 962 | ) |
| 963 | |
| 964 | self.assertEqual(result["title"], "Coffee tips") |
| 965 | self.assertEqual(result["caption"], "Save these three coffee tips.") |
| 966 | self.assertEqual(len(result["hashtags"]), 8) |
| 967 | self.assertEqual(result["hashtags"][0], "#shorts") |
| 968 | |
| 969 | def test_request_model_defaults_to_auto_language_tiktok(self): |
| 970 | body = VideoSocialMetadataRequest(video_subject="Test") |
| 971 | |
| 972 | self.assertEqual(body.language, "auto") |
| 973 | self.assertEqual(body.platform, "tiktok") |
| 974 | |
| 975 | def test_request_model_rejects_oversized_social_metadata_fields(self): |
| 976 | """ |
| 977 | 外部 API 不能接受无限长的脚本和语言参数，否则会直接放大 LLM |
| 978 | token 成本。schema 层先拦截，服务层再做内部调用兜底。 |
| 979 | """ |
| 980 | with self.assertRaises(ValidationError): |
| 981 | VideoSocialMetadataRequest(video_subject="x" * 501) |
| 982 | |
| 983 | with self.assertRaises(ValidationError): |
| 984 | VideoSocialMetadataRequest(video_subject="x", video_script="x" * 8001) |
| 985 | |
| 986 | with self.assertRaises(ValidationError): |
| 987 | VideoSocialMetadataRequest(video_subject="x", language="x" * 65) |
| 988 | |
| 989 | def test_build_prompt_clamps_direct_service_inputs(self): |
| 990 | prompt = llm.build_social_metadata_prompt( |
| 991 | video_subject="x" * 600, |
| 992 | video_script="y" * 9000, |
| 993 | language="en", |
| 994 | ) |
| 995 | |
| 996 | self.assertIn("x" * llm.MAX_SOCIAL_SUBJECT_LENGTH, prompt) |
| 997 | self.assertNotIn("x" * (llm.MAX_SOCIAL_SUBJECT_LENGTH + 1), prompt) |
| 998 | self.assertIn("y" * llm.MAX_SOCIAL_SCRIPT_LENGTH, prompt) |
| 999 | self.assertNotIn("y" * (llm.MAX_SOCIAL_SCRIPT_LENGTH + 1), prompt) |
| 1000 | |
| 1001 | def test_social_metadata_endpoint_response_shape(self): |
| 1002 | from fastapi.testclient import TestClient |
| 1003 | |
| 1004 | from app.asgi import app |
| 1005 | |
| 1006 | request_body = { |
| 1007 | "video_subject": "Tokyo coffee shops", |
| 1008 | "video_script": "Three quiet coffee shops for your next Tokyo morning.", |
| 1009 | "language": "en", |
| 1010 | "platform": "youtube_shorts", |
| 1011 | } |
| 1012 | llm_response = ( |
| 1013 | '{"title":"3 Quiet Tokyo Coffee Shops",' |
| 1014 | '"caption":"Save these spots for your next Tokyo morning.",' |
| 1015 | '"hashtags":["#Tokyo","#Coffee","#Shorts"]}' |
| 1016 | ) |
| 1017 | |
| 1018 | with patch.object(llm, "_generate_response", return_value=llm_response): |
| 1019 | response = TestClient(app).post( |
| 1020 | "/api/v1/social-metadata", |
| 1021 | json=request_body, |
| 1022 | ) |
| 1023 | |
| 1024 | self.assertEqual(response.status_code, 200) |
| 1025 | self.assertEqual( |
| 1026 | response.json(), |
| 1027 | { |
| 1028 | "status": 200, |
| 1029 | "message": "success", |
| 1030 | "data": { |
| 1031 | "title": "3 Quiet Tokyo Coffee Shops", |
| 1032 | "caption": "Save these spots for your next Tokyo morning.", |
| 1033 | "hashtags": ["#Tokyo", "#Coffee", "#Shorts"], |
| 1034 | }, |
| 1035 | }, |
| 1036 | ) |
| 1037 | |
| 1038 | |
| 1039 | FOUNDRY_KEY = os.environ.get("ANTHROPIC_FOUNDRY_API_KEY", "") |
| 1040 | FOUNDRY_BASE = "https://amanrai-test-resource.services.ai.azure.com/anthropic" |
| 1041 | FOUNDRY_MODEL = "azure_ai/claude-sonnet-4-6" |
| 1042 | |
| 1043 | |
| 1044 | @unittest.skipUnless( |
| 1045 | RUN_INTEGRATION_TESTS and FOUNDRY_KEY, |
| 1046 | "MPT_RUN_INTEGRATION_TESTS and ANTHROPIC_FOUNDRY_API_KEY not set", |
| 1047 | ) |
| 1048 | class TestLiteLLMLiveIntegration(unittest.TestCase): |
| 1049 | def setUp(self): |
| 1050 | self.original_app_config = dict(config.app) |
| 1051 | config.app["llm_provider"] = "litellm" |
| 1052 | config.app["litellm_model_name"] = FOUNDRY_MODEL |
| 1053 | os.environ["AZURE_AI_API_KEY"] = FOUNDRY_KEY |
| 1054 | os.environ["AZURE_AI_API_BASE"] = FOUNDRY_BASE |
| 1055 | |
| 1056 | def tearDown(self): |
| 1057 | config.app.clear() |
| 1058 | config.app.update(self.original_app_config) |
| 1059 | |
| 1060 | def test_live_litellm_completion(self): |
| 1061 | result = llm._generate_response("What is 2+2? Reply with just the number.") |
| 1062 | |
| 1063 | self.assertNotIn("Error:", result) |
| 1064 | self.assertIn("4", result) |
| 1065 | |
| 1066 | |
| 1067 | if __name__ == "__main__": |
| 1068 | unittest.main() |
| 1069 |