tests/test_dialogue_engine.py

from __future__ import annotations

import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from src import dialogue_engine


class DialogueEngineMultimodalTests(unittest.TestCase):
    def test_build_user_content_adds_image_url(self):
        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as handle:
            handle.write(b"\x89PNG\r\n\x1a\n")
            image_path = Path(handle.name)

        try:
            content = dialogue_engine._build_user_content(
                "看一下这张图",
                {"images": [{"path": str(image_path)}]},
            )
        finally:
            image_path.unlink(missing_ok=True)

        self.assertIsInstance(content, list)
        self.assertEqual(content[0], {"type": "text", "text": "看一下这张图"})
        self.assertEqual(content[1]["type"], "image_url")
        self.assertTrue(content[1]["image_url"]["url"].startswith("data:image/png;base64,"))

    def test_build_user_content_ignores_audio_attachment(self):
        content = dialogue_engine._build_user_content(
            "继续文字对话",
            {"images": [], "audio": [{"path": "ignored.wav"}]},
        )

        self.assertEqual(content, [{"type": "text", "text": "继续文字对话"}])

    def test_vllm_failure_returns_regular_error_without_audio_fallback(self):
        class FailingClient:
            def __init__(self, *args, **kwargs):
                pass

            def __enter__(self):
                return self

            def __exit__(self, *args):
                return False

            def stream(self, *args, **kwargs):
                raise RuntimeError("400 Bad Request")

        with patch.dict("os.environ", {"VC_VLLM_RETRIES": "1"}, clear=False):
            with patch("httpx.Client", FailingClient):
                events = list(
                    dialogue_engine._stream_vllm_reply(
                        "https://example.test",
                        "你好",
                        [],
                        {"character": {"voice": {}}},
                        {"images": []},
                        {"enabled": False},
                    )
                )

        errors = [event["message"] for event in events if event.get("type") == "error"]
        self.assertTrue(any("Modal vLLM 调用失败" in message for message in errors))


if __name__ == "__main__":
    unittest.main()