예제 소스

test_hf_img_integration.py

"""
Hugging Face Router 실제 호출 (네트워크 + 유효 토큰 필요).

기본은 건너뜁니다. 실행 예:

  HF_INTEGRATION_TEST=1 .venv/bin/python -m unittest tests.test_hf_img_integration -v

모델: 환경변수 HF_VISION_MODEL (없으면 test_hf_img.DEFAULT_VISION_MODEL).
토큰: HF_TOKEN 우선, 없으면 private.hf_key.HG_TOKEN (프로젝트와 동일 규칙).
"""
from __future__ import annotations

import os
import unittest
from io import BytesIO


def _integration_hf_token() -> str:
    t = os.environ.get("HF_TOKEN", "").strip()
    if t:
        return t
    try:
        from private.hf_key import HG_TOKEN

        return HG_TOKEN.strip() if isinstance(HG_TOKEN, str) else str(HG_TOKEN)
    except ImportError:
        return ""


def _tiny_jpeg_bytes() -> bytes:
    from PIL import Image

    buf = BytesIO()
    Image.new("RGB", (64, 64), color=(210, 30, 40)).save(buf, format="JPEG", quality=88)
    return buf.getvalue()


@unittest.skipUnless(
    os.environ.get("HF_INTEGRATION_TEST", "").strip() == "1",
    "실 Router 호출은 HF_INTEGRATION_TEST=1 일 때만 (토큰·네트워크 필요)",
)
class TestHfRouterVisionLive(unittest.TestCase):
    """이미지(data URL) + 텍스트를 Router chat/completions 로 보내 응답 문자열을 받는다."""

    @classmethod
    def setUpClass(cls) -> None:
        cls.token = _integration_hf_token()
        if not cls.token:
            raise unittest.SkipTest("HF_TOKEN 또는 private.hf_key.HG_TOKEN 필요")

        import test_hf_img as hf

        cls.hf = hf
        cls.model = os.environ.get("HF_VISION_MODEL", "").strip() or hf.DEFAULT_VISION_MODEL

    def test_live_image_plus_text_returns_non_empty_reply(self) -> None:
        img = _tiny_jpeg_bytes()
        prompt = (
            "This image is mostly one solid color. "
            "Answer in one short English phrase naming the dominant color (e.g. red)."
        )
        try:
            reply = self.hf._hf_vision_chat(
                token=self.token,
                image_bytes=img,
                user_text=prompt,
                model=self.model,
                timeout_sec=int(os.environ.get("HF_INTEGRATION_TIMEOUT", "180")),
            )
        except RuntimeError as e:
            err = str(e)
            if "HTTP 400" in err and (
                "model_not_supported" in err or "not a chat model" in err.lower()
            ):
                self.skipTest(
                    "Router가 모델/Provider를 거절했습니다. HF_VISION_MODEL 또는 "
                    "https://huggingface.co/playground 에서 본인 계정에 노출되는 "
                    f"chat 비전 모델로 바꿔 다시 실행하세요.\n{err[:500]}"
                )
            raise
        self.assertIsInstance(reply, str)
        self.assertGreaterEqual(len(reply.strip()), 2, msg=f"빈 응답: {reply!r}")

    def test_live_korean_opinion_style_prompt(self) -> None:
        """차트 대신 단색 이미지이므로 '유지' 등이 나와도 파서만 검증."""
        img = _tiny_jpeg_bytes()
        prompt = self.hf._build_analysis_prompt("TEST-TICKER")
        try:
            reply = self.hf._hf_vision_chat(
                token=self.token,
                image_bytes=img,
                user_text=prompt,
                model=self.model,
                timeout_sec=int(os.environ.get("HF_INTEGRATION_TIMEOUT", "180")),
            )
        except RuntimeError as e:
            err = str(e)
            if "HTTP 400" in err and (
                "model_not_supported" in err or "not a chat model" in err.lower()
            ):
                self.skipTest(
                    "모델/Provider 거절 — test_live_image_plus_text 와 동일하게 "
                    "HF_VISION_MODEL 을 Playground에서 확인 가능한 ID로 지정하세요.\n"
                    f"{err[:500]}"
                )
            raise
        opinion, _ = self.hf._parse_opinion(reply)
        self.assertIsInstance(reply, str)
        self.assertGreater(len(reply.strip()), 5)
        # 단색 이미지라 의견이 비어 있을 수 있음 — 파서는 예외 없이 동작하면 됨
        if opinion is not None:
            self.assertIn(opinion, ("매수", "매도", "유지"))


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