refactor(test): split test_server_patches, unify _real_apple_gpu, drop slow marker and module aliases

gmlx/serve/server.py

@@ -2,7 +2,7 @@
 
 Composes the gmlx loader bridge (:func:`server_bridge_vlm.install_gguf_server_bridge`),
 the multi-model residency pool (:func:`residency.install_gguf_residency_pool`),
-the config-driven HTTP surface (:func:`server_patches.install_server_patches`),
+the config-driven HTTP surface (:func:`gmlx.serve.patches.install_server_patches`),
 and mlx-vlm's FastAPI app + ``BatchGenerator`` continuous-batching engine into one
 GGUF-only HTTP server - **text**, **VLM** (LLM GGUF + float ``mmproj``), and
 **speculative/MTP** models.
@@ -92,7 +92,7 @@ def _import_serving():
     bare ModuleNotFoundError."""
     try:
         from . import residency  # noqa: F401
-        from . import patches as server_patches  # noqa: F401
+        from . import patches  # noqa: F401
         from . import bridge_vlm as server_bridge_vlm  # noqa: F401
     except ImportError as exc:
         root = (exc.name or "").split(".")[0]
@@ -2052,7 +2052,7 @@ def _on_sighup(_sig, _frame):
 
     import uvicorn
 
-    from . import patches as server_patches
+    from . import patches
 
     # Preload off the startup path: mlx-vlm's lifespan loads MLX_VLM_PRELOAD_MODEL
     # synchronously *before* the port accepts connections, so a big model makes the
@@ -2068,7 +2068,7 @@ def _on_sighup(_sig, _frame):
               if m != preload]
     if preload or extras:
         os.environ.pop("MLX_VLM_PRELOAD_MODEL", None)
-        server_patches.spawn_preload_warm(preload, extras)
+        patches.spawn_preload_warm(preload, extras)
         if extras:
             print(f"[server] preload: warming {', '.join(extras)} in background")
 
@@ -2087,7 +2087,7 @@ def _on_sighup(_sig, _frame):
                 # gmlx/mlx_vlm loggers, so the flag governs the whole server
                 # while the timestamped formatters and noise filters stay.
                 log_level=getattr(a, "log_level", None),
-                log_config=server_patches.uvicorn_log_config(
+                log_config=patches.uvicorn_log_config(
                     getattr(a, "log_level", None)))
     return 0
 

pyproject.toml

@@ -155,7 +155,6 @@ gmlx = ["_vendor/misaki/LICENSE", "_vendor/misaki/data/*",
 testpaths = ["tests"]
 markers = [
     "integration: end-to-end test needing a real GGUF (and, for parity, a llama.cpp binary); skipped unless KQUANT_TEST_GGUF_DIR is set",
-    "slow: long-running (large prompts / long generation)",
     "needs_kvarn_ops: dispatches kvarn Metal kernels; skipped when the mlx-kquant ops probe fails (CPU device, op-less wheel)",
     "needs_kvarn_row_ends: dispatches the decode kernels with per-row ends; skipped unless the installed mlx-kquant is 0.4.9 or later",
 ]

tests/assistant/test_assistant_memory.py

@@ -94,12 +94,12 @@ def test_prune_time_corruption_also_quarantined(tmp_path, monkeypatch):
     # open-time prune; construction must still self-heal, not abort.
     import sqlite3
 
-    import gmlx.assistant.memory as talk_memory
+    from gmlx.assistant import memory
 
     def bad_prune(self):
         raise sqlite3.DatabaseError("database disk image is malformed")
 
-    monkeypatch.setattr(talk_memory.MemoryStore, "_prune", bad_prune)
+    monkeypatch.setattr(memory.MemoryStore, "_prune", bad_prune)
     warned = []
     m = _store(tmp_path, warn=warned.append)
     m.remember("I like green tea", "Noted.")
@@ -296,8 +296,8 @@ def counting_execute(sql, *a):
 
 
 def test_cache_capacity_doubles(tmp_path, monkeypatch):
-    import gmlx.assistant.memory as talk_memory
-    monkeypatch.setattr(talk_memory, "_INITIAL_CAP", 1)
+    from gmlx.assistant import memory
+    monkeypatch.setattr(memory, "_INITIAL_CAP", 1)
     m = _store(tmp_path)
     m.remember("green tea daily", "")
     m.recall("tea?")                              # cache: capacity 1, n 1
@@ -311,8 +311,8 @@ def test_cache_capacity_doubles(tmp_path, monkeypatch):
 def test_recall_correct_after_growth(tmp_path, monkeypatch):
     # The promise behind the doubling rule: recall stays CORRECT after the
     # cache matrix has grown - rows appended through growth are recallable.
-    import gmlx.assistant.memory as talk_memory
-    monkeypatch.setattr(talk_memory, "_INITIAL_CAP", 1)
+    from gmlx.assistant import memory
+    monkeypatch.setattr(memory, "_INITIAL_CAP", 1)
     m = _store(tmp_path)
     m.remember("I like green tea", "")
     m.recall("tea?")                              # cache built at capacity 1
@@ -433,7 +433,7 @@ def test_old_schema_gains_recalled_column(tmp_path):
 
 
 def test_make_extractor_prompts_and_parses(monkeypatch):
-    import gmlx.assistant.memory as talk_memory
+    from gmlx.assistant import memory
     seen = {}
 
     def fake_stream(base_url, *, model, messages, max_tokens,
@@ -444,8 +444,8 @@ def fake_stream(base_url, *, model, messages, max_tokens,
         yield {"content": "NONE\n2) rides a red bike"}
         yield {"_finish": "stop"}
 
-    monkeypatch.setattr(talk_memory.talk_client, "stream_chat", fake_stream)
-    extract = talk_memory.make_extractor("http://h:1/v1", "m1")
+    monkeypatch.setattr(memory.talk_client, "stream_chat", fake_stream)
+    extract = memory.make_extractor("http://h:1/v1", "m1")
     assert extract("I like tea", "noted") == ["likes green tea",
                                               "rides a red bike"]
     assert seen["model"] == "m1"

tests/conftest.py

@@ -32,6 +32,8 @@
         _p = str(_sub)
         if _p not in sys.path:
             sys.path.insert(0, _p)
+if str(_TESTS_DIR) not in sys.path:
+    sys.path.insert(0, str(_TESTS_DIR))
 
 
 def pytest_configure(config):

tests/gen/test_batch_parity.py

@@ -90,7 +90,7 @@
 
 GREEDY = lambda x: mx.argmax(x, axis=-1)
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 
 # helpers

tests/gen/test_hy_v4_long_context.py

@@ -51,7 +51,7 @@
 # not the total depth.
 ABOVE_BOUNDARY = int(os.environ.get("GMLX_HY4_LONGCTX_TOKENS", "4096"))
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 
 def _greedy_text(model, tok, ids, n):

tests/gen/test_long_context.py

@@ -111,7 +111,7 @@
 _NEEDLE_QUERY = ("\nAs noted near the beginning of this document, the vault "
                  "access code written in the expedition logbook is")
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 
 # helpers

tests/helpers.py

@@ -0,0 +1,18 @@
+"""Shared test helpers importable from any test directory."""
+
+import os
+
+
+def _real_apple_gpu() -> bool:
+    """True on a real Apple GPU; False under KQUANT_FORCE_CPU, in a VM
+    (Paravirtual device), or when mlx cannot report a device. Import-time
+    callable so module-level pytestmark can use it (not a fixture)."""
+    if os.environ.get("KQUANT_FORCE_CPU"):
+        return False
+    try:
+        import mlx.core as mx
+
+        name = str(mx.device_info().get("device_name", ""))
+    except Exception:
+        return False
+    return "Apple" in name and "Paravirtual" not in name

tests/load/test_vlm_integration.py

@@ -21,7 +21,7 @@
 
 pytest.importorskip("mlx_vlm")
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 # A 30B-class VLM forward is a GPU workload; the forced-CPU stream keeps to
 # logic tests.

tests/models/test_qwen4exp_ple.py

@@ -4,12 +4,13 @@
 
 import dataclasses
 import os
-import re
 
 import mlx.core as mx
 import numpy as np
 import pytest
 
+from helpers import _real_apple_gpu
+
 import gmlx.models.qwen4_exp.model as q4
 from gmlx.models.qwen4_exp.model import ModelArgs, PLEEmbedding
 
@@ -33,16 +34,6 @@
 )
 
 
-def _real_apple_gpu() -> bool:
-    if os.environ.get("KQUANT_FORCE_CPU"):
-        return False
-    try:
-        name = str(mx.device_info().get("device_name", ""))
-    except Exception:
-        return False
-    return bool(re.search(r"Apple M\d", name))
-
-
 def _ple():
     flds = {f.name for f in dataclasses.fields(ModelArgs)}
     return PLEEmbedding(ModelArgs(

tests/models/test_qwen4exp_prefill_dispatch.py

@@ -5,12 +5,12 @@
 
 from __future__ import annotations
 
-import os
-import re
 
 import mlx.core as mx
 import pytest
 
+from helpers import _real_apple_gpu
+
 import gmlx.models.qwen4_exp.model as q4
 from gmlx.models.qwen4_exp.model import (
     Attention,
@@ -21,16 +21,6 @@
 )
 
 
-def _real_apple_gpu() -> bool:
-    if os.environ.get("KQUANT_FORCE_CPU"):
-        return False
-    try:
-        name = str(mx.device_info().get("device_name", ""))
-    except Exception:
-        return False
-    return bool(re.search(r"Apple M\d", name))
-
-
 gpu_only = pytest.mark.skipif(
     not _real_apple_gpu(), reason="needs a real Apple GPU")
 

tests/models/test_vlm_mtp_integration.py

@@ -22,7 +22,7 @@
 
 pytest.importorskip("mlx_vlm")
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 _NEEDS_GPU = pytest.mark.skipif(
     bool(os.environ.get("KQUANT_FORCE_CPU")),

tests/models/test_vlm_plain_image_integration.py

@@ -22,7 +22,7 @@
 
 pytest.importorskip("mlx_vlm")
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = pytest.mark.integration
 
 _NEEDS_GPU = pytest.mark.skipif(
     bool(os.environ.get("KQUANT_FORCE_CPU")),

tests/serve/conftest.py

@@ -0,0 +1,106 @@
+"""Shared serve-suite fixtures.
+
+The seam snapshot/restore fixture guards every test in this directory;
+it is a no-op when mlx-vlm is not installed."""
+from __future__ import annotations
+
+import importlib
+import sys
+
+import pytest
+
+try:
+    import gmlx.serve.bridge_vlm as serving
+    from gmlx.serve.patches import _common as sp_common
+    from gmlx.serve.patches import hardening as sp_hardening
+    _APP = importlib.import_module("mlx_vlm.server.app")
+    _UTILS = importlib.import_module("mlx_vlm.utils")
+    _PKG = importlib.import_module("mlx_vlm.server")
+except ImportError:
+    _APP = None
+
+
+_PATCH_MODULES = {
+    "test_server_patches",
+    "test_patches_chat_behavior",
+    "test_patches_routes",
+    "test_patches_sampling",
+}
+
+
+@pytest.fixture(autouse=True)
+def _restore_mlxvlm(request):
+    if _APP is None or request.module.__name__ not in _PATCH_MODULES:
+        yield
+        return
+    """Snapshot every mlx-vlm seam these patches mutate, restore after each test."""
+    fastapi_app = _APP.app
+    saved = {
+        "build_gen_args": _APP._build_gen_args,
+        "snapshot": _APP._server_runtime_snapshot,
+        "get_model_path": _UTILS.get_model_path,
+        "routes": sp_common._snapshot_routes(fastapi_app),
+        "handlers": dict(fastapi_app.exception_handlers),
+        "deps": getattr(getattr(_APP, "_protocol_deps", None), "build_gen_args", None),
+        "pool": getattr(_PKG, "_kq_residency_pool", None),
+        # Middleware installs (auth / host guard) append to user_middleware and
+        # set app.state flags; CORS hardening mutates a Middleware's kwargs in
+        # place - snapshot all three or one test's auth poisons the rest.
+        "middleware": list(fastapi_app.user_middleware),
+        "mw_kwargs": [(m, dict(getattr(m, "kwargs", {}) or {}))
+                      for m in fastapi_app.user_middleware],
+    }
+    openai = sys.modules.get("mlx_vlm.server.openai")
+    anthropic = sys.modules.get("mlx_vlm.server.anthropic")
+    saved["openai_bga"] = getattr(openai, "_build_gen_args", None)
+    saved["anthropic_bga"] = getattr(anthropic, "_build_gen_args", None)
+    apc = sys.modules.get("mlx_vlm.apc") or importlib.import_module("mlx_vlm.apc")
+    saved["apc_harvest"] = apc.harvest_blocks_from_batch_cache
+    saved["apc_lone_flag"] = getattr(apc, "_kq_lone_harvest", False)
+    gen = importlib.import_module("mlx_vlm.server.generation")
+    saved["to_template_kwargs"] = gen.GenerationArguments.to_template_kwargs
+    pu = importlib.import_module("mlx_vlm.prompt_utils")
+    saved["get_chat_template"] = pu.get_chat_template
+    saved["make_sampler"] = gen.ResponseGenerator._make_sampler
+    saved["make_tb_criteria"] = gen.ResponseGenerator._make_thinking_budget_criteria
+    schemas = importlib.import_module("mlx_vlm.server.schemas")
+    saved["stream_chunk_dump"] = schemas.ChatStreamChunk.model_dump_json
+    saved["stopping_call"] = _UTILS.StoppingCriteria.__call__
+    serving.clear_resolved_models()
+    yield
+    _UTILS.StoppingCriteria.__call__ = saved["stopping_call"]
+    apc.harvest_blocks_from_batch_cache = saved["apc_harvest"]
+    apc._kq_lone_harvest = saved["apc_lone_flag"]
+    gen.GenerationArguments.to_template_kwargs = saved["to_template_kwargs"]
+    pu.get_chat_template = saved["get_chat_template"]
+    gen.ResponseGenerator._make_sampler = saved["make_sampler"]
+    gen.ResponseGenerator._make_thinking_budget_criteria = saved["make_tb_criteria"]
+    schemas.ChatStreamChunk.model_dump_json = saved["stream_chunk_dump"]
+    _APP._build_gen_args = saved["build_gen_args"]
+    _APP._server_runtime_snapshot = saved["snapshot"]
+    _UTILS.get_model_path = saved["get_model_path"]
+    sp_common._restore_routes(fastapi_app, saved["routes"])
+    fastapi_app.exception_handlers.clear()
+    fastapi_app.exception_handlers.update(saved["handlers"])
+    if getattr(_APP, "_protocol_deps", None) is not None and saved["deps"] is not None:
+        _APP._protocol_deps.build_gen_args = saved["deps"]
+    if openai is not None:
+        openai._build_gen_args = saved["openai_bga"]
+    if anthropic is not None:
+        anthropic._build_gen_args = saved["anthropic_bga"]
+    if saved["pool"] is None:
+        if hasattr(_PKG, "_kq_residency_pool"):
+            delattr(_PKG, "_kq_residency_pool")
+    else:
+        _PKG._kq_residency_pool = saved["pool"]
+    fastapi_app.user_middleware[:] = saved["middleware"]
+    for m, kw in saved["mw_kwargs"]:
+        if getattr(m, "kwargs", None) is not None:
+            m.kwargs.clear()
+            m.kwargs.update(kw)
+    fastapi_app.middleware_stack = None       # force a rebuild from the restored list
+    for flag in (sp_hardening._AUTH_FLAG, sp_hardening._HOST_GUARD_FLAG,
+                 sp_hardening._JSON_CT_FLAG):
+        if hasattr(fastapi_app.state, flag):
+            delattr(fastapi_app.state, flag)
+    serving.clear_resolved_models()