diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index 60fe57031d99e..c88240f01c7b6 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -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 diff --git a/gateway/platforms/media_cache.py b/gateway/platforms/media_cache.py new file mode 100644 index 0000000000000..86768987acccd --- /dev/null +++ b/gateway/platforms/media_cache.py @@ -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) diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index 42a7bc650e356..8f5d7fececf96 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -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. diff --git a/gateway/platforms/signal.py b/gateway/platforms/signal.py index 59413c62cd927..2282193cc95d1 100644 --- a/gateway/platforms/signal.py +++ b/gateway/platforms/signal.py @@ -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]]: diff --git a/gateway/platforms/whatsapp_cloud.py b/gateway/platforms/whatsapp_cloud.py index b401e941e9e5c..84242adf2c168 100644 --- a/gateway/platforms/whatsapp_cloud.py +++ b/gateway/platforms/whatsapp_cloud.py @@ -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 diff --git a/tests/gateway/test_media_cache.py b/tests/gateway/test_media_cache.py new file mode 100644 index 0000000000000..66922bffdb089 --- /dev/null +++ b/tests/gateway/test_media_cache.py @@ -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"