Recheck narrowband TTS content with stronger ASR

Signed-off-by: tzhouam <tzhouam@connect.ust.hk>

tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py

@@ -148,6 +148,9 @@ def test_sample_rate_001(omni_server, online_client) -> None:
         "stream": False,
         "response_format": "wav",
         "sample_rate": 8000,
+        # Whisper-small can mishear the narrowband "kilohertz" pronunciation.
+        # Recheck the same WAV with stronger ASR, preserving the 0.9 gate.
+        "transcript_escalation_model": "large-v3",
         "task_type": "CustomVoice",
         "voice": "vivian",
     }

tests/helpers/tests/test_assertions.py

@@ -70,6 +70,56 @@ def test_short_transcript_unrelated_text_still_fails():
         )
 
 
+@pytest.mark.parametrize("result", ["small_matches", "strong_matches", "both_mismatch", "wrong_sample_rate"])
+def test_qwen3_tts_narrowband_asr_preserves_content_and_sample_rate_gates(monkeypatch, result):
+    from tests.e2e.online_serving import test_qwen3_tts_customvoice_expansion as speech
+
+    expected = "This response should be encoded as eight kilohertz audio."
+    sample_rate = 24_000 if result == "wrong_sample_rate" else 8000
+    buffer = io.BytesIO()
+    with wave.open(buffer, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(sample_rate)
+        wav.writeframes(_pcm_sine(sample_rate=sample_rate))
+    audio_bytes = buffer.getvalue()
+    models = []
+
+    def transcribe(raw_bytes, model_size="small", language=None):
+        assert raw_bytes == audio_bytes
+        models.append(model_size)
+        if result == "small_matches":
+            return expected
+        if model_size == "large-v3" and result == "strong_matches":
+            return "This response should be encoded as 8kHz audio."
+        return "This response should be encoded as 8K red audio."
+
+    monkeypatch.setattr(assertions, "convert_audio_bytes_to_text", transcribe)
+    monkeypatch.setattr(assertions, "_assert_preset_voice_gender_from_audio", lambda *_args, **_kwargs: None)
+
+    def request(config):
+        assert config["input"] == expected
+        assert_audio_speech_response(
+            OmniResponse(success=True, audio_bytes=audio_bytes, audio_format="audio/wav"), config, "full_model"
+        )
+
+    client = SimpleNamespace(send_audio_speech_request=request)
+    error = {
+        "both_mismatch": "after ASR escalation",
+        "wrong_sample_rate": "Expected sample_rate=8000",
+    }.get(result)
+    expectation = pytest.raises(AssertionError, match=error) if error else nullcontext()
+    with expectation:
+        speech.test_sample_rate_001(SimpleNamespace(model=speech.MODEL), client)
+    expected_models = {
+        "small_matches": ["small"],
+        "strong_matches": ["small", "large-v3"],
+        "both_mismatch": ["small", "large-v3"],
+        "wrong_sample_rate": [],
+    }[result]
+    assert models == expected_models
+
+
 def _capture_transcribe(monkeypatch) -> dict:
     captured: dict = {}