Create-knowledge-find-decode/tests/test_media_service.py

from __future__ import annotations

from acquisition.media.service import stabilize_media_urls


def test_stabilize_media_urls_keeps_order_and_positions():
    calls = []

    def uploader(url: str, media_type: str) -> str:
        calls.append((url, media_type))
        return f"https://cdn.test/{media_type}/{len(calls)}"

    rows = stabilize_media_urls(
        image_urls=["https://src.test/a.jpg", "", "https://src.test/b.jpg"],
        video_urls=["https://src.test/c.mp4"],
        uploader=uploader,
    )

    assert [(row.media_type, row.position) for row in rows] == [
        ("image", 1),
        ("image", 2),
        ("video", 3),
    ]
    assert rows[0].source_url == "https://src.test/a.jpg"
    assert rows[0].cdn_url == "https://cdn.test/image/1"
    assert rows[-1].cdn_url == "https://cdn.test/video/3"
    assert calls == [
        ("https://src.test/a.jpg", "image"),
        ("https://src.test/b.jpg", "image"),
        ("https://src.test/c.mp4", "video"),
    ]


def test_stabilize_media_urls_falls_back_to_source_on_upload_error():
    def uploader(url: str, media_type: str) -> str:
        raise RuntimeError(f"{media_type} upload failed")

    rows = stabilize_media_urls(
        image_urls=["https://src.test/a.jpg"],
        video_urls=["https://src.test/c.mp4"],
        uploader=uploader,
    )

    assert [row.cdn_url for row in rows] == [
        "https://src.test/a.jpg",
        "https://src.test/c.mp4",
    ]
    assert [row.status for row in rows] == ["ready", "ready"]


def test_stabilize_video_retries_then_uses_cdn():
    calls = []
    sleeps = []

    def uploader(url: str, media_type: str) -> str:
        calls.append((url, media_type))
        if len(calls) < 3:
            raise RuntimeError("temporary oss failure")
        return "https://cdn.test/video.mp4"

    rows = stabilize_media_urls(
        video_urls=["https://src.test/c.mp4"],
        uploader=uploader,
        video_fallback_to_source=False,
        video_retry_delays_seconds=(5, 10),
        sleep=sleeps.append,
    )

    assert len(calls) == 3
    assert sleeps == [5, 10]
    assert rows[0].cdn_url == "https://cdn.test/video.mp4"
    assert rows[0].status == "ready"


def test_stabilize_video_strict_mode_does_not_fallback_to_source():
    calls = []
    sleeps = []

    def uploader(url: str, media_type: str) -> str:
        calls.append((url, media_type))
        return url

    rows = stabilize_media_urls(
        video_urls=["https://src.test/c.mp4"],
        uploader=uploader,
        video_fallback_to_source=False,
        video_retry_delays_seconds=(5, 10),
        sleep=sleeps.append,
    )

    assert len(calls) == 3
    assert sleeps == [5, 10]
    assert rows[0].source_url == "https://src.test/c.mp4"
    assert rows[0].cdn_url == ""
    assert rows[0].status == "failed"
    assert rows[0].error_message == "oss_upload_returned_source_url"