diff --git a/FIXES.md b/FIXES.md new file mode 100644 index 000000000..5d2d827a5 --- /dev/null +++ b/FIXES.md @@ -0,0 +1,168 @@ +# FIXES.md — Gemma 4 merge from `llama-cpp-python-old` + +Date: 2026-07-03 + +This documents the merge of the custom Gemma 4 work (chat handler with vision, +audio, tool calling, and thinking-mode support) from the old working copy +(`C:\AI\llama-cpp-python-old`) onto the current upstream base (v0.3.32, +commit `346853c`). The strategy was a **curated merge**: keep the newer +upstream infrastructure everywhere, and port only the Gemma-specific additions +on top of it. All ported blocks were verified byte-identical to the old +implementations via AST source comparison. + +--- + +## 1. `llama_cpp/llama_chat_format.py` (+364 / −3 lines) + +### Line 10 — added import +```python +import threading +``` +Required by `Gemma4ChatHandler._format_lock`. + +### Line 1438 — comment clarification +The `format_gemma` header comment now reads +"Google's Gemma models (Gemma 2 and Gemma 3)" to distinguish it from the new +Gemma 4 format below it. + +### Lines 1458–1529 — added `format_gemma4` (registered as `"gemma4"`) +Text-only prompt formatter for Gemma 4: +- `` + `<|turn>role\n...` turn structure, assistant mapped to `model`. +- System messages rendered into a `<|channel>thought ... ` block. +- Accepts a `reasoning_budget` keyword (flows in automatically because the + generic handler forwards `**kwargs` to registered formatters). +- Stop tokens: `\n`, ``, ``. + +### Lines 3849–3904 — added `GemmaChatHandler(Llava15ChatHandler)` +Multimodal handler for Gemma 2/3-family vision models (PaliGemma, MedGemma) +using ``/`` tokens; system messages folded into a +user turn. **This replaces the empty stub the upstream base had at this spot** +(`class Gemma4ChatHandler(MTMDChatHandler): pass`) — the real Gemma 4 handler +now lives at line 4302 (see below). + +### Lines 4259–4294 — added `MultimodalGemmaChatHandler(Llava15ChatHandler)` +Minimal ``-style multimodal template with no default system +message (Gemma has no native system role). + +### Lines 4302–4488 — added full `Gemma4ChatHandler(Llava15ChatHandler)` +The complete Gemma 4 handler ported from the old repo: +- **Line 4316** — `DEFAULT_SYSTEM_MESSAGE = None`. +- **Line 4322** — class-level `_format_lock` (`threading.Lock`) protecting the + `CHAT_FORMAT` mutation in `__call__` against concurrent requests, so a + thinking-mode call cannot leak `<|think|>` tokens into a standard call on a + shared instance. +- **Lines 4324–4375** — `CHAT_FORMAT` Jinja template covering: + - system messages in `<|channel>thought ... `; + - user turns with media emitted before text — images (`image_url` string or + mapping) **and audio** (`input_audio` OpenAI schema or custom `audio` + schema, emitted as `data:audio/;base64,...` URIs); + - assistant turns as `<|turn>model ... `; + - tool calls (`<|tool_call>call:`) and tool + responses (`<|tool_response>response:...`). +- **Lines 4378–4412** — `get_image_urls()` override that extracts image URLs + *and* audio base64 payloads so the mtmd backend replaces them with media + marker embeddings. +- **Lines 4414–4488** — `__call__` override: + - `enable_thinking=True` injects `<|think|>` into the system thought channel + and the generation prompt, inserting a blank system message if none exists; + - clears llama state per call (`reset()`, `kv_cache_clear()`, `n_tokens = 0`, + `input_ids.fill(0)`, cached image embeds) for reliable multi-turn + multimodal use, matching the Qwen25VL handler pattern; + - template mutation + `super().__call__()` + restore runs under + `_format_lock` with a `finally` restore; + - non-streaming thinking-mode responses get a `message["thinking"]` field. + +### Wiring that now activates (no changes needed — already in the base) +- `llama_cpp/server/model.py:118-123` maps `chat_format in ("mtmd", "gemma4")` + to `MTMDChatHandler` / `Gemma4ChatHandler`; the latter was an empty stub and + is now the full implementation. +- README model table entry for `gemma-4` / `Gemma4ChatHandler` / `gemma4`. +- `examples/server/server.py` `gemma4-tool-call` parser and the colab + notebooks (`Gemma4-12B-QAT.ipynb`, `notebook.ipynb`) — identical in both + repos already. + +--- + +## 2. `tests/` — three new files + +### `tests/test_gemma4_chat_format.py` (729 lines, new) +Ported unchanged from the old repo. Covers `format_gemma4` (registration, +turn/channel tokens, system-in-thought-channel, role mapping, stop tokens, +reasoning_budget), `Gemma4ChatHandler.get_image_urls` (image mapping/string, +OpenAI `input_audio`, custom `audio`, mixed media, defaults), the +`Gemma4ChatHandler` template (turn tokens, media-before-text ordering, tool +call/response rendering, generation prompt), the `_format_lock`, and the +`MultimodalGemmaChatHandler` template. + +### `tests/test_mtmd_cpp.py` (332 lines, new) +Ported unchanged from the old repo. Structural tests for the +`llama_cpp.mtmd_cpp` ctypes bindings. + +### `tests/conftest.py` (65 lines, new — ported **with fixes**) +Mocks `ctypes.CDLL` and `pathlib.Path.exists` so the suite can import +`llama_cpp` on a source-only checkout (no compiled library). Two fixes were +applied on top of the old repo's version: + +- **Lines 18–23** (`_HAS_REAL_LIB`) and **line 48** (`if not _HAS_REAL_LIB:`) + — the old conftest patched unconditionally, which would have broken the + real-model integration tests on any machine/CI where the compiled library + *is* present. The mocks now activate only when `llama_cpp/lib` contains no + `.so`/`.dylib`/`.dll`. +- **Lines 52–65** (`_patched_add_dll_directory`) — Windows fix. On Windows, + `load_shared_library()` calls `os.add_dll_directory()` on the (nonexistent) + `llama_cpp/lib` directory *before* the mocked `CDLL` is ever reached, so the + old repo's suite could not even collect (`FileNotFoundError` at import). + Missing directories are now tolerated. + +--- + +## 3. Kept from the new upstream base (old-repo removals **not** merged) + +The old file appears to have had a standalone version pasted over newer +upstream code at some point, deleting upstream improvements. These were +deliberately **retained**: + +- `MTMDChatHandler` — the generic multimodal handler (old file deleted it, + which would have broken `chat_format="mtmd"` in `server/model.py`). +- Transformers-aligned Jinja environment in `Jinja2ChatFormatter`: the + `tojson` filter, the `{% generation %}` tag pass-through extension, and + `loopcontrols` (upstream #1486, #2018, #2226). +- Streaming logprobs conversion + (`_convert_text_completion_logprobs_to_chat`) in the functionary handler — + the old file replaced one call site with `logprobs: None` (a regression). +- `llama_model_n_layer_nextn` binding in `llama_cpp/llama_cpp.py` — the only + difference in that file; the old copy simply predates it (upstream #2318+). + +--- + +## 4. Deliberately not ported (available on request) + +- **Functionary `generate_streaming` chunk-safety refactor** (old commits + `f391065`, `7e04f11`): `uuid`/`time`-based fallbacks for `chunk_id`/ + `chunk_created`/`chunk_model` guarding against `UnboundLocalError` on empty + completion streams. Real fix, but it conflicts with upstream's newer + streaming-logprobs code in the same hunks and included the logprobs + regression noted above. Can be re-ported cleanly if functionary models are + in use. +- **`stop: ... = []` → `stop: ... = None` default changes** (5 signatures): + style-only (the default is never mutated); skipped to minimize divergence + from upstream. +- **Untracked dev files from the old repo**: `.editorconfig`, `build.txt`, + `monitor_inference-full.py`. + +--- + +## 5. Verification + +- `python -m py_compile llama_cpp/llama_chat_format.py` — OK. +- AST comparison of `format_gemma4`, `GemmaChatHandler`, + `MultimodalGemmaChatHandler`, `Gemma4ChatHandler` against the old file — + all byte-identical. +- `pytest tests/test_gemma4_chat_format.py tests/test_mtmd_cpp.py` — + **100 passed**. +- `pytest tests/` — **107 passed, 19 failed**; all 19 failures are + pre-existing real-model integration tests in `test_llama.py` + (`test_real_llama*`, `test_*_prompt_cache*`, `test_*_matches_fresh`, + `test_llama_cpp_tokenization`) that require the compiled library and + downloaded GGUF models, which this source-only checkout does not have. + They are unrelated to the merge and still run for real on a built tree. diff --git a/llama_cpp/llama_chat_format.py b/llama_cpp/llama_chat_format.py index 0034bdae9..4533d9526 100644 --- a/llama_cpp/llama_chat_format.py +++ b/llama_cpp/llama_chat_format.py @@ -7,6 +7,7 @@ import dataclasses import random import string +import threading from datetime import datetime from contextlib import ExitStack @@ -1434,7 +1435,7 @@ def format_saiga( return ChatFormatterResponse(prompt=_prompt.strip()) -# Chat format for Google's Gemma models, see more details and available models: +# Chat format for Google's Gemma models (Gemma 2 and Gemma 3), see more details and available models: # https://huggingface.co/collections/google/gemma-release-65d5efbccdbb8c4202ec078b @register_chat_format("gemma") def format_gemma( @@ -1454,6 +1455,80 @@ def format_gemma( return ChatFormatterResponse(prompt=_prompt, stop=_sep) +# Chat format for Google's Gemma 4 models, see more details: +# https://huggingface.co/google/gemma-4-E2B-it +# https://ai.google.dev/gemma/docs/core/prompt-structure +# Gemma 4 introduces new special tokens and native system role support +@register_chat_format("gemma4") +def format_gemma4( + messages: List[llama_types.ChatCompletionRequestMessage], + reasoning_budget: Optional[int] = None, + **kwargs: Any, +) -> ChatFormatterResponse: + """Format messages for Gemma 4 models using the new <|turn> and tokens. + + Gemma 4 introduces: + - Native system role support via <|channel>thought\n ... \n + - New turn-based tokens: <|turn>, , <|channel>, + - Thinking mode support via <|think|> token + - Tool calling support via <|tool_call>, , etc. + + This is a simplified formatter that handles basic text-only conversations. + For full multimodal and tool calling support, use the Gemma4ChatHandler class. + + Special tokens: + - : Beginning of sequence + - <|turn>: Start of turn + - : End of turn + - <|channel>: Start of channel + - : End of channel + - <|think|>: Thinking mode indicator + - <|tool_call>: Start of tool call + - : End of tool call + + Args: + messages: List of chat completion messages + reasoning_budget: Maximum number of tokens for thinking/reasoning (Gemma 4 feature) + **kwargs: Additional keyword arguments + """ + _bos_token = "" + _turn_start = "<|turn>" + _turn_end = "\n" + _channel_start = "<|channel>" + _channel_end = "\n" + + _prompt = _bos_token + + # Check for system message - in Gemma 4, system messages go in a thought channel + system_message = _get_system_message(messages) + if system_message: + _prompt += f"{_channel_start}thought\n{system_message}{_channel_end}" + + # Format conversation turns + for message in messages: + role = message["role"] + content = message.get("content", "") + + # Skip system messages as they're handled separately + if role == "system": + continue + + # Map role to Gemma 4 role names + if role == "assistant": + gemma_role = "model" + else: + gemma_role = role + + _prompt += f"{_turn_start}{gemma_role}\n{content}{_turn_end}" + + # Add generation prompt + _prompt += f"{_turn_start}model\n" + + return ChatFormatterResponse( + prompt=_prompt, stop=[_turn_end, "", ""] + ) + + # Tricky chat formats that require custom chat handlers @@ -3771,8 +3846,62 @@ def from_pretrained( ) -class Gemma4ChatHandler(MTMDChatHandler): - pass +class GemmaChatHandler(Llava15ChatHandler): + """Chat handler for Gemma-based multimodal models (e.g., PaliGemma, MedGemma). + + Gemma models use / control tokens instead of + the LLaVA-style USER:/ASSISTANT: format. The text-only 'gemma' chat format + is already registered (see format_gemma), but multimodal Gemma models that + require a Llava-style vision pipeline need a dedicated handler so the + correct chat template is applied when chat_handler takes precedence over + chat_format in the resolution order. + + See: https://ai.google.dev/gemma/docs/formatting + """ + + DEFAULT_SYSTEM_MESSAGE = None # Gemma models do not natively support a system role + + CHAT_FORMAT = ( + "{% for message in messages %}" + # System messages are folded into a user turn (Gemma has no system role) + "{% if message.role == 'system' %}" + "user\n{{ message.content }}\n" + "{% endif %}" + # User message (handles both plain string and multimodal content list) + "{% if message.role == 'user' %}" + "user\n" + "{% if message.content is string %}" + "{{ message.content }}" + "{% endif %}" + "{% if message.content is iterable and message.content is not string %}" + # Emit image tokens first + "{% for content in message.content %}" + "{% if content.type == 'image_url' and content.image_url is string %}" + "{{ content.image_url }}" + "{% endif %}" + "{% if content.type == 'image_url' and content.image_url is mapping %}" + "{{ content.image_url.url }}" + "{% endif %}" + "{% endfor %}" + # Then emit text tokens + "{% for content in message.content %}" + "{% if content.type == 'text' %}" + "{{ content.text }}" + "{% endif %}" + "{% endfor %}" + "{% endif %}" + "\n" + "{% endif %}" + # Assistant message + "{% if message.role == 'assistant' and message.content is not none %}" + "model\n{{ message.content }}\n" + "{% endif %}" + "{% endfor %}" + # Generation prompt + "{% if add_generation_prompt %}" + "model\n" + "{% endif %}" + ) class ObsidianChatHandler(Llava15ChatHandler): @@ -4127,6 +4256,238 @@ def __call__(self, **kwargs): return super().__call__(**kwargs) +class MultimodalGemmaChatHandler(Llava15ChatHandler): + DEFAULT_SYSTEM_MESSAGE: Optional[str] = None + + CHAT_FORMAT = ( + "{% for message in messages %}" + "{% if message.role == 'user' %}" + "user\n" + "{% if message.content is string %}" + "{{ message.content }}" + "{% endif %}" + "{% if message.content is iterable %}" + "{% for content in message.content %}" + "{% if content.type == 'image_url' and content.image_url is string %}" + "{{ content.image_url }}" + "{% endif %}" + "{% if content.type == 'image_url' and content.image_url is mapping %}" + "{{ content.image_url.url }}" + "{% endif %}" + "{% endfor %}" + "{% for content in message.content %}" + "{% if content.type == 'text' %}" + "{{ content.text }}" + "{% endif %}" + "{% endfor %}" + "{% endif %}" + "\n" + "{% endif %}" + "{% if message.role == 'assistant' and message.content is not none %}" + "model\n" + "{{ message.content }}\n" + "{% endif %}" + "{% endfor %}" + "{% if add_generation_prompt %}" + "model\n" + "{% endif %}" + ) + + +# ============================================================ +# GEMMA 4 CHAT HANDLER - FULLY CORRECTED & POLISHED +# ============================================================ + + +class Gemma4ChatHandler(Llava15ChatHandler): + """Chat handler for Gemma 4 models with full multimodal and tool calling support. + + Gemma 4 introduces new special tokens and native system role support: + - <|turn>: Start of turn + - : End of turn + - <|channel>: Start of channel (for system/thought messages) + - : End of channel + - <|think|>: Thinking mode indicator + - <|tool_call>: Start of tool call + - : End of tool call + - <|tool_response|>: Tool response marker + """ + + DEFAULT_SYSTEM_MESSAGE: Optional[str] = None + + # Protects CHAT_FORMAT mutation in __call__ from concurrent requests. + # Without this, a thinking-mode call that sets self.CHAT_FORMAT between + # the assignment and super().__call__() can race with a standard call on + # a shared instance, leaking <|think|> tokens into the standard response. + _format_lock: threading.Lock = threading.Lock() + + CHAT_FORMAT = ( + "{% for message in messages %}" + # 1. System messages go in a thought channel + "{% if message.role == 'system' %}" + "<|channel>thought\n{{ message.content }}\n" + "{% endif %}" + # 2. User message (handles both plain string and multimodal media) + "{% if message.role == 'user' %}" + "<|turn>user\n" + "{% if message.content is string %}" + "{{ message.content }}" + "{% endif %}" + "{% if message.content is iterable and message.content is not string %}" + # Emit Media Embeddings (Images AND Audio) + "{% for content in message.content %}" + "{% if content.type == 'image_url' %}" + "{% if content.image_url is string %}{{ content.image_url }}{% else %}{{ content.image_url.url }}{% endif %}" + "{% elif content.type == 'input_audio' %}" + "data:audio/{{ content.input_audio.format }};base64,{{ content.input_audio.data }}" + "{% elif content.type == 'audio' %}" + "data:audio/{{ content.audio.format }};base64,{{ content.audio.data }}" + "{% endif %}" + "{% endfor %}" + # Then emit text tokens + "{% for content in message.content %}" + "{% if content.type == 'text' %}{{ content.text }}{% endif %}" + "{% endfor %}" + "{% endif %}" + "\n" + "{% endif %}" + # 3. Assistant message + "{% if message.role == 'assistant' and message.content is not none %}" + "<|turn>model\n{{ message.content }}\n" + "{% endif %}" + # 4. Tool Calls (Agentic Workflow Handshakes) + "{% if message.role == 'assistant' and message.tool_calls %}" + "<|turn>model\n" + "{% for tool_call in message.tool_calls %}" + "<|tool_call>call:{{ tool_call.function.name }}{{ tool_call.function.arguments }}\n" + "{% endfor %}" + "\n" + "{% endif %}" + # 5. Tool Responses + "{% if message.role == 'tool' %}" + "<|tool_response>response:{{ message.name | default(message.tool_call_id) }}{{ message.content }}\n" + "{% endif %}" + "{% endfor %}" + # 6. Generation prompt + "{% if add_generation_prompt %}" + "<|turn>model\n" + "{% endif %}" + ) + + @staticmethod + def get_image_urls( + messages: List[llama_types.ChatCompletionRequestMessage], + ) -> List[str]: + """ + Overrides the base Llava15ChatHandler method. + Extracts both image URLs and audio base64 data strings so they can be processed + and replaced by the mtmd C++ media marker embeddings in the backend. + """ + media_urls: List[str] = [] + for message in messages: + if message["role"] == "user" and message.get("content"): + for content in message["content"]: + if isinstance(content, dict) and "type" in content: + # Extract Vision + if content["type"] == "image_url": + if ( + isinstance(content["image_url"], dict) + and "url" in content["image_url"] + ): + media_urls.append(content["image_url"]["url"]) + else: + media_urls.append(content["image_url"]) + + # Extract Audio (Supports OpenAI's 'input_audio' AND custom 'audio' schemas) + elif content["type"] in ["input_audio", "audio"]: + audio_data = content.get("input_audio") or content.get( + "audio" + ) + if audio_data: + fmt = audio_data.get("format", "wav") + data = audio_data.get("data", "") + # Standardize the output so `load_image` successfully base64-decodes the bytes + media_urls.append(f"data:audio/{fmt};base64,{data}") + + return media_urls + + def __call__(self, **kwargs): + """ + Overrides the __call__ pipeline to dynamically intercept and enable Thinking Mode + by injecting the required control token seamlessly into the Jinja template. + Also performs state clearing for reliable multimodal (vision + audio) support + across multiple chat turns, matching other vision handlers like Qwen25VL. + """ + enable_thinking = kwargs.get("enable_thinking", False) + original_format = self.CHAT_FORMAT + + if enable_thinking: + # Inject <|think|> into BOTH the initial system thought channel AND + # the assistant generation prompt so thinking starts the response turn. + # This follows Gemma 4 docs for triggering native thinking mode. + modified_format = original_format.replace( + "<|channel>thought\n", "<|channel>thought\n<|think|>\n" + ).replace( + "{% if add_generation_prompt %}\n<|turn>model\n{% endif %}", + "{% if add_generation_prompt %}\n<|turn>model\n<|think|>\n{% endif %}", + ) + + # Gemma requires a system block for the thought channel to exist. + # If the user hasn't provided one, we dynamically append a blank one. + messages = kwargs.get("messages", []) + if not any(m.get("role") == "system" for m in messages): + kwargs["messages"] = [{"role": "system", "content": ""}] + messages + else: + modified_format = original_format + + # Clear state for multiple runs (critical for vision/audio + thinking in chat) + llama = kwargs.get("llama") + if llama is not None: + llama.reset() + if hasattr(llama, "_ctx") and llama._ctx is not None: + llama._ctx.kv_cache_clear() + llama.n_tokens = 0 + if hasattr(llama, "input_ids"): + llama.input_ids.fill(0) + + # Clear any handler state (e.g. cached embeds from previous multimodal turn) + if hasattr(self, "_last_image_embed"): + self._last_image_embed = None + self._last_image_hash = None + + # Hold _format_lock for the entire mutation→call→restore cycle so that + # a concurrent standard request cannot observe the thinking-mode template. + # The lock is released before post-processing, which doesn't touch + # CHAT_FORMAT, so contention is minimised. + with type(self)._format_lock: + self.CHAT_FORMAT = modified_format + try: + result = super().__call__(**kwargs) + finally: + # Always restore, even if super().__call__() raises. + self.CHAT_FORMAT = original_format + + # Post-process non-streaming responses when thinking mode is enabled + # to provide clear structure: 'thinking' field (contains reasoning) + 'content' (final answer). + # Note: Since Gemma 4 outputs thinking + final answer in a single generation, + # 'thinking' currently holds the full generated text (including reasoning). + # Future: parse on model-specific end-of-thinking markers (e.g. <|end_think|>) if emitted. + if ( + enable_thinking + and not kwargs.get("stream", False) + and isinstance(result, dict) + ): + for choice in result.get("choices", []): + if "message" in choice: + content = choice["message"].get("content", "") or "" + choice["message"]["thinking"] = ( + content # structured access for test app + ) + # content remains the complete response (thinking + final answer) for compatibility + + return result + + @register_chat_completion_handler("chatml-function-calling") def chatml_function_calling( llama: llama.Llama, diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 000000000..c63befbde --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,65 @@ +""" +Shared test configuration and fixtures. + +This conftest.py applies module-level patches to allow importing llama_cpp +without requiring the compiled shared libraries to be present. The patches +run at collection time (before any test module is imported) so that every +test in the suite can import llama_cpp freely. + +The patches are only applied when no compiled library is found on disk, so a +built tree (or CI) still runs the integration tests against the real library. +""" + +import ctypes +import os +import pathlib +from unittest.mock import MagicMock + +_LIB_DIR = pathlib.Path(__file__).resolve().parent.parent / "llama_cpp" / "lib" +_HAS_REAL_LIB = _LIB_DIR.is_dir() and any( + path.suffix in (".so", ".dylib", ".dll") + for path in _LIB_DIR.iterdir() + if path.is_file() +) + +# --------------------------------------------------------------------------- +# Patch ctypes.CDLL and pathlib.Path.exists so that load_shared_library() +# succeeds without needing actual .so / .dylib / .dll files on disk. +# --------------------------------------------------------------------------- + +_mock_cdll = MagicMock() +_original_path_exists = pathlib.Path.exists +_original_ctypes_cdll = ctypes.CDLL + + +def _patched_path_exists(self: pathlib.Path) -> bool: + """Return True for paths that look like the llama / mtmd shared library.""" + name = str(self) + if "lib" in name and any(token in name for token in ("llama", "mtmd")): + return True + return _original_path_exists(self) + + +def _patched_cdll(path: str, **kwargs) -> MagicMock: + """Return a mock CDLL object instead of actually loading a shared library.""" + return _mock_cdll + + +if not _HAS_REAL_LIB: + pathlib.Path.exists = _patched_path_exists # type: ignore[method-assign] + ctypes.CDLL = _patched_cdll # type: ignore[assignment] + + # On Windows, load_shared_library() also registers the (possibly + # nonexistent) llama_cpp/lib directory in the DLL search path before + # loading. Tolerate a missing directory so imports succeed on a + # source-only checkout. + if hasattr(os, "add_dll_directory"): + _original_add_dll_directory = os.add_dll_directory + + def _patched_add_dll_directory(path: str): + try: + return _original_add_dll_directory(path) + except (FileNotFoundError, OSError): + return MagicMock() + + os.add_dll_directory = _patched_add_dll_directory # type: ignore[assignment] diff --git a/tests/test_gemma4_chat_format.py b/tests/test_gemma4_chat_format.py new file mode 100644 index 000000000..fd0897b0f --- /dev/null +++ b/tests/test_gemma4_chat_format.py @@ -0,0 +1,729 @@ +""" +Tests for the Gemma 4 chat format additions introduced in this PR: + - format_gemma4() registered formatter + - Gemma4ChatHandler (CHAT_FORMAT template + get_image_urls() static method) + - NanoLlavaChatHandler CHAT_FORMAT template + - MultimodalGemmaChatHandler CHAT_FORMAT template + - __init__.__version__ value + +All tests are pure-Python and do not require a compiled shared library or +an actual model file. +""" + +import warnings + +import jinja2 +import pytest + +import llama_cpp +import llama_cpp.llama_chat_format as llama_chat_format +from llama_cpp.llama_chat_format import ( + Gemma4ChatHandler, + MultimodalGemmaChatHandler, + NanoLlavaChatHandler, + format_gemma4, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _render(chat_format: str, messages, add_generation_prompt: bool = True) -> str: + """Render a Jinja2 CHAT_FORMAT string with the given messages.""" + env = jinja2.Environment(undefined=jinja2.Undefined, keep_trailing_newline=True) + tmpl = env.from_string(chat_format) + return tmpl.render(messages=messages, add_generation_prompt=add_generation_prompt) + + +# =========================================================================== +# format_gemma4() +# =========================================================================== + + +class TestFormatGemma4: + """Tests for the format_gemma4 registered chat formatter.""" + + def test_registered_as_gemma4(self): + """'gemma4' must be a registered chat completion handler.""" + # Access the registry directly; get_chat_formats() is not a public API + registry = llama_chat_format.LlamaChatCompletionHandlerRegistry() + assert "gemma4" in registry._chat_handlers + + def test_basic_user_message(self): + """Single user message produces correct BOS + turn tokens.""" + messages = [{"role": "user", "content": "Hello"}] + resp = format_gemma4(messages=messages) + assert resp.prompt.startswith("") + assert "<|turn>user\nHello\n" in resp.prompt + # Generation prompt is appended + assert resp.prompt.endswith("<|turn>model\n") + + def test_system_message_in_thought_channel(self): + """System messages are wrapped in <|channel>thought….""" + messages = [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "Hi"}, + ] + resp = format_gemma4(messages=messages) + assert ( + "<|channel>thought\nYou are a helpful assistant.\n" in resp.prompt + ) + # System message should NOT appear as a regular turn + assert "<|turn>system" not in resp.prompt + + def test_system_message_skipped_in_turns(self): + """System messages do not produce a <|turn>system… block.""" + messages = [ + {"role": "system", "content": "System prompt"}, + {"role": "user", "content": "Question"}, + ] + resp = format_gemma4(messages=messages) + assert "<|turn>system" not in resp.prompt + + def test_assistant_role_mapped_to_model(self): + """The 'assistant' role must be mapped to 'model' in the prompt.""" + messages = [ + {"role": "user", "content": "Hi"}, + {"role": "assistant", "content": "Hello there!"}, + {"role": "user", "content": "How are you?"}, + ] + resp = format_gemma4(messages=messages) + assert "<|turn>model\nHello there!\n" in resp.prompt + assert "<|turn>assistant" not in resp.prompt + + def test_multi_turn_conversation(self): + """Multi-turn conversation produces interleaved user/model turns.""" + messages = [ + {"role": "user", "content": "Question 1"}, + {"role": "assistant", "content": "Answer 1"}, + {"role": "user", "content": "Question 2"}, + ] + resp = format_gemma4(messages=messages) + assert "<|turn>user\nQuestion 1\n" in resp.prompt + assert "<|turn>model\nAnswer 1\n" in resp.prompt + assert "<|turn>user\nQuestion 2\n" in resp.prompt + + def test_stop_tokens(self): + """format_gemma4 returns the expected stop token list.""" + messages = [{"role": "user", "content": "Hi"}] + resp = format_gemma4(messages=messages) + assert isinstance(resp.stop, list) + assert "\n" in resp.stop + assert "" in resp.stop + + def test_empty_system_message_omitted_from_thought_channel(self): + """An empty string system message produces no thought channel block.""" + messages = [ + {"role": "system", "content": ""}, + {"role": "user", "content": "Hi"}, + ] + resp = format_gemma4(messages=messages) + # _get_system_message returns "" → falsy → no channel block inserted + assert "<|channel>thought" not in resp.prompt + + def test_no_system_message_no_thought_channel(self): + """Without a system message there should be no thought channel.""" + messages = [{"role": "user", "content": "Hi"}] + resp = format_gemma4(messages=messages) + assert "<|channel>thought" not in resp.prompt + + def test_reasoning_budget_accepted(self): + """reasoning_budget kwarg must not raise.""" + messages = [{"role": "user", "content": "Hi"}] + resp = format_gemma4(messages=messages, reasoning_budget=512) + assert resp.prompt # non-empty + + def test_bos_token_present(self): + """ is always the very first token.""" + messages = [{"role": "user", "content": "Test"}] + resp = format_gemma4(messages=messages) + assert resp.prompt.startswith("") + + def test_prompt_is_string(self): + """The returned prompt is always a plain string.""" + messages = [{"role": "user", "content": "Test"}] + resp = format_gemma4(messages=messages) + assert isinstance(resp.prompt, str) + + def test_user_role_unchanged(self): + """The 'user' role is kept as-is (not remapped).""" + messages = [{"role": "user", "content": "Hello"}] + resp = format_gemma4(messages=messages) + assert "<|turn>user\n" in resp.prompt + + +# =========================================================================== +# Gemma4ChatHandler.get_image_urls() +# =========================================================================== + + +class TestGemma4ChatHandlerGetImageUrls: + """Tests for the Gemma4ChatHandler.get_image_urls static method.""" + + def test_image_url_as_mapping(self): + """image_url given as a dict with a 'url' key is extracted correctly.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/img.jpg"}, + }, + {"type": "text", "text": "Describe this"}, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert urls == ["https://example.com/img.jpg"] + + def test_image_url_as_string(self): + """image_url given as a plain string is extracted directly.""" + messages = [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": "https://example.com/photo.png"}, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert urls == ["https://example.com/photo.png"] + + def test_input_audio_openai_schema(self): + """OpenAI 'input_audio' content type is converted to a data URI.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "input_audio", + "input_audio": {"format": "wav", "data": "AUDIO_BASE64"}, + }, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert len(urls) == 1 + assert urls[0] == "data:audio/wav;base64,AUDIO_BASE64" + + def test_audio_custom_schema(self): + """Custom 'audio' content type is also converted to a data URI.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "audio", + "audio": {"format": "mp3", "data": "MP3_BASE64"}, + }, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert len(urls) == 1 + assert urls[0] == "data:audio/mp3;base64,MP3_BASE64" + + def test_mixed_image_and_audio(self): + """Multiple media items in one message are all extracted in order.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/a.jpg"}, + }, + { + "type": "input_audio", + "input_audio": {"format": "wav", "data": "WAV64"}, + }, + {"type": "text", "text": "Describe and transcribe"}, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert len(urls) == 2 + assert "https://example.com/a.jpg" in urls + assert "data:audio/wav;base64,WAV64" in urls + + def test_no_media_returns_empty_list(self): + """Text-only messages produce an empty list.""" + messages = [ + {"role": "user", "content": "No media here"}, + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert urls == [] + + def test_non_user_roles_ignored(self): + """Only 'user' role messages are scanned for media.""" + messages = [ + { + "role": "assistant", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/x.jpg"}, + }, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert urls == [] + + def test_audio_default_format_wav(self): + """Missing 'format' key in audio dict defaults to 'wav'.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "input_audio", + "input_audio": {"data": "ABC"}, + # no 'format' key + }, + ], + } + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + # The default format falls back to "wav" per the implementation + assert len(urls) == 1 + assert urls[0].startswith("data:audio/wav;base64,") + + def test_multiple_user_messages(self): + """Media from multiple user messages are all collected.""" + messages = [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/1.jpg"}, + }, + ], + }, + {"role": "assistant", "content": "OK"}, + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/2.jpg"}, + }, + ], + }, + ] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert len(urls) == 2 + + def test_string_content_user_message_ignored(self): + """String-content user messages (no media parts) return empty list.""" + messages = [{"role": "user", "content": "plain text"}] + urls = Gemma4ChatHandler.get_image_urls(messages) + assert urls == [] + + +# =========================================================================== +# Gemma4ChatHandler.CHAT_FORMAT template +# =========================================================================== + + +class TestGemma4ChatHandlerTemplate: + """Tests for the Gemma4ChatHandler Jinja2 CHAT_FORMAT string.""" + + def test_system_message_in_thought_channel(self): + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + {"role": "system", "content": "Be concise."}, + {"role": "user", "content": "Hi"}, + ], + ) + assert "<|channel>thought\nBe concise.\n" in prompt + + def test_user_turn_tokens(self): + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hello"}], + ) + assert "<|turn>user\nHello\n" in prompt + + def test_assistant_turn_tokens(self): + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + {"role": "user", "content": "Q"}, + {"role": "assistant", "content": "A"}, + ], + add_generation_prompt=False, + ) + assert "<|turn>model\nA\n" in prompt + + def test_generation_prompt_appended(self): + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hi"}], + add_generation_prompt=True, + ) + assert prompt.endswith("<|turn>model\n") + + def test_no_generation_prompt_when_false(self): + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hi"}], + add_generation_prompt=False, + ) + assert not prompt.endswith("<|turn>model\n") + + def test_image_url_mapping_embedded(self): + """image_url as a mapping (dict with 'url') is embedded in the prompt.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/pic.jpg"}, + }, + {"type": "text", "text": "Describe"}, + ], + } + ], + ) + assert "https://example.com/pic.jpg" in prompt + assert "Describe" in prompt + + def test_image_url_string_embedded(self): + """image_url as a plain string is embedded in the prompt.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": "https://example.com/img.png", + }, + {"type": "text", "text": "Caption"}, + ], + } + ], + ) + assert "https://example.com/img.png" in prompt + + def test_input_audio_data_uri_embedded(self): + """input_audio content is serialised as a data: URI.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "input_audio", + "input_audio": {"format": "wav", "data": "TESTDATA"}, + }, + {"type": "text", "text": "Transcribe"}, + ], + } + ], + ) + assert "data:audio/wav;base64,TESTDATA" in prompt + + def test_audio_custom_schema_data_uri_embedded(self): + """audio (custom schema) content is serialised as a data: URI.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "audio", + "audio": {"format": "mp3", "data": "MP3DATA"}, + }, + {"type": "text", "text": "Listen"}, + ], + } + ], + ) + assert "data:audio/mp3;base64,MP3DATA" in prompt + + def test_tool_call_format(self): + """Tool-call messages produce the expected <|tool_call> block.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "function": { + "name": "get_weather", + "arguments": '{"location": "Paris"}', + } + } + ], + } + ], + add_generation_prompt=False, + ) + assert "<|tool_call>call:get_weather" in prompt + + def test_tool_response_format(self): + """Tool-response messages produce the expected <|tool_response> block.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "tool", + "name": "get_weather", + "content": "Sunny, 22°C", + } + ], + add_generation_prompt=False, + ) + assert "<|tool_response>response:get_weather" in prompt + assert "Sunny, 22°C" in prompt + + def test_media_appears_before_text_in_user_message(self): + """Media tokens must be emitted before text tokens in the prompt.""" + prompt = _render( + Gemma4ChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "TEXT_PART"}, + { + "type": "image_url", + "image_url": {"url": "https://example.com/a.jpg"}, + }, + ], + } + ], + ) + img_pos = prompt.index("https://example.com/a.jpg") + txt_pos = prompt.index("TEXT_PART") + assert img_pos < txt_pos, "image URL should appear before text in the prompt" + + +# =========================================================================== +# Gemma4ChatHandler thread-safety attribute +# =========================================================================== + + +class TestGemma4ChatHandlerAttributes: + """Tests for class-level attributes of Gemma4ChatHandler.""" + + def test_format_lock_is_thread_lock(self): + """_format_lock must be a threading.Lock or RLock instance.""" + import threading + + assert isinstance(Gemma4ChatHandler._format_lock, type(threading.Lock())) + + def test_default_system_message_is_none(self): + assert Gemma4ChatHandler.DEFAULT_SYSTEM_MESSAGE is None + + def test_chat_format_is_string(self): + assert isinstance(Gemma4ChatHandler.CHAT_FORMAT, str) + + +# =========================================================================== +# NanoLlavaChatHandler CHAT_FORMAT template +# =========================================================================== + + +class TestNanoLlavaChatHandlerTemplate: + """Tests for the NanoLlavaChatHandler Jinja2 CHAT_FORMAT string.""" + + def test_default_system_message(self): + assert NanoLlavaChatHandler.DEFAULT_SYSTEM_MESSAGE == "Answer the question" + + def test_basic_user_message(self): + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "What is 2+2?"}], + ) + assert "<|im_start|>user\n" in prompt + assert "What is 2+2?" in prompt + assert "<|im_end|>" in prompt + + def test_generation_prompt(self): + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hi"}], + add_generation_prompt=True, + ) + assert prompt.endswith("<|im_start|>assistant\n") + + def test_system_message_in_chatml_block(self): + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [ + {"role": "system", "content": "Answer the question"}, + {"role": "user", "content": "Hi"}, + ], + ) + assert "<|im_start|>system\n" in prompt + assert "Answer the question" in prompt + + def test_image_url_string_before_text(self): + """Image URL (string) is emitted before text in the user block.""" + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": "https://img.example.com/x.jpg", + }, + {"type": "text", "text": "Describe"}, + ], + } + ], + ) + assert "https://img.example.com/x.jpg" in prompt + img_pos = prompt.index("https://img.example.com/x.jpg") + txt_pos = prompt.index("Describe") + assert img_pos < txt_pos + + def test_image_url_mapping_before_text(self): + """Image URL (mapping with 'url' key) is emitted before text.""" + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://img.example.com/y.jpg"}, + }, + {"type": "text", "text": "Caption"}, + ], + } + ], + ) + assert "https://img.example.com/y.jpg" in prompt + img_pos = prompt.index("https://img.example.com/y.jpg") + txt_pos = prompt.index("Caption") + assert img_pos < txt_pos + + def test_assistant_message(self): + prompt = _render( + NanoLlavaChatHandler.CHAT_FORMAT, + [ + {"role": "user", "content": "Q"}, + {"role": "assistant", "content": "A"}, + ], + add_generation_prompt=False, + ) + assert "<|im_start|>assistant\nA<|im_end|>" in prompt + + +# =========================================================================== +# MultimodalGemmaChatHandler CHAT_FORMAT template +# =========================================================================== + + +class TestMultimodalGemmaChatHandlerTemplate: + """Tests for the MultimodalGemmaChatHandler CHAT_FORMAT template.""" + + def test_default_system_message_is_none(self): + assert MultimodalGemmaChatHandler.DEFAULT_SYSTEM_MESSAGE is None + + def test_user_turn_tokens(self): + prompt = _render( + MultimodalGemmaChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hello"}], + ) + assert "user\n" in prompt + assert "\n" in prompt + + def test_assistant_turn_tokens(self): + prompt = _render( + MultimodalGemmaChatHandler.CHAT_FORMAT, + [ + {"role": "user", "content": "Q"}, + {"role": "assistant", "content": "A"}, + ], + add_generation_prompt=False, + ) + assert "model\nA\n" in prompt + + def test_generation_prompt(self): + prompt = _render( + MultimodalGemmaChatHandler.CHAT_FORMAT, + [{"role": "user", "content": "Hi"}], + add_generation_prompt=True, + ) + assert prompt.endswith("model\n") + + def test_image_url_mapping_embedded(self): + prompt = _render( + MultimodalGemmaChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "https://example.com/gemma_img.jpg"}, + }, + {"type": "text", "text": "Describe"}, + ], + } + ], + ) + assert "https://example.com/gemma_img.jpg" in prompt + + def test_image_url_string_embedded(self): + prompt = _render( + MultimodalGemmaChatHandler.CHAT_FORMAT, + [ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": "https://example.com/pic.jpg", + }, + {"type": "text", "text": "Look"}, + ], + } + ], + ) + assert "https://example.com/pic.jpg" in prompt + + +# =========================================================================== +# __init__.__version__ +# =========================================================================== + + +class TestPackageVersion: + """Tests for the package version string exposed via __init__.py.""" + + def test_version_is_string(self): + assert isinstance(llama_cpp.__version__, str) + + def test_version_non_empty(self): + assert llama_cpp.__version__ + + def test_version_format(self): + """Version must follow a major.minor.patch pattern.""" + parts = llama_cpp.__version__.split(".") + assert len(parts) == 3, ( + f"Expected 3 version parts, got: {llama_cpp.__version__!r}" + ) + for part in parts: + assert part.isdigit(), f"Non-numeric version component: {part!r}" diff --git a/tests/test_mtmd_cpp.py b/tests/test_mtmd_cpp.py new file mode 100644 index 000000000..5450737fc --- /dev/null +++ b/tests/test_mtmd_cpp.py @@ -0,0 +1,332 @@ +""" +Tests for llama_cpp/mtmd_cpp.py changes introduced in this PR: + + - mtmd_caps Structure (inp_vision, inp_audio fields) + - mtmd_input_text Structure (_fields_ layout) + - mtmd_decoder_pos Structure (t, x, y, z fields) + - mtmd_context_params Structure (image_marker AND new media_marker field) + - MTMD_INPUT_CHUNK_TYPE_* integer constants + - mtmd_get_audio_sample_rate function signature + - mtmd_get_audio_bitrate() deprecated wrapper: + - emits DeprecationWarning + - delegates to mtmd_get_audio_sample_rate + - returns its return value unchanged + +The shared library is mocked out by tests/conftest.py so no compiled .so +file is required. +""" + +import ctypes +import warnings +from unittest.mock import MagicMock, patch + +import pytest + +import llama_cpp.mtmd_cpp as mtmd_cpp + + +# =========================================================================== +# Integer constants +# =========================================================================== + + +class TestMtmdChunkTypeConstants: + """The MTMD_INPUT_CHUNK_TYPE_* constants must have the correct integer values.""" + + def test_text_constant(self): + assert mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_TEXT == 0 + + def test_image_constant(self): + assert mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_IMAGE == 1 + + def test_audio_constant(self): + assert mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_AUDIO == 2 + + def test_all_constants_are_distinct(self): + values = [ + mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_TEXT, + mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_IMAGE, + mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_AUDIO, + ] + assert len(set(values)) == len(values) + + def test_constants_are_integers(self): + assert isinstance(mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_TEXT, int) + assert isinstance(mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_IMAGE, int) + assert isinstance(mtmd_cpp.MTMD_INPUT_CHUNK_TYPE_AUDIO, int) + + +# =========================================================================== +# mtmd_caps Structure +# =========================================================================== + + +class TestMtmdCapsStructure: + """Tests for the mtmd_caps ctypes Structure.""" + + def test_is_ctypes_structure(self): + assert issubclass(mtmd_cpp.mtmd_caps, ctypes.Structure) + + def test_has_inp_vision_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_caps._fields_] + assert "inp_vision" in field_names + + def test_has_inp_audio_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_caps._fields_] + assert "inp_audio" in field_names + + def test_exactly_two_fields(self): + assert len(mtmd_cpp.mtmd_caps._fields_) == 2 + + def test_inp_vision_is_cbool(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_caps._fields_} + assert field_map["inp_vision"] is ctypes.c_bool + + def test_inp_audio_is_cbool(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_caps._fields_} + assert field_map["inp_audio"] is ctypes.c_bool + + def test_instance_creation(self): + """mtmd_caps can be instantiated without arguments.""" + caps = mtmd_cpp.mtmd_caps() + # Fields default to False + assert isinstance(caps.inp_vision, bool) + assert isinstance(caps.inp_audio, bool) + + def test_field_assignment(self): + caps = mtmd_cpp.mtmd_caps() + caps.inp_vision = True + caps.inp_audio = True + assert caps.inp_vision is True + assert caps.inp_audio is True + + +# =========================================================================== +# mtmd_input_text Structure +# =========================================================================== + + +class TestMtmdInputTextStructure: + """Tests for the mtmd_input_text ctypes Structure.""" + + def test_is_ctypes_structure(self): + assert issubclass(mtmd_cpp.mtmd_input_text, ctypes.Structure) + + def test_has_text_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_input_text._fields_] + assert "text" in field_names + + def test_has_add_special_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_input_text._fields_] + assert "add_special" in field_names + + def test_has_parse_special_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_input_text._fields_] + assert "parse_special" in field_names + + def test_exactly_three_fields(self): + assert len(mtmd_cpp.mtmd_input_text._fields_) == 3 + + def test_text_is_char_p(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_input_text._fields_} + assert field_map["text"] is ctypes.c_char_p + + def test_add_special_is_cbool(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_input_text._fields_} + assert field_map["add_special"] is ctypes.c_bool + + def test_parse_special_is_cbool(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_input_text._fields_} + assert field_map["parse_special"] is ctypes.c_bool + + def test_instance_creation(self): + it = mtmd_cpp.mtmd_input_text() + it.text = b"Hello" + it.add_special = True + it.parse_special = False + assert it.text == b"Hello" + assert it.add_special is True + assert it.parse_special is False + + +# =========================================================================== +# mtmd_decoder_pos Structure +# =========================================================================== + + +class TestMtmdDecoderPosStructure: + """Tests for the mtmd_decoder_pos ctypes Structure.""" + + def test_is_ctypes_structure(self): + assert issubclass(mtmd_cpp.mtmd_decoder_pos, ctypes.Structure) + + def test_has_t_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_decoder_pos._fields_] + assert "t" in field_names + + def test_has_x_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_decoder_pos._fields_] + assert "x" in field_names + + def test_has_y_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_decoder_pos._fields_] + assert "y" in field_names + + def test_has_z_field(self): + field_names = [f[0] for f in mtmd_cpp.mtmd_decoder_pos._fields_] + assert "z" in field_names + + def test_exactly_four_fields(self): + assert len(mtmd_cpp.mtmd_decoder_pos._fields_) == 4 + + def test_all_fields_are_uint32(self): + for _, ctype in mtmd_cpp.mtmd_decoder_pos._fields_: + assert ctype is ctypes.c_uint32 + + def test_instance_assignment(self): + dp = mtmd_cpp.mtmd_decoder_pos() + dp.t = 1 + dp.x = 2 + dp.y = 3 + dp.z = 4 + assert dp.t == 1 + assert dp.x == 2 + assert dp.y == 3 + assert dp.z == 4 + + +# =========================================================================== +# mtmd_context_params Structure +# =========================================================================== + + +class TestMtmdContextParamsStructure: + """Tests for the mtmd_context_params ctypes Structure, specifically the new + media_marker field and the kept image_marker field.""" + + def _field_names(self): + return [f[0] for f in mtmd_cpp.mtmd_context_params._fields_] + + def test_is_ctypes_structure(self): + assert issubclass(mtmd_cpp.mtmd_context_params, ctypes.Structure) + + def test_has_image_marker_field(self): + """Deprecated image_marker field must still be present for compatibility.""" + assert "image_marker" in self._field_names() + + def test_has_media_marker_field(self): + """New media_marker field must be present.""" + assert "media_marker" in self._field_names() + + def test_image_marker_is_char_p(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_context_params._fields_} + assert field_map["image_marker"] is ctypes.c_char_p + + def test_media_marker_is_char_p(self): + field_map = {f[0]: f[1] for f in mtmd_cpp.mtmd_context_params._fields_} + assert field_map["media_marker"] is ctypes.c_char_p + + def test_has_use_gpu_field(self): + assert "use_gpu" in self._field_names() + + def test_has_n_threads_field(self): + assert "n_threads" in self._field_names() + + def test_has_warmup_field(self): + assert "warmup" in self._field_names() + + def test_has_flash_attn_type_field(self): + assert "flash_attn_type" in self._field_names() + + +# =========================================================================== +# mtmd_get_audio_bitrate() deprecated wrapper +# =========================================================================== + + +class TestMtmdGetAudioBitrateDeprecated: + """Tests for the mtmd_get_audio_bitrate() DeprecationWarning wrapper.""" + + def test_emits_deprecation_warning(self): + """Calling mtmd_get_audio_bitrate must issue a DeprecationWarning.""" + fake_ctx = object() + with patch.object(mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=16000): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mtmd_cpp.mtmd_get_audio_bitrate(fake_ctx) + + assert len(caught) >= 1 + categories = [w.category for w in caught] + assert DeprecationWarning in categories + + def test_warning_message_mentions_new_name(self): + """The deprecation message should name the replacement function.""" + with patch.object(mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=0): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + mtmd_cpp.mtmd_get_audio_bitrate(None) + + dep_warnings = [w for w in caught if issubclass(w.category, DeprecationWarning)] + assert dep_warnings, "Expected at least one DeprecationWarning" + msg = str(dep_warnings[0].message) + assert "mtmd_get_audio_sample_rate" in msg + + def test_delegates_to_sample_rate_function(self): + """mtmd_get_audio_bitrate must call mtmd_get_audio_sample_rate exactly once.""" + fake_ctx = object() + with patch.object( + mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=44100 + ) as mock_fn: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + mtmd_cpp.mtmd_get_audio_bitrate(fake_ctx) + + mock_fn.assert_called_once_with(fake_ctx) + + def test_returns_sample_rate_value(self): + """The return value of mtmd_get_audio_bitrate must equal the sample rate.""" + expected = 22050 + with patch.object( + mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=expected + ): + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + result = mtmd_cpp.mtmd_get_audio_bitrate(None) + + assert result == expected + + def test_passes_context_argument_through(self): + """The ctx argument is forwarded unchanged to mtmd_get_audio_sample_rate.""" + sentinel_ctx = object() + with patch.object( + mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=8000 + ) as mock_fn: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + mtmd_cpp.mtmd_get_audio_bitrate(sentinel_ctx) + + args, _ = mock_fn.call_args + assert args[0] is sentinel_ctx + + def test_minus_one_propagated_for_no_audio(self): + """A return value of -1 (no audio support) is passed through intact.""" + with patch.object(mtmd_cpp, "mtmd_get_audio_sample_rate", return_value=-1): + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + result = mtmd_cpp.mtmd_get_audio_bitrate(None) + + assert result == -1 + + +# =========================================================================== +# mtmd_get_audio_sample_rate function exists +# =========================================================================== + + +class TestMtmdGetAudioSampleRate: + """Smoke-tests for the new mtmd_get_audio_sample_rate function.""" + + def test_function_exists(self): + """mtmd_get_audio_sample_rate must be defined in the module.""" + assert hasattr(mtmd_cpp, "mtmd_get_audio_sample_rate") + assert callable(mtmd_cpp.mtmd_get_audio_sample_rate)