| |
| |
| """The user-facing surfaces on the GPUs: Diarizer(...)(file), Diarizer.run(..., out_dir), Diarization, and the |
| anuvani-diar command line, against the raw-audio oracle and the release's own golden hashes (golden/expected.json).""" |
| import hashlib |
| import json |
| import subprocess |
| import sys |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from conftest import PROJ, REL |
|
|
| GOLDEN = json.loads((REL / "golden/expected.json").read_text())["files"] |
| ORACLE = PROJ / "artifacts/oracle_audio" |
| INPUTS = {i["sid"]: i for i in json.loads((PROJ / "artifacts/golden12_inputs.json").read_text())} |
| FILES = sorted(INPUTS)[:4] |
|
|
|
|
| def sha(b): |
| return hashlib.sha256(b).hexdigest() |
|
|
|
|
| @pytest.fixture(scope="module") |
| def diarizer(): |
| from anuvani_diar import Diarizer |
| return Diarizer() |
|
|
|
|
| @pytest.mark.gpu |
| def test_call_and_run(diarizer, tmp_path): |
| sid = FILES[0] |
| r = diarizer(INPUTS[sid]["path"]) |
| assert r.ok and r.uri == sid |
| r.save_rttm(tmp_path / "one.rttm") |
| assert sha((tmp_path / "one.rttm").read_bytes()) == GOLDEN[sid]["indic_rttm_sha256"] |
| assert r.segments and r.speakers and r.duration_s > 0 |
| out = tmp_path / "out" |
| got = list(diarizer.run([INPUTS[s]["path"] for s in FILES], out_dir=out)) |
| assert len(got) == len(FILES) and all(g.ok for g in got) |
| for s in FILES: |
| assert sha((out / f"{s}.rttm").read_bytes()) == GOLDEN[s]["indic_rttm_sha256"], s |
|
|
|
|
| @pytest.mark.gpu |
| def test_cli_matches_diarize_py(tmp_path): |
| out = tmp_path / "out" |
| paths = [INPUTS[s]["path"] for s in FILES] |
| cli = [sys.executable, "-s", "-m", "anuvani_diar.cli", "--out", str(out), *paths] |
| r = subprocess.run(cli, capture_output=True, text=True) |
| assert r.returncode == 0, r.stderr[-2000:] |
| run = json.loads((out / "run.json").read_text()) |
| assert len(run["files"]) == len(FILES) |
| assert set(run) == {"version", "model_config", "bundle_sha256", "device", "batch_size", "threads", "python", "files"} |
| for s, rec in zip(FILES, run["files"]): |
| assert Path(rec["input"]).stem == s |
| assert rec["rttm_sha256"] == GOLDEN[s]["indic_rttm_sha256"] == sha(Path(rec["rttm"]).read_bytes()) |
| assert set(rec) == {"input", "rttm", "duration_s", "seconds", "rttm_sha256"} |
|
|
|
|
| @pytest.mark.gpu |
| def test_install_check(tmp_path): |
| """`anuvani-diar-check`: the release's 12 golden fingerprints plus the 1 h, 3 h and 6 h files, one command.""" |
| audio = tmp_path / "audio" |
| audio.mkdir() |
| for s, it in INPUTS.items(): |
| (audio / f"{s}.flac").symlink_to(it["path"]) |
| r = subprocess.run([sys.executable, "-s", "-m", "anuvani_diar.check", "--audio-dir", str(audio), |
| "--long-dir", str(PROJ / "artifacts/test_audio")], capture_output=True, text=True) |
| assert r.returncode == 0, r.stdout[-2000:] + r.stderr[-2000:] |
| assert "15/15 outputs identical" in r.stdout and "WARNING" not in r.stdout |
|
|
|
|
| def test_diarizer_refuses_to_start_without_its_network_check(monkeypatch): |
| """No silent skip: without assets/nn_golden.json the networks are never used unchecked.""" |
| from anuvani_diar import api |
| monkeypatch.setattr(api, "default_golden", lambda: None) |
| with pytest.raises(FileNotFoundError, match="nn_golden.json"): |
| api.Diarizer() |
|
|
