test_speaker_emotion_docs.py

"""Execute the bilingual recipe and the real rich-text presentation helper."""

import ast
import json
import os
import runpy
import sys
import types
from pathlib import Path

import pytest
from bs4 import BeautifulSoup
from markdown import markdown

ROOT = Path(__file__).resolve().parents[1]
GUIDES = ["speaker_emotion.md", "speaker_emotion_zh.md"]


def test_bilingual_task_guides_exist():
    for name in GUIDES:
        assert (ROOT / "docs" / name).is_file()


@pytest.fixture(params=GUIDES)
def recipe(request):
    path = ROOT / "docs" / request.param
    soup = BeautifulSoup(markdown(path.read_text(), extensions=["fenced_code"]), "html.parser")
    blocks = soup.select("code.language-python")
    assert len(blocks) == 1
    ast.parse(blocks[0].get_text())
    return compile(blocks[0].get_text(), str(path), "exec")


class Waveform:
    ndim = 1

    def __init__(self, length=48000, ndim=1):
        self.length, self.ndim = length, ndim

    def __len__(self):
        return self.length


class Embedding:
    ndim = 2

    def __init__(self, rows=None):
        self.rows = [[0.25, -0.5, 0.75]] if rows is None else rows
        self.shape = (len(self.rows), len(self.rows[0]) if self.rows else 0)
        self.detached = self.on_cpu = False

    def detach(self):
        self.detached = True
        return self

    def cpu(self):
        self.on_cpu = True
        return self

    def tolist(self):
        assert self.detached and self.on_cpu
        return self.rows


def run_recipe(recipe, monkeypatch, tmp_path, task="embedding", results=None,
               sample_rate=16000, length=48000, ndim=1, output=None):
    model_dir = tmp_path / "checkpoint"
    model_dir.mkdir(exist_ok=True)
    constructors, calls = [], []
    if results is None:
        results = [{"spk_embedding": Embedding()}] if task == "embedding" else [
            {"key": "sample", "text": "<|zh|><|HAPPY|><|Speech|><|withitn|>你好"}
        ]

    class Model:
        def __init__(self, **kwargs):
            constructors.append(kwargs)

        def generate(self, **kwargs):
            calls.append(kwargs)
            return results

    helper = runpy.run_path(str(ROOT / "funasr/utils/postprocess_utils.py"))
    postprocess = types.ModuleType("funasr.utils.postprocess_utils")
    postprocess.rich_transcription_postprocess = helper["rich_transcription_postprocess"]
    monkeypatch.setitem(sys.modules, "funasr", types.SimpleNamespace(AutoModel=Model))
    monkeypatch.setitem(sys.modules, "funasr.utils", types.ModuleType("funasr.utils"))
    monkeypatch.setitem(sys.modules, "funasr.utils.postprocess_utils", postprocess)
    monkeypatch.setitem(sys.modules, "soundfile", types.SimpleNamespace(
        read=lambda *a, **k: (Waveform(length, ndim), sample_rate)))
    output = output or tmp_path / "result.json"
    monkeypatch.setattr(sys, "argv", ["attributes.py", task, str(model_dir), "speech.wav", str(output)])
    scope = {"__name__": "__main__"}
    exec(recipe, scope)
    return json.loads(output.read_text()), calls, constructors


def test_embedding_is_a_finite_cpu_serializable_vector_not_identity(recipe, monkeypatch, tmp_path):
    data, calls, constructors = run_recipe(recipe, monkeypatch, tmp_path)
    assert data["result"] == {"spk_embedding": [[0.25, -0.5, 0.75]]}
    assert "name" not in data["result"] and "text" not in data["result"]
    assert calls[0]["batch_size"] == 1 and calls[0]["fs"] == 16000
    assert "cache" not in calls[0] and "language" not in calls[0]
    assert constructors[0]["device"] == "cpu"
    assert constructors[0]["trust_remote_code"] is False
    assert all(constructors[0][p] is None for p in ("vad_model", "punc_model", "spk_model"))
    if os.name == "posix":
        assert (tmp_path / "result.json").stat().st_mode & 0o777 == 0o600


def test_raw_tags_survive_real_lossy_display_processing(recipe, monkeypatch, tmp_path):
    raw = "<|zh|><|HAPPY|><|Speech|><|withitn|>你好"
    data, calls, _ = run_recipe(recipe, monkeypatch, tmp_path, task="sensevoice", results=[{"key": "sample", "text": raw}])
    assert data["result"] == [{"key": "sample", "raw_tagged_text": raw, "display_text": "你好😊"}]
    assert calls[0]["language"] == "auto"
    assert calls[0]["use_itn"] is True and calls[0]["output_timestamp"] is False
    assert "return_raw_text" not in calls[0]


def test_multiple_sensevoice_results_are_preserved_in_order(recipe, monkeypatch, tmp_path):
    raw = ["<|en|><|NEUTRAL|><|Speech|><|withitn|>Hello", "<|zh|><|SAD|><|Speech|><|withitn|>再见"]
    data, _, _ = run_recipe(recipe, monkeypatch, tmp_path, task="sensevoice", results=[{"text": text} for text in raw])
    assert [item["raw_tagged_text"] for item in data["result"]] == raw
    assert all(item["key"] is None for item in data["result"])


@pytest.mark.parametrize("rate,length,ndim", [(8000, 48000, 1), (16000, 0, 1), (16000, 48000, 2)])
def test_invalid_audio_rejected(recipe, monkeypatch, tmp_path, rate, length, ndim):
    with pytest.raises(ValueError, match="nonempty mono 16 kHz"):
        run_recipe(recipe, monkeypatch, tmp_path, sample_rate=rate, length=length, ndim=ndim)
    assert not (tmp_path / "result.json").exists()


@pytest.mark.parametrize("rows", [[], [[]], [[], []], [[1.0], [2.0]], [[float("nan")]]])
def test_invalid_embedding_is_not_published_as_json(recipe, monkeypatch, tmp_path, rows):
    with pytest.raises(ValueError):
        run_recipe(recipe, monkeypatch, tmp_path, results=[{"spk_embedding": Embedding(rows)}])
    assert not (tmp_path / "result.json").exists()


@pytest.mark.parametrize("task", ["embedding", "sensevoice"])
def test_empty_result_is_not_invented(recipe, monkeypatch, tmp_path, task):
    with pytest.raises(ValueError, match="No result"):
        run_recipe(recipe, monkeypatch, tmp_path, task=task, results=[])
    assert not (tmp_path / "result.json").exists()


def test_existing_output_is_not_overwritten(recipe, monkeypatch, tmp_path):
    path = tmp_path / "prior.json"
    path.write_text("existing result")
    with pytest.raises(FileExistsError):
        run_recipe(recipe, monkeypatch, tmp_path, output=path)
    assert path.read_text() == "existing result"


def test_malformed_rich_text_is_not_silently_discarded(recipe, monkeypatch, tmp_path):
    with pytest.raises(ValueError, match="Expected SenseVoice text"):
        run_recipe(recipe, monkeypatch, tmp_path, task="sensevoice", results=[{"text": 123}])
    assert not (tmp_path / "result.json").exists()