94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
import base64
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
from app.pipeline import cover, storage
|
|
|
|
|
|
# Real PNG bytes (1x1 red pixel) so PIL can open
|
|
_TINY_PNG = bytes.fromhex(
|
|
"89504e470d0a1a0a0000000d49484452000000010000000108020000009077"
|
|
"53de0000000c4944415478da6300010000050001"
|
|
"0d0a2db40000000049454e44ae426082"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_storage(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(storage, "VIDEO_DATA_DIR", str(tmp_path))
|
|
return tmp_path
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_dalle_success_saves_jpg(tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
|
|
image_url = "https://oaidalleapiprodscus.blob.core.windows.net/x.png"
|
|
respx.post("https://api.openai.com/v1/images/generations").mock(
|
|
return_value=Response(200, json={"data": [{"url": image_url}]})
|
|
)
|
|
respx.get(image_url).mock(return_value=Response(200, content=_TINY_PNG))
|
|
|
|
out = await cover.generate(pipeline_id=42, genre="lo-fi",
|
|
prompt_template="moody anime", mood="chill",
|
|
track_title="Test")
|
|
assert out["used_fallback"] is False
|
|
assert out["url"].startswith("/media/videos/42/cover")
|
|
assert (tmp_storage / "42" / "cover.jpg").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_dalle_http_error_falls_back_to_gradient(tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
|
|
respx.post("https://api.openai.com/v1/images/generations").mock(
|
|
return_value=Response(504)
|
|
)
|
|
out = await cover.generate(pipeline_id=43, genre="phonk",
|
|
prompt_template="dark drift", mood="aggressive",
|
|
track_title="Midnight Drive")
|
|
assert out["used_fallback"] is True
|
|
assert (tmp_storage / "43" / "cover.jpg").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_api_key_falls_back(tmp_storage, monkeypatch):
|
|
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
|
out = await cover.generate(pipeline_id=44, genre="ambient",
|
|
prompt_template="x", mood="calm",
|
|
track_title="Calm")
|
|
assert out["used_fallback"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_dalle_with_feedback_appends_to_prompt(tmp_storage, monkeypatch):
|
|
import json as _json
|
|
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
|
|
captured = {}
|
|
def hook(req):
|
|
captured["body"] = _json.loads(req.content)
|
|
return Response(200, json={"data": [{"url": "https://x"}]})
|
|
respx.post("https://api.openai.com/v1/images/generations").mock(side_effect=hook)
|
|
respx.get("https://x").mock(return_value=Response(200, content=_TINY_PNG))
|
|
out = await cover.generate(pipeline_id=45, genre="lo-fi",
|
|
prompt_template="moody anime", mood="chill",
|
|
track_title="X", feedback="더 어둡게")
|
|
assert "더 어둡게" in captured["body"]["prompt"]
|
|
assert out["used_fallback"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_dalle_b64_response_handled(tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
|
|
b64 = base64.b64encode(_TINY_PNG).decode()
|
|
respx.post("https://api.openai.com/v1/images/generations").mock(
|
|
return_value=Response(200, json={"data": [{"b64_json": b64}]})
|
|
)
|
|
out = await cover.generate(pipeline_id=46, genre="lo-fi",
|
|
prompt_template="x", mood="", track_title="X")
|
|
assert out["used_fallback"] is False
|
|
assert (tmp_storage / "46" / "cover.jpg").exists()
|