52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import os
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
from app.pipeline import background, storage
|
|
|
|
|
|
@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_fetch_video_loop_success(tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("PEXELS_API_KEY", "k")
|
|
video_url = "https://videos.pexels.com/video-files/123/sample.mp4"
|
|
respx.get("https://api.pexels.com/videos/search").mock(
|
|
return_value=Response(200, json={
|
|
"videos": [{
|
|
"id": 123, "duration": 10,
|
|
"video_files": [
|
|
{"quality": "hd", "width": 1920, "link": video_url},
|
|
],
|
|
}],
|
|
})
|
|
)
|
|
respx.get(video_url).mock(return_value=Response(200, content=b"\x00" * 4096))
|
|
|
|
result = await background.fetch_video_loop(pipeline_id=10, keyword="rainy window")
|
|
assert result["used_fallback"] is False
|
|
assert (tmp_storage / "10" / "loop.mp4").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_video_loop_no_api_key(tmp_storage, monkeypatch):
|
|
monkeypatch.delenv("PEXELS_API_KEY", raising=False)
|
|
result = await background.fetch_video_loop(pipeline_id=11, keyword="rain")
|
|
assert result["used_fallback"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_fetch_video_loop_zero_results(tmp_storage, monkeypatch):
|
|
monkeypatch.setenv("PEXELS_API_KEY", "k")
|
|
respx.get("https://api.pexels.com/videos/search").mock(
|
|
return_value=Response(200, json={"videos": []})
|
|
)
|
|
result = await background.fetch_video_loop(pipeline_id=12, keyword="impossible-keyword")
|
|
assert result["used_fallback"] is True
|