minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[mm] sglang-mm: server vision pipeline core (fetch/driver/pipeline) + Qwen VL (#32364)

Browse Source 
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kan Wu
2026-08-04 00:47:00 -07:00
committed by GitHub
co-authored by Claude Fable 5
parent 154f0ac662
commit 17d19081d9
31 changed files with 3223 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/unit/multimodal/rust/_mm_rust_utils.py
+92
View File
@@ -0,0 +1,92 @@
"""Shared fixtures for the native Rust multimodal suites.
Imported via ``sys.path`` from the sibling suites (unittest runs these files by
path, so a package-relative import would break ``python <file>``); the module
name is deliberately specific so it cannot shadow another suite's helpers on the
process-global ``sys.path``.
"""
import io
import json
import numpy as np
from PIL import Image
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import is_in_ci
register_cpu_ci(
est_time=0, suite="base-a-test-cpu", disabled="Rust multimodal test helpers"
)
def load_core():
"""The Rust ``_core`` extension, or ``None`` (→ skip) when not built
locally. In CI a missing extension is a hard failure, never a silent
skip — the CPU suite builds it from source."""
try:
from sglang.srt.multimodal import _core
return _core
except ImportError:
if is_in_ci():
raise
return None
IMAGE_TOKEN_ID = 900
VISION_START_ID = 901
VISION_END_ID = 902
VIDEO_TOKEN_ID = 903
PROCESSOR_CONFIGS = {
"qwen2_vl": dict(
patch_size=14,
merge_size=2,
temporal_patch_size=2,
min_pixels=56 * 56,
max_pixels=28 * 28 * 1280,
image_mean=[0.48145466, 0.4578275, 0.40821073],
image_std=[0.26862954, 0.26130258, 0.27577711],
),
"qwen2_5_vl": dict(
patch_size=14,
merge_size=2,
temporal_patch_size=2,
min_pixels=56 * 56,
max_pixels=28 * 28 * 1280,
image_mean=[0.5] * 3,
image_std=[0.5] * 3,
),
"qwen3_5": dict(
patch_size=16,
merge_size=2,
temporal_patch_size=2,
min_pixels=65536,
max_pixels=16777216,
image_mean=[0.5] * 3,
image_std=[0.5] * 3,
),
}
def make_image(width, height, seed=0):
rng = np.random.default_rng(seed)
y, x = np.mgrid[0:height, 0:width]
base = np.stack(
(x * 255 / max(width - 1, 1), y * 255 / max(height - 1, 1), (x + y) % 256),
axis=-1,
)
return Image.fromarray(
np.clip(base + rng.integers(0, 24, base.shape), 0, 255).astype(np.uint8)
)
def image_bytes(width, height, seed=0):
buffer = io.BytesIO()
make_image(width, height, seed).save(buffer, format="PNG")
return buffer.getvalue()
def spec_json(config, image_token_id=IMAGE_TOKEN_ID):
return json.dumps({"family": "qwen_vl", "image_token_id": image_token_id, **config})
test/registered/unit/multimodal/rust/inkling/test_bindings.py
+74
View File
@@ -0,0 +1,74 @@
"""Smoke coverage for the ``_core.inkling`` PyO3 bindings.
Covers the ``#[pyfunction]``s in ``rust/sglang-mm/src/inkling/mod.rs``
(``patchify_rgb`` / ``decode_patchify`` / ``decode_patchify_batch`` /
``preprocess_images`` / ``rescale_patchify_hash``) plus
``_core.common.content_hash``.
The inkling bindings are otherwise exercised only by the GPU e2e model test;
this pins the binding surface (signatures, dtypes, cross-binding consistency)
in the CPU suite so a rework of the extension can't silently break them.
"""
import sys
import unittest
from pathlib import Path
import numpy as np
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import image_bytes, load_core # noqa: E402
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
CORE = load_core()
PATCH_SIZE = 16
@unittest.skipUnless(CORE, "sglang-mm extension not built")
class TestInklingBindings(CustomTestCase):
def test_bindings_are_consistent(self):
data = image_bytes(50, 34)
h, w, patches = CORE.inkling.decode_patchify(data, PATCH_SIZE)
self.assertEqual((h, w), (34, 50))
patches = np.asarray(patches)
self.assertEqual(patches.dtype, np.uint16)
# One padded patch column, ceil rows (the inkling grid convention).
expected_len = -(-h // PATCH_SIZE) * (w // PATCH_SIZE + 1) * PATCH_SIZE**2 * 3
self.assertEqual(patches.size, expected_len)
# patchify_rgb on the decoded array must match the fused decode path.
dh, dw, rgb = CORE.common.image_decode_rgb(data)
arr = np.asarray(rgb).reshape(dh, dw, 3)
np.testing.assert_array_equal(
np.asarray(CORE.inkling.patchify_rgb(arr, PATCH_SIZE)), patches
)
# Batch and hashed variants agree with the single-image call.
[(bh, bw, batch)] = CORE.inkling.decode_patchify_batch([data], PATCH_SIZE)
self.assertEqual((bh, bw), (h, w))
np.testing.assert_array_equal(np.asarray(batch), patches)
[(ph, pw, pre, phash)] = CORE.inkling.preprocess_images([data], PATCH_SIZE)
self.assertEqual((ph, pw), (h, w))
np.testing.assert_array_equal(np.asarray(pre), patches)
self.assertEqual(phash, CORE.common.content_hash(data))
rh, rw, rpatches, rhash = CORE.inkling.rescale_patchify_hash(
arr, data, PATCH_SIZE
)
self.assertEqual((rh, rw), (h, w))
np.testing.assert_array_equal(np.asarray(rpatches), patches)
self.assertEqual(rhash, phash)
def test_invalid_inputs_rejected(self):
with self.assertRaises(ValueError):
CORE.inkling.decode_patchify(b"junk", PATCH_SIZE)
with self.assertRaises(ValueError):
CORE.inkling.decode_patchify(image_bytes(16, 16), 0)
if __name__ == "__main__":
unittest.main()
test/registered/unit/multimodal/rust/qwen/test_driver.py
+109
View File
@@ -0,0 +1,109 @@
"""Native driver error paths: out-of-scope and malformed inputs are rejected.
Covers ``process`` in ``rust/sglang-mm/src/driver.rs`` (via the
``_core.qwen_vl.process_native_mm`` binding). The wire-payload parsing that
feeds this driver (modality/shape rejection) lives in ``sglang-server``'s
message layer and is tested with the integration PR.
There is no Python fallback path, so the server rejects every driver error
back to the client as a 400; the message must say why (placeholder mismatch,
undecodable image, missing prompt). This pins that contract for each
rejection class.
"""
import io
import sys
import unittest
from pathlib import Path
import numpy as np
from PIL import Image
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import ( # noqa: E402
IMAGE_TOKEN_ID,
PROCESSOR_CONFIGS,
VISION_END_ID,
VISION_START_ID,
image_bytes,
load_core,
spec_json,
)
register_cpu_ci(est_time=10, suite="base-a-test-cpu")
QWEN_CORE = getattr(load_core(), "qwen_vl", None)
SPEC = spec_json(PROCESSOR_CONFIGS["qwen2_5_vl"])
IMAGE_IDS = [7, VISION_START_ID, IMAGE_TOKEN_ID, VISION_END_ID, 8]
def gif_bytes():
buffer = io.BytesIO()
Image.fromarray(np.zeros((16, 16, 3), dtype=np.uint8)).save(buffer, format="GIF")
return buffer.getvalue()
@unittest.skipUnless(
QWEN_CORE and hasattr(QWEN_CORE, "process_native_mm"),
"sglang-mm native Qwen driver not built",
)
class TestNativeDriverErrorPaths(CustomTestCase):
def assert_rejected(self, input_ids, images, pattern, spec=SPEC):
with self.assertRaisesRegex(ValueError, pattern):
QWEN_CORE.process_native_mm(input_ids, images, spec)
def test_degenerate_geometry_rejected_not_panicked(self):
"""A thin image against a tight ``max_pixels`` floors a side of the
smart_resize target to 0. That used to panic on a worker thread inside
the resize coefficient math; a Rust panic surfaces as ``PanicException``
(a ``BaseException``), not ``ValueError``, so this asserts the request
is rejected cleanly rather than crashing the pipeline."""
config = dict(PROCESSOR_CONFIGS["qwen2_5_vl"], min_pixels=3136, max_pixels=3136)
self.assert_rejected(
IMAGE_IDS,
[image_bytes(2000, 10)],
"smart_resize",
spec=spec_json(config),
)
def test_placeholder_count_mismatches_rejected(self):
cases = {
"no placeholder": ([7, 8], [image_bytes(80, 80)]),
"more images": (IMAGE_IDS, [image_bytes(80, 80), image_bytes(88, 80, 1)]),
"more placeholders": (IMAGE_IDS + IMAGE_IDS, [image_bytes(80, 80)]),
}
for name, (ids, images) in cases.items():
with self.subTest(case=name):
self.assert_rejected(ids, images, "placeholder")
def test_undecodable_images_rejected(self):
# Corrupt bytes are outside the native decoder's scope; the server
# rejects them as a 400.
self.assert_rejected(IMAGE_IDS, [b"junk"], "decode")
def test_gif_serves_through_the_pipeline(self):
"""GIF moved from rejected to served when the pure-Rust webp/gif/bmp
decoders were enabled; this pins the accept side of that contract flip
(the reject side used to be asserted here and broke in CI)."""
_, _, grids, _, offsets, _, _ = QWEN_CORE.process_native_mm(
IMAGE_IDS, [gif_bytes()], SPEC
)
self.assertEqual(len(grids), 1)
self.assertEqual(len(offsets), 1)
def test_missing_text_and_input_ids_rejected(self):
for input_ids in (None, []):
with self.subTest(input_ids=input_ids):
self.assert_rejected(
input_ids, [image_bytes(80, 80)], "without text or input_ids"
)
def test_image_free_request_rejected(self):
self.assert_rejected(IMAGE_IDS, [], "image sources")
if __name__ == "__main__":
unittest.main()
test/registered/unit/multimodal/rust/qwen/test_preprocess.py
+105
View File
@@ -0,0 +1,105 @@
"""Qwen native image preprocessing parity against Transformers.
Covers ``QwenVlProcessor::process_item`` and ``smart_resize`` in
``rust/sglang-mm/src/qwen_vl/mod.rs`` (via the ``_core.qwen_vl.preprocess``
and ``smart_resize_py`` bindings), against the HF Qwen2-VL image processors
and the Python ``smart_resize``.
Both HF processors are pinned, because they resample differently and only one
of them is what a server actually runs. On transformers 5.x
``Qwen2VLImageProcessor`` is the torchvision path (the ``Fast`` suffix was
dropped) and is what ``AutoImageProcessor`` hands SGLang by default;
``Qwen2VLImageProcessorPil`` is the PIL path, reachable via
``--disable-fast-image-processor``. The Rust resize is a bit-exact clone of
PIL's fixed-point kernel, so the PIL processor is asserted exactly and the
torchvision one carries the cross-implementation envelope.
"""
import sys
import unittest
from pathlib import Path
import numpy as np
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import ( # noqa: E402
PROCESSOR_CONFIGS,
image_bytes,
load_core,
make_image,
spec_json,
)
register_cpu_ci(est_time=20, suite="base-a-test-cpu")
QWEN_CORE = getattr(load_core(), "qwen_vl", None)
SIZES = ((640, 480), (1024, 683), (50, 40), (300, 301))
@unittest.skipUnless(QWEN_CORE, "sglang-mm Qwen binding not built")
class TestQwenImagePreprocess(CustomTestCase):
def _assert_matches(self, processor, max_diff, mean_diff):
for family, config in PROCESSOR_CONFIGS.items():
hf = processor(**config)
for index, size in enumerate(SIZES):
with self.subTest(family=family, size=size):
image = make_image(*size, seed=index)
actual, grid = QWEN_CORE.preprocess(
image_bytes(*size, seed=index), spec_json(config)
)
expected = hf(images=[image], return_tensors="pt")
self.assertEqual(grid, tuple(expected.image_grid_thw[0].tolist()))
diff = np.abs(
np.asarray(actual).reshape(expected.pixel_values.shape)
- expected.pixel_values.numpy()
)
# LessEqual, not Less: max_diff=0.0 is a real bound here.
self.assertLessEqual(diff.max(), max_diff)
self.assertLessEqual(diff.mean(), mean_diff)
def test_features_match_pil_processor_exactly(self):
"""Against the PIL processor the native path is bit-exact, so this is
asserted with zero tolerance: every stage (smart_resize geometry,
the fixed-point bicubic kernel, rescale/normalize, HF patch order) is
pinned, and any drift in any of them shows up here rather than being
absorbed by a tolerance."""
from transformers.models.qwen2_vl.image_processing_pil_qwen2_vl import (
Qwen2VLImageProcessorPil,
)
self._assert_matches(Qwen2VLImageProcessorPil, max_diff=0.0, mean_diff=0.0)
def test_features_match_torchvision_processor_within_envelope(self):
"""The torchvision processor is what a default server runs, so its
divergence is bounded separately — it is a different antialiased-bicubic
implementation, which the bit-exact PIL assertion above says nothing
about. Measured worst case is max 0.030 / mean 6.7e-5 (≈2 u8 levels
after normalize with the qwen2_vl std)."""
from transformers.models.qwen2_vl.image_processing_qwen2_vl import (
Qwen2VLImageProcessor,
)
self._assert_matches(Qwen2VLImageProcessor, max_diff=0.035, mean_diff=1e-3)
def test_smart_resize_matches_python(self):
from sglang.srt.multimodal.processors.qwen_vl import smart_resize
cases = (
(1365, 2048, 28, 3136, 12845056),
(3000, 4000, 28, 3136, 1003520),
(20, 20, 28, 3136, 12845056),
(1365, 2048, 32, 65536, 16777216),
(4000, 48, 32, 4, 1 << 30),
)
for case in cases:
with self.subTest(case=case):
self.assertEqual(QWEN_CORE.smart_resize_py(*case), smart_resize(*case))
if __name__ == "__main__":
unittest.main()
test/registered/unit/multimodal/rust/qwen/test_token_layout_mrope.py
+104
View File
@@ -0,0 +1,104 @@
"""Qwen placeholder expansion, offsets, and M-RoPE parity.
Covers ``layout_by_placeholder`` / ``apply_layout`` in
``rust/sglang-mm/src/common/token_layout.rs`` and ``mrope_image_only`` in
``rust/sglang-mm/src/qwen_vl/mod.rs`` (via the
``_core.qwen_vl.process_native_mm`` and ``mrope_image_only_py``
bindings), against ``BaseMultimodalProcessor`` expansion/offsets and
``MRotaryEmbedding.get_rope_index``.
"""
import sys
import unittest
from pathlib import Path
import numpy as np
import torch
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import ( # noqa: E402
IMAGE_TOKEN_ID,
PROCESSOR_CONFIGS,
VIDEO_TOKEN_ID,
VISION_END_ID,
VISION_START_ID,
image_bytes,
load_core,
spec_json,
)
register_cpu_ci(est_time=10, suite="base-a-test-cpu")
QWEN_CORE = getattr(load_core(), "qwen_vl", None)
@unittest.skipUnless(
QWEN_CORE and hasattr(QWEN_CORE, "process_native_mm"),
"sglang-mm native Qwen driver not built",
)
class TestQwenPromptGeometry(CustomTestCase):
def test_placeholder_expansion_and_offsets(self):
from sglang.srt.multimodal.processors.base_processor import (
BaseMultimodalProcessor,
)
config = PROCESSOR_CONFIGS["qwen2_5_vl"]
for image_count in (1, 2):
ids = [7]
for _ in range(image_count):
ids.extend((VISION_START_ID, IMAGE_TOKEN_ID, VISION_END_ID, 8))
images = [image_bytes(96 + 8 * i, 80, i) for i in range(image_count)]
with self.subTest(image_count=image_count):
actual_ids, _, grids, _, offsets, _, _ = QWEN_CORE.process_native_mm(
ids, images, spec_json(config)
)
counts = [t * h * w // config["merge_size"] ** 2 for t, h, w in grids]
expected_ids = BaseMultimodalProcessor._expand_input_ids(
ids, counts, IMAGE_TOKEN_ID
)
self.assertEqual(actual_ids, expected_ids)
self.assertEqual(
offsets,
BaseMultimodalProcessor.get_mm_items_offset(
torch.tensor(expected_ids), IMAGE_TOKEN_ID
),
)
def test_mrope_matches_model_reference(self):
"""The single Rust image-only M-RoPE must match ``get_rope_index``
for every native Qwen family (image-only makes them coincide)."""
from sglang.srt.layers.rotary_embedding import MRotaryEmbedding
grids = [(1, 4, 6), (1, 6, 4)]
ids, items = [10], []
for grid in grids:
ids.extend((VISION_START_ID, IMAGE_TOKEN_ID))
start = len(ids) - 1
ids.extend([IMAGE_TOKEN_ID] * (np.prod(grid) // 4 - 1))
items.append((start, len(ids) - 1, *grid))
ids.extend((VISION_END_ID, 11))
actual, delta = QWEN_CORE.mrope_image_only_py(len(ids), items, 2)
actual = np.asarray(actual).reshape(3, -1)
for model_type in ("qwen2_vl", "qwen2_5_vl", "qwen3_vl", "qwen3_5"):
with self.subTest(model_type=model_type):
expected, expected_delta = MRotaryEmbedding.get_rope_index(
spatial_merge_size=2,
image_token_id=IMAGE_TOKEN_ID,
video_token_id=VIDEO_TOKEN_ID,
vision_start_token_id=VISION_START_ID,
model_type=model_type,
tokens_per_second=2 if model_type == "qwen2_5_vl" else None,
input_ids=torch.tensor(ids).unsqueeze(0),
image_grid_thw=torch.tensor(grids),
video_grid_thw=None,
)
np.testing.assert_array_equal(actual, expected.squeeze(1).numpy())
self.assertEqual(delta, int(expected_delta.item()))
if __name__ == "__main__":
unittest.main()
test/registered/unit/multimodal/rust/shared/test_fetch.py
+83
View File
@@ -0,0 +1,83 @@
"""Model-independent parity tests for Rust multimodal source loading.
Covers ``fetch_bytes`` in ``rust/sglang-mm/src/common/fetch.rs`` (via the
``_core.common.fetch_bytes`` binding), against the Python reference
``sglang.srt.utils.common.get_image_bytes``.
"""
import base64
import http.server
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from sglang.srt.utils.common import get_image_bytes
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import load_core # noqa: E402
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
CORE = load_core()
FETCH = CORE and CORE.common.fetch_bytes
@unittest.skipUnless(FETCH, "sglang-mm fetch binding not built")
class TestRustMediaSourceLoading(CustomTestCase):
DATA = b"native-mm-source"
def test_inline_sources(self):
encoded = base64.b64encode(self.DATA).decode()
for source in (encoded, f"data:application/octet-stream;base64,{encoded}"):
with self.subTest(source=source[:8]):
self.assertEqual(bytes(FETCH(source)), get_image_bytes(source))
def test_file_sources(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "image.png"
path.write_bytes(self.DATA)
self.assertEqual(bytes(FETCH(str(path))), get_image_bytes(str(path)))
# `file://` is asserted against the payload, not the Python helper,
# on purpose: `get_image_bytes` passes the un-stripped URL straight
# to `open()`, so the reference raises here. The native path strips
# the scheme and succeeds — a deliberate divergence, not an omission.
with self.assertRaises(OSError):
get_image_bytes(path.as_uri())
self.assertEqual(bytes(FETCH(path.as_uri())), self.DATA)
def test_http_source(self):
data = self.DATA
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.end_headers()
self.wfile.write(data)
def log_message(self, *_):
pass
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
url = f"http://127.0.0.1:{server.server_port}/image"
self.assertEqual(bytes(FETCH(url)), get_image_bytes(url))
finally:
server.shutdown()
server.server_close()
thread.join()
def test_invalid_sources_fail(self):
for source in ("not base64!", "/definitely/missing/image.png"):
with self.subTest(source=source):
with self.assertRaises(ValueError):
FETCH(source)
if __name__ == "__main__":
unittest.main()
test/registered/unit/multimodal/rust/shared/test_image_decode.py
+102
View File
@@ -0,0 +1,102 @@
"""Model-independent image decode parity for native Rust MM.
Covers ``decode_rgb`` in ``rust/sglang-mm/src/common/mod.rs`` (via the
``_core.common.image_decode_rgb`` binding), against PIL's
``Image.open(...).convert("RGB")``.
"""
import io
import sys
import unittest
from pathlib import Path
import numpy as np
from PIL import Image
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from _mm_rust_utils import load_core # noqa: E402
register_cpu_ci(est_time=5, suite="base-a-test-cpu")
CORE = load_core()
DECODE = CORE and CORE.common.image_decode_rgb
def encode(image, fmt, **kwargs):
buffer = io.BytesIO()
image.save(buffer, format=fmt, **kwargs)
return buffer.getvalue()
@unittest.skipUnless(DECODE, "sglang-mm decode binding not built")
class TestRustImageDecode(CustomTestCase):
def assert_matches_pil(self, data, tolerance=0):
height, width, pixels = DECODE(data)
expected = np.asarray(Image.open(io.BytesIO(data)).convert("RGB"))
self.assertEqual((height, width), expected.shape[:2])
actual = np.asarray(pixels).reshape(expected.shape)
diff = np.abs(actual.astype(int) - expected.astype(int))
self.assertLessEqual(diff.max(), tolerance)
def test_png_modes_match_pil(self):
rgb = np.random.default_rng(1).integers(0, 256, (19, 23, 3), dtype=np.uint8)
cases = [
("RGB", Image.fromarray(rgb)),
("L", Image.fromarray(rgb[..., 0])),
("RGBA", Image.fromarray(np.dstack((rgb, rgb[..., 0])))),
("P", Image.fromarray(rgb).quantize(colors=16)),
]
for mode, image in cases:
with self.subTest(mode=mode):
self.assert_matches_pil(encode(image, "PNG"))
def test_jpeg_modes_match_with_decoder_tolerance(self):
rgb = np.random.default_rng(2).integers(0, 256, (31, 29, 3), dtype=np.uint8)
image = Image.fromarray(rgb)
exif = image.getexif()
exif[274] = 6 # EXIF orientation: neither decoder applies it
cases = [
("RGB", encode(image, "JPEG")),
("L", encode(Image.fromarray(rgb[..., 0]), "JPEG")),
("CMYK", encode(image.convert("CMYK"), "JPEG")),
("EXIF-rotated", encode(image, "JPEG", exif=exif)),
]
for mode, data in cases:
with self.subTest(mode=mode):
self.assert_matches_pil(data, tolerance=3)
def test_lossless_formats_match_pil_exactly(self):
"""GIF/BMP/lossless-WebP joined the native decoder with the pure-Rust
webp/gif/bmp enablement. Their reconstruction is exact by format spec,
so parity with PIL is pinned at zero tolerance like PNG."""
rgb = np.random.default_rng(4).integers(0, 256, (17, 21, 3), dtype=np.uint8)
image = Image.fromarray(rgb)
cases = {
"gif": encode(image.quantize(colors=64), "GIF"),
"bmp": encode(image, "BMP"),
"webp-lossless": encode(image, "WEBP", lossless=True),
}
for name, data in cases.items():
with self.subTest(fmt=name):
self.assert_matches_pil(data)
def test_unsupported_inputs_fail(self):
gray16 = Image.fromarray(np.zeros((9, 9), dtype=np.uint16))
cases = {
"corrupt": b"not an image",
# >8-bit depths must error (the request is then rejected — there
# is no Python fallback) — never silently diverge from PIL's
# clipping.
"png16": encode(gray16, "PNG"),
}
for name, data in cases.items():
with self.subTest(input=name):
with self.assertRaises(ValueError):
DECODE(data)
if __name__ == "__main__":
unittest.main()