refactor(telegram): generalize observed-media caching into a reusable primitive

Collapse the per-type observed-media dispatch into one platform-agnostic cache_media_bytes() helper in gateway/platforms/base.py. Any adapter can now hand it raw attachment bytes + a filename/MIME hint; it classifies against the shared MIME registries, routes to the right cache_*_from_bytes helper, sandbox-translates the path, and returns a CachedMedia with a ready context_note(). Telegram's observed-group path shrinks to: size-gate, download, call the helper, annotate. Also dedupes the addressed-media type ladder into _media_message_type(). Net: contributor's Telegram-only +595 LOC becomes a +210/-32 production change, with the reusable primitive available to Discord/Slack/Signal/etc. Co-authored-by: Glucksberg <markuscontasul@gmail.com>
2026-06-01 20:18:41 -07:00
parent f768e75ecf
commit fa3b06b035
4 changed files with 420 additions and 619 deletions
@@ -155,3 +155,64 @@ class TestSupportedDocumentTypes:
    )
    def test_expected_extensions_present(self, ext):
        assert ext in SUPPORTED_DOCUMENT_TYPES
+
+
+# ---------------------------------------------------------------------------
+# TestCacheMediaBytes — the unified, platform-agnostic caching primitive
+# ---------------------------------------------------------------------------
+
+# 1x1 transparent PNG (passes cache_image_from_bytes validation)
+_PNG_1PX = bytes.fromhex(
+    "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4"
+    "890000000d49444154789c6360000002000154a24f5f0000000049454e44ae426082"
+)
+
+
+class TestCacheMediaBytes:
+    def test_pdf_routes_to_document(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(b"%PDF-1.4 body", filename="report.pdf", mime_type="application/pdf")
+        assert result is not None
+        assert result.kind == "document"
+        assert result.media_type == "application/pdf"
+        assert "report.pdf" in result.display_name
+        assert os.path.exists(result.path)
+        assert "report.pdf" in result.context_note()
+
+    def test_png_routes_to_image(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(_PNG_1PX, filename="photo.png", mime_type="image/png")
+        assert result is not None
+        assert result.kind == "image"
+        assert result.media_type == "image/png"
+        assert os.path.exists(result.path)
+
+    def test_native_photo_without_filename_uses_default_kind(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(_PNG_1PX, filename="", mime_type="", default_kind="image")
+        assert result is not None
+        assert result.kind == "image"
+
+    def test_mp4_routes_to_video(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(b"\x00\x00\x00\x18ftypmp42", filename="clip.mp4", mime_type="video/mp4")
+        assert result is not None
+        assert result.kind == "video"
+        assert result.media_type == "video/mp4"
+
+    def test_mime_only_resolves_extension(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(b"col1,col2\n1,2", filename="", mime_type="text/csv")
+        assert result is not None
+        assert result.kind == "document"
+        assert result.media_type == "text/csv"
+
+    def test_unsupported_document_returns_none(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(b"MZ", filename="malware.exe", mime_type="application/x-msdownload")
+        assert result is None
+
+    def test_invalid_image_returns_none(self):
+        from gateway.platforms.base import cache_media_bytes
+        result = cache_media_bytes(b"<html>not an image</html>", filename="x.png", mime_type="image/png")
+        assert result is None