File size: 5,004 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Tests for `mindxtrain.deploy.api_client` — particularly the new
`swap_mindx_fallback_model` which closes the publish-side of the loop with
the mindX PATCH /v1/config/fallback-model endpoint shipped earlier.
We mock httpx with a MockTransport so the tests run offline and assert the
exact request shape (URL, method, body, bearer header) without any network.
"""
from __future__ import annotations
import json
import httpx
import pytest
from pydantic import ValidationError
from mindxtrain.deploy import api_client
from mindxtrain.deploy.api_client import (
MindXFallbackSwap,
swap_mindx_fallback_model,
)
def _capturing_transport(*, recorder: dict, response_body: dict, status: int = 200):
"""Build a httpx MockTransport that records the request and returns `response_body`."""
def handler(request: httpx.Request) -> httpx.Response:
recorder["method"] = request.method
recorder["url"] = str(request.url)
recorder["headers"] = dict(request.headers)
recorder["body"] = json.loads(request.content.decode("utf-8")) if request.content else None
return httpx.Response(status, json=response_body)
return httpx.MockTransport(handler)
def test_swap_payload_validates():
"""MindXFallbackSwap is `extra=forbid` and rejects an empty model name."""
with pytest.raises(ValidationError):
MindXFallbackSwap(model="") # min_length=1 violation
p = MindXFallbackSwap(model="pythai/mindx-fallback-qwen3-1.5b")
assert p.provider == "vllm"
assert p.model == "pythai/mindx-fallback-qwen3-1.5b"
def test_swap_sends_patch_with_correct_body(monkeypatch):
recorder: dict = {}
transport = _capturing_transport(
recorder=recorder,
response_body={
"success": True,
"provider": "vllm",
"previous": "Qwen/Qwen3-0.6B",
"current": "pythai/mindx-fallback-qwen3-1.5b",
"config_file": "models/vllm.yaml",
},
)
# Patch httpx.Client so the function picks up our transport.
orig_client = httpx.Client
def _client(*args, **kwargs):
kwargs["transport"] = transport
return orig_client(*args, **kwargs)
monkeypatch.setattr(httpx, "Client", _client)
monkeypatch.setenv("MINDXTRAIN_API_BASE_URL", "https://mindx.pythai.net")
monkeypatch.delenv("MINDXTRAIN_API_KEY", raising=False)
out = swap_mindx_fallback_model(
provider="vllm",
model="pythai/mindx-fallback-qwen3-1.5b",
)
assert recorder["method"] == "PATCH"
assert recorder["url"] == "https://mindx.pythai.net/v1/config/fallback-model"
assert recorder["body"] == {"provider": "vllm", "model": "pythai/mindx-fallback-qwen3-1.5b"}
# No bearer header when MINDXTRAIN_API_KEY is unset.
assert "authorization" not in {k.lower() for k in recorder["headers"]}
assert out["current"] == "pythai/mindx-fallback-qwen3-1.5b"
def test_swap_sends_bearer_when_key_set(monkeypatch):
recorder: dict = {}
transport = _capturing_transport(
recorder=recorder,
response_body={"success": True, "previous": "x", "current": "y"},
)
orig_client = httpx.Client
monkeypatch.setattr(
httpx, "Client",
lambda *a, **kw: orig_client(*a, **{**kw, "transport": transport}),
)
monkeypatch.setenv("MINDXTRAIN_API_KEY", "secret-token")
swap_mindx_fallback_model(provider="vllm", model="x/y")
assert recorder["headers"].get("authorization") == "Bearer secret-token"
def test_swap_honours_api_url_override(monkeypatch):
recorder: dict = {}
transport = _capturing_transport(
recorder=recorder,
response_body={"success": True, "previous": "a", "current": "b"},
)
orig_client = httpx.Client
monkeypatch.setattr(
httpx, "Client",
lambda *a, **kw: orig_client(*a, **{**kw, "transport": transport}),
)
monkeypatch.delenv("MINDXTRAIN_API_KEY", raising=False)
swap_mindx_fallback_model(
provider="vllm",
model="x/y",
api_url="http://localhost:8080/", # trailing slash should be stripped
)
assert recorder["url"] == "http://localhost:8080/v1/config/fallback-model"
def test_swap_raises_on_http_error(monkeypatch):
recorder: dict = {}
transport = _capturing_transport(
recorder=recorder,
response_body={"detail": "unknown model"},
status=422,
)
orig_client = httpx.Client
monkeypatch.setattr(
httpx, "Client",
lambda *a, **kw: orig_client(*a, **{**kw, "transport": transport}),
)
monkeypatch.delenv("MINDXTRAIN_API_KEY", raising=False)
with pytest.raises(httpx.HTTPStatusError):
swap_mindx_fallback_model(provider="vllm", model="bogus/model")
def test_swap_exposed_in_module_all():
"""Function and payload model must be in __all__ so `from … import *` users see them."""
assert "swap_mindx_fallback_model" in api_client.__all__
assert "MindXFallbackSwap" in api_client.__all__
|