mirror of
https://github.com/modelscope/FunASR.git
synced 2026-10-12 02:19:18 +00:00
232 lines
9.0 KiB
Python
232 lines
9.0 KiB
Python
"""Model-free tests for the real Gradio UI and multipart HTTP client."""
|
|
|
|
import importlib.util
|
|
import json
|
|
from email import policy
|
|
from email.parser import BytesParser
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
import threading
|
|
import wave
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"gradio_app", ROOT / "examples/openai_api/gradio_app.py"
|
|
)
|
|
app = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def audio(tmp_path):
|
|
path = tmp_path / "sample.wav"
|
|
with wave.open(str(path), "wb") as stream:
|
|
stream.setnchannels(1)
|
|
stream.setsampwidth(2)
|
|
stream.setframerate(16000)
|
|
stream.writeframes(b"\x00\x00" * 1600)
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def endpoint():
|
|
state = {
|
|
"requests": [],
|
|
"status": 200,
|
|
"metadata": True,
|
|
"payload": {
|
|
"text": "\u4f60\u597d",
|
|
"segments": [
|
|
{"start": 0, "end": 0.5, "text": "\u4f60", "speaker": "speaker_0"},
|
|
{"start": 0.5, "end": 1, "text": "\u597d", "speaker": 1},
|
|
{"text": "[S02] untouched", "speaker": None},
|
|
],
|
|
},
|
|
}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *_args):
|
|
pass
|
|
|
|
def reply(self, status, payload):
|
|
body = json.dumps(payload).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
state["requests"].append(("GET", self.path, {}))
|
|
self.reply(200 if state["metadata"] else 404, {"path": self.path})
|
|
|
|
def do_POST(self):
|
|
body = self.rfile.read(int(self.headers["Content-Length"]))
|
|
message = BytesParser(policy=policy.default).parsebytes(
|
|
("Content-Type: " + self.headers["Content-Type"] + "\r\n\r\n").encode() + body
|
|
)
|
|
fields = {
|
|
part.get_param("name", header="content-disposition"): part.get_payload(decode=True)
|
|
for part in message.iter_parts()
|
|
}
|
|
state["requests"].append(("POST", self.path, fields))
|
|
self.reply(state["status"], state["payload"])
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
|
worker.start()
|
|
try:
|
|
yield f"http://127.0.0.1:{server.server_port}", state
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
worker.join(timeout=5)
|
|
assert not worker.is_alive()
|
|
|
|
|
|
def component(demo, label):
|
|
matches = [item for item in demo.config["components"] if item["props"].get("label") == label]
|
|
assert len(matches) == 1
|
|
return matches[0]["props"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"backend,model,formats,default_format",
|
|
[
|
|
("funasr", "sensevoice", ["json", "verbose_json"], "verbose_json"),
|
|
("vllm", "moss-transcribe-diarize", ["json", "diarized_json"], "diarized_json"),
|
|
("sglang-omni", "OpenMOSS-Team/MOSS-Transcribe-Diarize", ["verbose_json"], "verbose_json"),
|
|
],
|
|
)
|
|
def test_real_ui_backend_contract(backend, model, formats, default_format):
|
|
demo = app.build_app("http://127.0.0.1:8000", 300, backend=backend)
|
|
try:
|
|
model_control = component(demo, "Model alias")
|
|
assert model_control["value"] == model
|
|
choices = [choice[1] for choice in model_control["choices"]]
|
|
assert model in choices
|
|
if backend == "sglang-omni":
|
|
assert model_control["choices"] == [("MOSS-Transcribe-Diarize", model)]
|
|
if backend == "funasr":
|
|
assert choices == ["sensevoice", "paraformer", "paraformer-en", "fun-asr-nano", "moss-transcribe-diarize"]
|
|
response = component(demo, "Response format")
|
|
assert [choice[0] for choice in response["choices"]] == formats
|
|
assert response["value"] == default_format
|
|
assert component(demo, "API base URL")["value"] == "http://127.0.0.1:8000"
|
|
for label in ["API base URL", "Model alias", "Response format", "Timeout seconds"]:
|
|
assert component(demo, label)["min_width"] >= 240
|
|
assert component(demo, "Audio")["type"] == "filepath"
|
|
finally:
|
|
demo.close()
|
|
|
|
|
|
def test_original_build_app_call_remains_supported():
|
|
demo = app.build_app("http://127.0.0.1:8000", 300)
|
|
try:
|
|
assert component(demo, "Model alias")["value"] == "sensevoice"
|
|
assert component(demo, "Response format")["value"] == "verbose_json"
|
|
finally:
|
|
demo.close()
|
|
|
|
|
|
@pytest.mark.parametrize("backend", ["funasr", "vllm", "sglang-omni"])
|
|
def test_served_model_override_is_exact(backend):
|
|
demo = app.build_app("http://127.0.0.1:8000", 300, backend=backend, default_model="operator/custom-model")
|
|
try:
|
|
control = component(demo, "Model alias")
|
|
assert control["value"] == "operator/custom-model"
|
|
assert "operator/custom-model" in [choice[1] for choice in control["choices"]]
|
|
finally:
|
|
demo.close()
|
|
|
|
|
|
def test_cli_profile_and_model_defaults(monkeypatch):
|
|
for name in ["BASE_URL", "GRADIO_HOST", "GRADIO_PORT", "TIMEOUT"]:
|
|
monkeypatch.delenv(name, raising=False)
|
|
args = app.parse_args([])
|
|
assert (args.backend, args.model, args.host, args.port, args.share) == ("funasr", None, "127.0.0.1", 7860, False)
|
|
args = app.parse_args(["--backend", "vllm", "--model", "served-id", "--base-url", "http://127.0.0.1:8898"])
|
|
assert (args.backend, args.model, args.base_url) == ("vllm", "served-id", "http://127.0.0.1:8898")
|
|
with pytest.raises(SystemExit):
|
|
app.parse_args(["--backend", "not-a-backend"])
|
|
|
|
|
|
@pytest.mark.parametrize("model,format_", [
|
|
("moss-transcribe-diarize", "verbose_json"),
|
|
("moss-transcribe-diarize", "diarized_json"),
|
|
("OpenMOSS-Team/MOSS-Transcribe-Diarize", "verbose_json"),
|
|
("operator/custom-model", "json"),
|
|
])
|
|
def test_real_http_preserves_fields_and_response(endpoint, audio, model, format_):
|
|
url, state = endpoint
|
|
transcript, raw = app.transcribe_audio(url + "/", str(audio), model, format_, 5)
|
|
assert transcript == state["payload"]["text"]
|
|
assert json.loads(raw) == state["payload"]
|
|
assert state["requests"] == [("POST", "/v1/audio/transcriptions", {
|
|
"file": audio.read_bytes(), "model": model.encode(), "response_format": format_.encode(),
|
|
})]
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [{"text": "only text"}, {"text": "", "segments": []}, {"segments": [{"text": "[S00] raw"}]}])
|
|
def test_no_speaker_or_segment_fabrication(endpoint, audio, payload):
|
|
url, state = endpoint
|
|
state["payload"] = payload
|
|
text, raw = app.transcribe_audio(url, str(audio), "moss-transcribe-diarize", "verbose_json", 5)
|
|
assert text == payload.get("text", "")
|
|
assert json.loads(raw) == payload
|
|
|
|
|
|
def test_optional_metadata_failure_does_not_gate_transcription(endpoint, audio):
|
|
url, state = endpoint
|
|
state["metadata"] = False
|
|
assert "HTTP 404" in app.safe_check(url, 5)
|
|
text, raw = app.safe_transcribe(url, str(audio), "moss-transcribe-diarize", "diarized_json", 5)
|
|
assert text == state["payload"]["text"]
|
|
assert json.loads(raw) == state["payload"]
|
|
|
|
|
|
def test_service_check_and_http_failure(endpoint, audio):
|
|
url, state = endpoint
|
|
assert json.loads(app.check_service(url, 5)) == {
|
|
"health": {"path": "/health"}, "models": {"path": "/v1/models"},
|
|
}
|
|
state["status"] = 422
|
|
state["payload"] = {"detail": "unsupported model"}
|
|
text, raw = app.safe_transcribe(url, str(audio), "bad-model", "verbose_json", 5)
|
|
assert text == ""
|
|
assert "HTTP 422" in raw and "unsupported model" in raw
|
|
|
|
|
|
def test_missing_audio_never_calls_backend(endpoint):
|
|
url, state = endpoint
|
|
assert app.safe_transcribe(url, None, "sensevoice", "json", 5)[0] == ""
|
|
assert state["requests"] == []
|
|
|
|
|
|
@pytest.mark.parametrize("backend,model,format_", [
|
|
("funasr", "moss-transcribe-diarize", "verbose_json"),
|
|
("vllm", "moss-transcribe-diarize", "diarized_json"),
|
|
("sglang-omni", "OpenMOSS-Team/MOSS-Transcribe-Diarize", "verbose_json"),
|
|
])
|
|
def test_live_gradio_upload_and_event_binding(endpoint, audio, backend, model, format_):
|
|
from gradio_client import Client, handle_file
|
|
|
|
url, state = endpoint
|
|
demo = app.build_app(url, 5, backend=backend, default_model=model)
|
|
try:
|
|
_, local_url, _ = demo.launch(server_name="127.0.0.1", prevent_thread_lock=True, quiet=True)
|
|
client = Client(local_url, verbose=False)
|
|
transcript, raw = client.predict(url, handle_file(str(audio)), model, format_, 5, api_name="/safe_transcribe")
|
|
assert transcript == state["payload"]["text"]
|
|
assert json.loads(raw) == state["payload"]
|
|
post = state["requests"][-1]
|
|
assert post[:2] == ("POST", "/v1/audio/transcriptions")
|
|
assert post[2]["model"] == model.encode()
|
|
assert post[2]["response_format"] == format_.encode()
|
|
assert post[2]["file"] == audio.read_bytes()
|
|
finally:
|
|
demo.close()
|