| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748 |
- 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"]
|
