refactor(gateway): shared media-cache mime dispatch for adapter downloads (per-adapter overrides preserve historical mappings)
This commit is contained in:
teknium1
Teknium
parent
7b6a67f82e
commit
bf15259e33
|
|
@ -31,8 +31,31 @@ from gateway.platforms.base import (
|
|||
cache_audio_from_bytes,
|
||||
cache_document_from_bytes,
|
||||
)
|
||||
from .media_cache import ext_for_mime
|
||||
from gateway.platforms.helpers import strip_markdown
|
||||
|
||||
# Historical BlueBubbles mime→ext maps, preserved verbatim as overrides for
|
||||
# the shared dispatch in gateway.platforms.media_cache. Both maps are
|
||||
# CLOSED: unlisted mimes fall back to .jpg / .mp3 (never mimetypes).
|
||||
_BLUEBUBBLES_IMAGE_EXT_OVERRIDES = {
|
||||
"image/jpeg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
"image/heic": ".jpg", # preserves historical bluebubbles mapping
|
||||
"image/heif": ".jpg", # preserves historical bluebubbles mapping
|
||||
"image/tiff": ".jpg", # preserves historical bluebubbles mapping
|
||||
}
|
||||
_BLUEBUBBLES_AUDIO_EXT_OVERRIDES = {
|
||||
"audio/mp3": ".mp3",
|
||||
"audio/mpeg": ".mp3",
|
||||
"audio/ogg": ".ogg",
|
||||
"audio/wav": ".wav",
|
||||
"audio/x-caf": ".mp3", # preserves historical bluebubbles mapping
|
||||
"audio/mp4": ".m4a",
|
||||
"audio/aac": ".m4a", # preserves historical bluebubbles mapping (shared table says .aac)
|
||||
}
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -808,29 +831,27 @@ class BlueBubblesAdapter(BasePlatformAdapter):
|
|||
transfer_name = att_meta.get("transferName", "")
|
||||
|
||||
if mime.startswith("image/"):
|
||||
ext_map = {
|
||||
"image/jpeg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
"image/heic": ".jpg",
|
||||
"image/heif": ".jpg",
|
||||
"image/tiff": ".jpg",
|
||||
}
|
||||
ext = ext_map.get(mime, ".jpg")
|
||||
ext = ext_for_mime(
|
||||
mime,
|
||||
overrides=_BLUEBUBBLES_IMAGE_EXT_OVERRIDES,
|
||||
# Historical map was closed: any unlisted image mime
|
||||
# fell back to .jpg without consulting mimetypes.
|
||||
use_defaults=False,
|
||||
use_mimetypes=False,
|
||||
fallback=".jpg",
|
||||
) or ".jpg"
|
||||
return cache_image_from_bytes(data, ext)
|
||||
|
||||
if mime.startswith("audio/"):
|
||||
ext_map = {
|
||||
"audio/mp3": ".mp3",
|
||||
"audio/mpeg": ".mp3",
|
||||
"audio/ogg": ".ogg",
|
||||
"audio/wav": ".wav",
|
||||
"audio/x-caf": ".mp3",
|
||||
"audio/mp4": ".m4a",
|
||||
"audio/aac": ".m4a",
|
||||
}
|
||||
ext = ext_map.get(mime, ".mp3")
|
||||
ext = ext_for_mime(
|
||||
mime,
|
||||
overrides=_BLUEBUBBLES_AUDIO_EXT_OVERRIDES,
|
||||
# Historical map was closed: any unlisted audio mime
|
||||
# fell back to .mp3 without consulting mimetypes.
|
||||
use_defaults=False,
|
||||
use_mimetypes=False,
|
||||
fallback=".mp3",
|
||||
) or ".mp3"
|
||||
return cache_audio_from_bytes(data, ext)
|
||||
|
||||
# Videos, documents, and everything else
|
||||
|
|
|
|||
|
|
@ -0,0 +1,202 @@
|
|||
"""Shared mime↔extension dispatch for inbound (downloaded) platform media.
|
||||
|
||||
Historically every gateway adapter hand-rolled its own mime→extension map
|
||||
before handing downloaded bytes to the cache primitives in
|
||||
``gateway.platforms.base`` (``cache_image_from_bytes``,
|
||||
``cache_audio_from_bytes``, ``cache_document_from_bytes``). Those maps
|
||||
*disagree* with each other on purpose — e.g. BlueBubbles coerces
|
||||
``image/heic`` to ``.jpg`` because downstream vision tools can't read HEIC,
|
||||
while WhatsApp Cloud pins ``audio/ogg`` to ``.ogg`` (not the RFC-correct
|
||||
``.oga`` Python's ``mimetypes`` returns) because the STT pipeline whitelists
|
||||
extensions.
|
||||
|
||||
This module owns:
|
||||
|
||||
* ``DEFAULT_MIME_TO_EXT`` — the union table of entries the adapters already
|
||||
agree on (plus a few uncontroversial document types).
|
||||
* ``DEFAULT_EXT_TO_MIME`` — the canonical inverse (used by Signal to map a
|
||||
sniffed extension back to a content type).
|
||||
* ``ext_for_mime`` / ``mime_for_ext`` — lookup helpers that accept
|
||||
per-adapter ``overrides`` so each adapter's historical (divergent)
|
||||
behavior is preserved byte-for-byte.
|
||||
* ``cache_media_bytes`` — one-call dispatch: classify the mime, resolve the
|
||||
extension, and write to the right cache (image / audio / document).
|
||||
|
||||
Behavior-preservation contract: adapters that had divergent maps pass them
|
||||
as ``overrides`` (and, where their historical code never consulted
|
||||
``mimetypes`` or a shared table, disable those fallbacks via
|
||||
``use_defaults`` / ``use_mimetypes``). The parity tests in
|
||||
``tests/gateway/test_media_cache.py`` hardcode the historical outputs as
|
||||
the contract.
|
||||
|
||||
NOTE: ``gateway/platforms/weixin.py`` also has a private mime map
|
||||
(``_mime_from_filename``) but is intentionally NOT migrated here — another
|
||||
in-flight branch edits that file. Follow-up: fold it in once that lands.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import mimetypes
|
||||
import uuid
|
||||
from typing import Mapping, Optional
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared tables
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Union of the per-adapter maps where the adapters already agree (or where
|
||||
# only one adapter pinned the type and no other adapter contradicts it).
|
||||
# Entries deliberately favor the common-in-the-wild extension over the
|
||||
# RFC-correct one (``audio/ogg`` → ``.ogg``, not ``.oga``) because the
|
||||
# downstream STT/vision pipelines whitelist real-world extensions.
|
||||
DEFAULT_MIME_TO_EXT: dict[str, str] = {
|
||||
# --- images (bluebubbles + whatsapp_cloud agree; matches mimetypes) ---
|
||||
"image/jpeg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
# --- audio ---
|
||||
"audio/ogg": ".ogg", # bluebubbles + whatsapp_cloud agree
|
||||
"audio/x-opus+ogg": ".ogg", # whatsapp voice notes (opus-in-ogg)
|
||||
"audio/opus": ".ogg", # whatsapp voice notes (opus-in-ogg)
|
||||
"audio/mpeg": ".mp3",
|
||||
"audio/mp3": ".mp3", # non-standard but seen in the wild
|
||||
"audio/wav": ".wav",
|
||||
"audio/mp4": ".m4a", # bluebubbles + whatsapp_cloud agree
|
||||
"audio/x-m4a": ".m4a",
|
||||
"audio/aac": ".aac",
|
||||
# --- video / documents (from signal's inverse table) ---
|
||||
"video/mp4": ".mp4",
|
||||
"application/pdf": ".pdf",
|
||||
"application/zip": ".zip",
|
||||
}
|
||||
|
||||
# Canonical inverse. Kept explicit (rather than mechanically inverted)
|
||||
# because the forward table is many-to-one — e.g. both ``audio/mpeg`` and
|
||||
# ``audio/mp3`` map to ``.mp3`` and the inverse must pick the canonical
|
||||
# mime. This is byte-identical to Signal's historical ``_EXT_TO_MIME``.
|
||||
DEFAULT_EXT_TO_MIME: dict[str, str] = {
|
||||
".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
|
||||
".gif": "image/gif", ".webp": "image/webp",
|
||||
".ogg": "audio/ogg", ".mp3": "audio/mpeg", ".wav": "audio/wav",
|
||||
".m4a": "audio/mp4", ".aac": "audio/aac",
|
||||
".mp4": "video/mp4", ".pdf": "application/pdf",
|
||||
".zip": "application/zip",
|
||||
}
|
||||
|
||||
|
||||
def _normalize_mime(mime: str) -> str:
|
||||
"""Lowercase and strip any ``; charset=...`` style parameters."""
|
||||
return (mime or "").split(";")[0].strip().lower()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lookups
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def ext_for_mime(
|
||||
mime: str,
|
||||
*,
|
||||
overrides: Optional[Mapping[str, str]] = None,
|
||||
use_defaults: bool = True,
|
||||
use_mimetypes: bool = True,
|
||||
fallback: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
"""Resolve a mime type to a file extension (including the dot).
|
||||
|
||||
Resolution order: ``overrides`` → ``DEFAULT_MIME_TO_EXT`` (if
|
||||
``use_defaults``) → ``mimetypes.guess_extension`` (if
|
||||
``use_mimetypes``) → ``fallback``.
|
||||
|
||||
Adapters with historical divergent maps pass them via ``overrides``
|
||||
and disable the stages their old code never consulted, keeping their
|
||||
outputs byte-identical to the pre-refactor behavior.
|
||||
"""
|
||||
primary = _normalize_mime(mime)
|
||||
if not primary:
|
||||
return fallback
|
||||
if overrides:
|
||||
ext = overrides.get(primary)
|
||||
if ext:
|
||||
return ext
|
||||
if use_defaults:
|
||||
ext = DEFAULT_MIME_TO_EXT.get(primary)
|
||||
if ext:
|
||||
return ext
|
||||
if use_mimetypes:
|
||||
ext = mimetypes.guess_extension(primary)
|
||||
if ext:
|
||||
return ext
|
||||
return fallback
|
||||
|
||||
|
||||
def mime_for_ext(
|
||||
ext: str,
|
||||
*,
|
||||
overrides: Optional[Mapping[str, str]] = None,
|
||||
fallback: str = "application/octet-stream",
|
||||
) -> str:
|
||||
"""Inverse lookup: file extension → canonical mime type.
|
||||
|
||||
Resolution order: ``overrides`` → ``DEFAULT_EXT_TO_MIME`` → ``fallback``.
|
||||
"""
|
||||
key = (ext or "").strip().lower()
|
||||
if overrides:
|
||||
mime = overrides.get(key)
|
||||
if mime:
|
||||
return mime
|
||||
return DEFAULT_EXT_TO_MIME.get(key, fallback)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# One-call cache dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def cache_media_bytes(
|
||||
data: bytes,
|
||||
mime: str,
|
||||
*,
|
||||
filename_hint: str = "",
|
||||
kind_hint: Optional[str] = None,
|
||||
ext_overrides: Optional[Mapping[str, str]] = None,
|
||||
) -> str:
|
||||
"""Cache downloaded media bytes and return the local file path.
|
||||
|
||||
Picks the image / audio / document cache primitive from
|
||||
``gateway.platforms.base`` based on the mime class (or an explicit
|
||||
``kind_hint`` of ``"image"``, ``"audio"`` or ``"document"``).
|
||||
``filename_hint`` is used for document caching (falls back to a
|
||||
generated name with the resolved extension). ``ext_overrides`` is
|
||||
threaded through to :func:`ext_for_mime` for adapters that need their
|
||||
historical mappings.
|
||||
"""
|
||||
# Local import: base is a large module and some adapters import this
|
||||
# module very early; keep import-time coupling minimal.
|
||||
from gateway.platforms.base import (
|
||||
cache_audio_from_bytes,
|
||||
cache_document_from_bytes,
|
||||
cache_image_from_bytes,
|
||||
)
|
||||
|
||||
primary = _normalize_mime(mime)
|
||||
kind = kind_hint
|
||||
if kind is None:
|
||||
if primary.startswith("image/"):
|
||||
kind = "image"
|
||||
elif primary.startswith("audio/"):
|
||||
kind = "audio"
|
||||
else:
|
||||
kind = "document"
|
||||
|
||||
if kind == "image":
|
||||
ext = ext_for_mime(primary, overrides=ext_overrides, fallback=".jpg") or ".jpg"
|
||||
return cache_image_from_bytes(data, ext)
|
||||
if kind == "audio":
|
||||
ext = ext_for_mime(primary, overrides=ext_overrides, fallback=".ogg") or ".ogg"
|
||||
return cache_audio_from_bytes(data, ext)
|
||||
|
||||
filename = filename_hint
|
||||
if not filename:
|
||||
ext = ext_for_mime(primary, overrides=ext_overrides, fallback=".bin")
|
||||
filename = f"file_{uuid.uuid4().hex[:8]}{ext}"
|
||||
return cache_document_from_bytes(data, filename)
|
||||
|
|
@ -71,6 +71,7 @@ from gateway.platforms.base import (
|
|||
cache_image_from_bytes,
|
||||
)
|
||||
from gateway.platforms.helpers import strip_markdown
|
||||
from gateway.platforms.media_cache import ext_for_mime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -1791,7 +1792,14 @@ class QQAdapter(BasePlatformAdapter):
|
|||
return None
|
||||
|
||||
if content_type.startswith("image/"):
|
||||
ext = mimetypes.guess_extension(content_type) or ".jpg"
|
||||
# preserves historical qqbot mapping: trust mimetypes'
|
||||
# guess (never the shared table) and fall back to .jpg.
|
||||
ext = ext_for_mime(
|
||||
content_type,
|
||||
use_defaults=False,
|
||||
use_mimetypes=True,
|
||||
fallback=".jpg",
|
||||
) or ".jpg"
|
||||
return cache_image_from_bytes(data, ext)
|
||||
elif content_type == "voice" or content_type.startswith("audio/"):
|
||||
# QQ voice messages are typically .amr or .silk format.
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ from gateway.platforms.base import (
|
|||
cache_image_from_url,
|
||||
)
|
||||
from gateway.platforms.helpers import redact_phone
|
||||
from gateway.platforms.media_cache import DEFAULT_EXT_TO_MIME, mime_for_ext
|
||||
from tools.audio_container import CONTAINER_TO_EXT, sniff_container
|
||||
from gateway.platforms.signal_format import markdown_to_signal
|
||||
from gateway.platforms.signal_rate_limit import (
|
||||
|
|
@ -122,18 +123,17 @@ def _is_audio_ext(ext: str) -> bool:
|
|||
return ext.lower() in {".mp3", ".wav", ".ogg", ".m4a", ".aac"}
|
||||
|
||||
|
||||
_EXT_TO_MIME = {
|
||||
".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
|
||||
".gif": "image/gif", ".webp": "image/webp",
|
||||
".ogg": "audio/ogg", ".mp3": "audio/mpeg", ".wav": "audio/wav",
|
||||
".m4a": "audio/mp4", ".aac": "audio/aac",
|
||||
".mp4": "video/mp4", ".pdf": "application/pdf", ".zip": "application/zip",
|
||||
}
|
||||
# Historical Signal ext→mime table now lives in
|
||||
# gateway.platforms.media_cache.DEFAULT_EXT_TO_MIME (byte-identical);
|
||||
# kept as a module alias for backwards compatibility with any callers
|
||||
# that referenced the private name.
|
||||
_EXT_TO_MIME = DEFAULT_EXT_TO_MIME
|
||||
|
||||
|
||||
def _ext_to_mime(ext: str) -> str:
|
||||
"""Map file extension to MIME type."""
|
||||
return _EXT_TO_MIME.get(ext.lower(), "application/octet-stream")
|
||||
# preserves historical signal mapping (shared table matches verbatim)
|
||||
return mime_for_ext(ext, fallback="application/octet-stream")
|
||||
|
||||
|
||||
def _remux_aac_to_m4a(aac_data: bytes) -> Optional[Tuple[bytes, str]]:
|
||||
|
|
|
|||
|
|
@ -79,6 +79,7 @@ from gateway.platforms.base import (
|
|||
SUPPORTED_DOCUMENT_TYPES,
|
||||
)
|
||||
from gateway.platforms.whatsapp_common import WhatsAppBehaviorMixin
|
||||
from gateway.platforms.media_cache import ext_for_mime
|
||||
from gateway import rich_sent_store
|
||||
from hermes_constants import get_hermes_dir
|
||||
|
||||
|
|
@ -164,18 +165,25 @@ async def _read_limited_request_body(request: Any, max_bytes: int) -> bytes:
|
|||
def _ext_for_mime(mime: str) -> Optional[str]:
|
||||
"""Resolve a mime type to the file extension we want on disk.
|
||||
|
||||
Consults the override map first so types like ``audio/ogg`` produce
|
||||
the extension downstream tools actually accept (``.ogg``, not the
|
||||
technically-correct-but-broken ``.oga``). Falls back to Python's
|
||||
``mimetypes.guess_extension`` for anything we haven't pinned.
|
||||
Thin wrapper over the shared dispatch in
|
||||
``gateway.platforms.media_cache``. Consults the WhatsApp override map
|
||||
first so types like ``audio/ogg`` produce the extension downstream
|
||||
tools actually accept (``.ogg``, not the technically-correct-but-broken
|
||||
``.oga``). Falls back to Python's ``mimetypes.guess_extension`` for
|
||||
anything we haven't pinned — the shared default table is skipped so
|
||||
behavior stays byte-identical to the historical implementation.
|
||||
"""
|
||||
if not mime:
|
||||
return None
|
||||
primary = mime.split(";")[0].strip().lower()
|
||||
override = _WHATSAPP_MIME_EXTENSION_OVERRIDES.get(primary)
|
||||
if override:
|
||||
return override
|
||||
return mimetypes.guess_extension(primary) or None
|
||||
return ext_for_mime(
|
||||
mime,
|
||||
# preserves historical whatsapp_cloud mapping: overrides →
|
||||
# mimetypes → None, never the shared default table.
|
||||
overrides=_WHATSAPP_MIME_EXTENSION_OVERRIDES,
|
||||
use_defaults=False,
|
||||
use_mimetypes=True,
|
||||
fallback=None,
|
||||
)
|
||||
|
||||
|
||||
# Inbound media cache lives under the user's hermes dir so it survives
|
||||
|
|
|
|||
|
|
@ -0,0 +1,262 @@
|
|||
"""Contract tests for gateway.platforms.media_cache — the shared mime↔ext
|
||||
dispatch — plus per-adapter parity spot-checks that hardcode each adapter's
|
||||
HISTORICAL (pre-refactor) mappings as the contract.
|
||||
|
||||
If any of these fail, an adapter's downloaded-media filenames changed —
|
||||
that's a behavioral regression, not a test to update casually.
|
||||
"""
|
||||
|
||||
import mimetypes
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.platforms.media_cache import (
|
||||
DEFAULT_EXT_TO_MIME,
|
||||
DEFAULT_MIME_TO_EXT,
|
||||
cache_media_bytes,
|
||||
ext_for_mime,
|
||||
mime_for_ext,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared table contract
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSharedTable:
|
||||
def test_defaults_resolve(self):
|
||||
for mime, ext in DEFAULT_MIME_TO_EXT.items():
|
||||
assert ext_for_mime(mime) == ext
|
||||
|
||||
def test_overrides_always_win(self):
|
||||
# Every override is honored even when the default table or
|
||||
# mimetypes disagree.
|
||||
assert ext_for_mime("image/heic", overrides={"image/heic": ".jpg"}) == ".jpg"
|
||||
assert ext_for_mime("audio/ogg", overrides={"audio/ogg": ".weird"}) == ".weird"
|
||||
assert ext_for_mime("image/jpeg", overrides={"image/jpeg": ".jpeg"}) == ".jpeg"
|
||||
|
||||
def test_mime_parameters_stripped(self):
|
||||
assert ext_for_mime("audio/ogg; codecs=opus") == ".ogg"
|
||||
assert ext_for_mime("IMAGE/JPEG; charset=binary") == ".jpg"
|
||||
|
||||
def test_unknown_mime_falls_back_to_mimetypes_then_fallback(self):
|
||||
# Known to mimetypes but not our table.
|
||||
assert ext_for_mime("image/bmp") == mimetypes.guess_extension("image/bmp")
|
||||
# Unknown everywhere → explicit fallback.
|
||||
assert ext_for_mime("application/x-no-such-type", fallback=".bin") == ".bin"
|
||||
assert ext_for_mime("application/x-no-such-type") is None
|
||||
|
||||
def test_empty_mime_returns_fallback(self):
|
||||
assert ext_for_mime("") is None
|
||||
assert ext_for_mime("", fallback=".bin") == ".bin"
|
||||
|
||||
def test_stage_gating(self):
|
||||
# use_defaults=False skips the shared table.
|
||||
assert ext_for_mime(
|
||||
"audio/ogg", use_defaults=False, use_mimetypes=False, fallback=".x"
|
||||
) == ".x"
|
||||
# use_mimetypes=False skips the mimetypes fallback.
|
||||
assert ext_for_mime("image/bmp", use_mimetypes=False) is None
|
||||
|
||||
def test_inverse_map_consistent_with_forward(self):
|
||||
# Round-trip: every inverse entry's mime maps forward to an ext
|
||||
# whose inverse is the same mime (canonical closure).
|
||||
for ext, mime in DEFAULT_EXT_TO_MIME.items():
|
||||
fwd_ext = ext_for_mime(mime)
|
||||
assert fwd_ext is not None
|
||||
assert mime_for_ext(fwd_ext) == mime
|
||||
|
||||
def test_mime_for_ext_fallback_and_case(self):
|
||||
assert mime_for_ext(".JPG") == "image/jpeg"
|
||||
assert mime_for_ext(".unknown") == "application/octet-stream"
|
||||
assert mime_for_ext(".unknown", fallback="x/y") == "x/y"
|
||||
assert mime_for_ext(".pdf", overrides={".pdf": "custom/pdf"}) == "custom/pdf"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# cache_media_bytes dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCacheMediaBytes:
|
||||
PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
|
||||
|
||||
def test_image_dispatch(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_image_cache_dir", lambda: tmp_path
|
||||
)
|
||||
path = cache_media_bytes(self.PNG, "image/png")
|
||||
assert path.endswith(".png")
|
||||
|
||||
def test_audio_dispatch(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_audio_cache_dir", lambda: tmp_path
|
||||
)
|
||||
path = cache_media_bytes(b"RIFF\x00\x00\x00\x00WAVEfmt ", "audio/wav")
|
||||
assert path.endswith(".wav")
|
||||
|
||||
def test_document_dispatch_uses_filename_hint(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_document_cache_dir", lambda: tmp_path
|
||||
)
|
||||
path = cache_media_bytes(b"%PDF-1.4", "application/pdf",
|
||||
filename_hint="report.pdf")
|
||||
assert path.endswith("_report.pdf")
|
||||
|
||||
def test_document_dispatch_generates_name(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_document_cache_dir", lambda: tmp_path
|
||||
)
|
||||
path = cache_media_bytes(b"%PDF-1.4", "application/pdf")
|
||||
assert path.endswith(".pdf")
|
||||
|
||||
def test_kind_hint_forces_cache(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_document_cache_dir", lambda: tmp_path
|
||||
)
|
||||
# Image mime but explicit document hint → document cache.
|
||||
path = cache_media_bytes(self.PNG, "image/png", kind_hint="document",
|
||||
filename_hint="pic.png")
|
||||
assert path.endswith("_pic.png")
|
||||
|
||||
def test_ext_overrides_threaded(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.get_image_cache_dir", lambda: tmp_path
|
||||
)
|
||||
path = cache_media_bytes(
|
||||
self.PNG, "image/png", ext_overrides={"image/png": ".png2"}
|
||||
)
|
||||
assert path.endswith(".png2")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-adapter parity: HISTORICAL mappings hardcoded as the contract
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestBlueBubblesParity:
|
||||
"""Historical closed maps from bluebubbles._download_attachment."""
|
||||
|
||||
IMAGE_CASES = {
|
||||
"image/jpeg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
"image/heic": ".jpg", # historically coerced to .jpg
|
||||
"image/heif": ".jpg", # historically coerced to .jpg
|
||||
"image/tiff": ".jpg", # historically coerced to .jpg
|
||||
"image/bmp": ".jpg", # unlisted → historical .jpg fallback
|
||||
}
|
||||
AUDIO_CASES = {
|
||||
"audio/mp3": ".mp3",
|
||||
"audio/mpeg": ".mp3",
|
||||
"audio/ogg": ".ogg",
|
||||
"audio/wav": ".wav",
|
||||
"audio/x-caf": ".mp3", # historically coerced to .mp3
|
||||
"audio/mp4": ".m4a",
|
||||
"audio/aac": ".m4a", # historically .m4a (NOT .aac)
|
||||
"audio/flac": ".mp3", # unlisted → historical .mp3 fallback
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize("mime,expected", sorted(IMAGE_CASES.items()))
|
||||
def test_image_map(self, mime, expected):
|
||||
from gateway.platforms.bluebubbles import _BLUEBUBBLES_IMAGE_EXT_OVERRIDES
|
||||
got = ext_for_mime(
|
||||
mime,
|
||||
overrides=_BLUEBUBBLES_IMAGE_EXT_OVERRIDES,
|
||||
use_defaults=False,
|
||||
use_mimetypes=False,
|
||||
fallback=".jpg",
|
||||
)
|
||||
assert got == expected
|
||||
|
||||
@pytest.mark.parametrize("mime,expected", sorted(AUDIO_CASES.items()))
|
||||
def test_audio_map(self, mime, expected):
|
||||
from gateway.platforms.bluebubbles import _BLUEBUBBLES_AUDIO_EXT_OVERRIDES
|
||||
got = ext_for_mime(
|
||||
mime,
|
||||
overrides=_BLUEBUBBLES_AUDIO_EXT_OVERRIDES,
|
||||
use_defaults=False,
|
||||
use_mimetypes=False,
|
||||
fallback=".mp3",
|
||||
)
|
||||
assert got == expected
|
||||
|
||||
|
||||
class TestWhatsAppCloudParity:
|
||||
"""Historical _ext_for_mime: overrides → mimetypes → None."""
|
||||
|
||||
CASES = {
|
||||
# Pinned overrides (Meta-sent types the STT pipeline needs pinned).
|
||||
"audio/ogg": ".ogg", # NOT mimetypes' .oga
|
||||
"audio/x-opus+ogg": ".ogg",
|
||||
"audio/opus": ".ogg",
|
||||
"audio/mp4": ".m4a",
|
||||
"audio/x-m4a": ".m4a",
|
||||
"image/jpeg": ".jpg", # NOT the legacy .jpe
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize("mime,expected", sorted(CASES.items()))
|
||||
def test_pinned_overrides(self, mime, expected):
|
||||
from gateway.platforms.whatsapp_cloud import _ext_for_mime
|
||||
assert _ext_for_mime(mime) == expected
|
||||
|
||||
def test_unpinned_falls_to_mimetypes(self):
|
||||
from gateway.platforms.whatsapp_cloud import _ext_for_mime
|
||||
assert _ext_for_mime("application/pdf") == mimetypes.guess_extension(
|
||||
"application/pdf"
|
||||
)
|
||||
|
||||
def test_unknown_returns_none(self):
|
||||
from gateway.platforms.whatsapp_cloud import _ext_for_mime
|
||||
assert _ext_for_mime("application/x-no-such-type") is None
|
||||
assert _ext_for_mime("") is None
|
||||
|
||||
def test_parameters_stripped(self):
|
||||
from gateway.platforms.whatsapp_cloud import _ext_for_mime
|
||||
assert _ext_for_mime("audio/ogg; codecs=opus") == ".ogg"
|
||||
|
||||
|
||||
class TestSignalParity:
|
||||
"""Historical _EXT_TO_MIME table from signal.py, verbatim."""
|
||||
|
||||
HISTORICAL = {
|
||||
".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
|
||||
".gif": "image/gif", ".webp": "image/webp",
|
||||
".ogg": "audio/ogg", ".mp3": "audio/mpeg", ".wav": "audio/wav",
|
||||
".m4a": "audio/mp4", ".aac": "audio/aac",
|
||||
".mp4": "video/mp4", ".pdf": "application/pdf",
|
||||
".zip": "application/zip",
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize("ext,expected", sorted(HISTORICAL.items()))
|
||||
def test_table(self, ext, expected):
|
||||
from gateway.platforms.signal import _ext_to_mime
|
||||
assert _ext_to_mime(ext) == expected
|
||||
assert _ext_to_mime(ext.upper()) == expected
|
||||
|
||||
def test_unknown_ext(self):
|
||||
from gateway.platforms.signal import _ext_to_mime
|
||||
assert _ext_to_mime(".xyz") == "application/octet-stream"
|
||||
|
||||
def test_shared_table_matches_historical_verbatim(self):
|
||||
assert DEFAULT_EXT_TO_MIME == self.HISTORICAL
|
||||
|
||||
|
||||
class TestQQBotParity:
|
||||
"""Historical qqbot image path: mimetypes.guess_extension or '.jpg'."""
|
||||
|
||||
@pytest.mark.parametrize("mime", [
|
||||
"image/jpeg", "image/png", "image/gif", "image/webp", "image/bmp",
|
||||
])
|
||||
def test_trusts_mimetypes(self, mime):
|
||||
historical = mimetypes.guess_extension(mime) or ".jpg"
|
||||
got = ext_for_mime(
|
||||
mime, use_defaults=False, use_mimetypes=True, fallback=".jpg"
|
||||
) or ".jpg"
|
||||
assert got == historical
|
||||
|
||||
def test_unknown_image_mime_falls_back_to_jpg(self):
|
||||
got = ext_for_mime(
|
||||
"image/x-no-such-type",
|
||||
use_defaults=False, use_mimetypes=True, fallback=".jpg",
|
||||
) or ".jpg"
|
||||
assert got == ".jpg"
|
||||
Loading…
Reference in New Issue