337 lines
13 KiB
Python
337 lines
13 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for duplicate-speaker detection and approval-safe merging.
|
|
|
|
The rules that matter here are the ones that REFUSE to merge. A missed duplicate is a small
|
|
manual fix; a wrong merge silently rewrites a speaker and can invalidate a voice someone
|
|
paid for. So the veto cases are tested as carefully as the positive ones.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from umbraaudio.analysis.speaker_dedup import (
|
|
MIN_SIMILARITY,
|
|
SpeakerFacts,
|
|
candidate_groups,
|
|
classify,
|
|
content_tokens,
|
|
is_designation,
|
|
load_speaker_facts,
|
|
merge_group,
|
|
normalize_name,
|
|
similarity,
|
|
unresolved_speaker_names,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
def _f(name: str, **kwargs) -> SpeakerFacts: # noqa: ANN003
|
|
return SpeakerFacts(name=name, **kwargs)
|
|
|
|
|
|
def _db(path: Path, rows: list[tuple]) -> sqlite3.Connection: # noqa: ANN001
|
|
con = sqlite3.connect(str(path))
|
|
con.executescript(
|
|
"CREATE TABLE chapters (id INTEGER PRIMARY KEY);"
|
|
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, chapter_id INTEGER, "
|
|
"utterance_index INTEGER, text TEXT, type TEXT, speaker_name TEXT);"
|
|
)
|
|
for i, (chapter_id, idx, utype, speaker) in enumerate(rows, start=1):
|
|
con.execute(
|
|
"INSERT INTO utterances VALUES (?,?,?,?,?,?)",
|
|
(i, chapter_id, idx, "line", utype, speaker),
|
|
)
|
|
con.commit()
|
|
return con
|
|
|
|
|
|
class TestNormalisation:
|
|
def test_determiner_is_dropped(self) -> None:
|
|
assert normalize_name("The Girl") == normalize_name("Girl")
|
|
assert normalize_name("One Girl Helping") == "girl helping"
|
|
|
|
def test_ordinals_become_digits(self) -> None:
|
|
""""The first guard" and "Guard 1" are the same person written two ways."""
|
|
assert normalize_name("The first guard") == normalize_name("Guard 1")
|
|
assert normalize_name("The second guard") == normalize_name("Guard 2")
|
|
|
|
def test_superlatives_are_stemmed(self) -> None:
|
|
assert normalize_name("The biggest one") == normalize_name("The Big One")
|
|
|
|
def test_parentheses_do_not_split_identity(self) -> None:
|
|
assert normalize_name("Aide (Royal)") == "aide royal"
|
|
|
|
def test_is_designation(self) -> None:
|
|
assert is_designation("The Girl helping")
|
|
assert is_designation("Sergeant")
|
|
assert not is_designation("Hallr")
|
|
assert not is_designation("Sergeant Atonal")
|
|
|
|
def test_content_tokens_keeps_the_distinguishing_word(self) -> None:
|
|
assert content_tokens("Walking oldest") == {"walking", "old"}
|
|
|
|
|
|
class TestVetoes:
|
|
"""Rules that must REFUSE a merge."""
|
|
|
|
def test_different_members_of_a_crowd_are_not_twins(self) -> None:
|
|
"""The oldest man and the tallest man are two people."""
|
|
score, signals = similarity(_f("The Oldest"), _f("The Tallest"))
|
|
assert score < 0.5
|
|
assert any("LIKELY DIFFERENT PEOPLE" in s for s in signals)
|
|
|
|
def test_crowd_veto_survives_stemming(self) -> None:
|
|
score, signals = similarity(_f("Walking oldest"), _f("Walking Tallest"))
|
|
assert score < 0.5
|
|
assert any("LIKELY DIFFERENT PEOPLE" in s for s in signals)
|
|
|
|
def test_guard_one_is_not_guard_two(self) -> None:
|
|
score, _ = similarity(_f("Guard 1"), _f("Guard 2"))
|
|
assert score < 0.5
|
|
|
|
def test_different_roles_are_not_twins(self) -> None:
|
|
"""A rider and a soldier-rider share the word "rider" and are different roles."""
|
|
score, _ = similarity(_f("Soldier (Rider)"), _f("One Rider"))
|
|
assert score < MIN_SIMILARITY, "must stay below the merge threshold"
|
|
|
|
def test_two_proper_names_need_a_high_bar(self) -> None:
|
|
""""Anfisa" and "Tanisha" are 0.77 similar as strings and are different people."""
|
|
assert similarity(_f("Anfisa"), _f("Tanisha"))[0] == 0.0
|
|
# ...but a genuine misspelling of one name still pairs up.
|
|
assert similarity(_f("Rielan"), _f("Rieland"))[0] >= 0.88
|
|
|
|
def test_oldest_and_old_woman_are_not_merged(self) -> None:
|
|
"""Regression: a 'shared token + role noun' rule used to propose this."""
|
|
assert similarity(_f("The Oldest"), _f("Old Woman"))[0] < 0.72
|
|
|
|
|
|
class TestPositives:
|
|
def test_case_variant_is_identical(self) -> None:
|
|
assert similarity(_f("Scum leader"), _f("scum leader"))[0] == 1.0
|
|
|
|
def test_determiner_variant_is_strong(self) -> None:
|
|
score, _ = similarity(_f("The Sergeant"), _f("Sergeant"))
|
|
assert score >= 0.93
|
|
|
|
def test_activity_modifier_is_the_same_person(self) -> None:
|
|
score, signals = similarity(_f("The Girl"), _f("The Girl helping"))
|
|
assert score >= 0.9
|
|
assert any("activity only" in s for s in signals)
|
|
|
|
def test_ordinal_wording_is_the_same_person(self) -> None:
|
|
assert similarity(_f("The first guard"), _f("Guard 1"))[0] >= 0.99
|
|
|
|
|
|
class TestSurvivorChoice:
|
|
def test_approval_is_never_sacrificed(self) -> None:
|
|
group = [
|
|
_f("Big One", utterances=400, chapters=20),
|
|
_f("The Biggest One", utterances=0, approved=True),
|
|
]
|
|
facts = {f.name: f for f in group}
|
|
groups = candidate_groups(facts)
|
|
assert groups and groups[0].survivor == "The Biggest One"
|
|
|
|
def test_voice_reference_is_preferred_over_volume(self) -> None:
|
|
"""A created reference outranks raw utterance count."""
|
|
group = [_f("One Sergeant", utterances=50), _f("Sergeant", utterances=1, has_ref=True)]
|
|
facts = {f.name: f for f in group}
|
|
assert candidate_groups(facts)[0].survivor == "Sergeant"
|
|
|
|
def test_volume_wins_when_nothing_else_does(self) -> None:
|
|
group = [_f("One Soldier", utterances=3), _f("Soldier", utterances=90)]
|
|
facts = {f.name: f for f in group}
|
|
assert candidate_groups(facts)[0].survivor == "Soldier"
|
|
|
|
|
|
class TestClassification:
|
|
def test_silent_twin_is_auto(self) -> None:
|
|
group = [_f("Sergeant", utterances=14), _f("The Sergeant", utterances=0)]
|
|
verdict, _ = classify(group, 0.99)
|
|
assert verdict == "auto"
|
|
|
|
def test_two_speakers_heard_is_review(self) -> None:
|
|
group = [_f("The Girl", utterances=4), _f("The Girl helping", utterances=4)]
|
|
verdict, _ = classify(group, 0.99)
|
|
assert verdict == "review"
|
|
|
|
def test_two_approved_voices_is_blocked(self) -> None:
|
|
group = [_f("A", utterances=5, approved=True), _f("B", utterances=3, approved=True)]
|
|
verdict, reasons = classify(group, 1.0)
|
|
assert verdict == "blocked"
|
|
assert any("approved voice" in r for r in reasons)
|
|
|
|
def test_conflicting_voice_references_block(self) -> None:
|
|
group = [
|
|
_f("A", utterances=9, has_ref=True, ref_sha256="aaaa"),
|
|
_f("B", utterances=4, has_ref=True, ref_sha256="bbbb"),
|
|
]
|
|
verdict, reasons = classify(group, 1.0)
|
|
assert verdict == "blocked"
|
|
assert any("conflicting voice references" in r for r in reasons)
|
|
|
|
def test_weak_chain_cannot_be_auto(self) -> None:
|
|
"""A strong pair plus a weak third member must not launder into 'auto'."""
|
|
group = [_f("Walking oldest", utterances=5), _f("The Oldest", utterances=0)]
|
|
verdict, _ = classify(group, 0.80)
|
|
assert verdict == "review"
|
|
|
|
|
|
class TestUnresolvedNames:
|
|
def test_case_variant_spoken_name_is_reported(self, tmp_path: Path) -> None:
|
|
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "scum leader")])
|
|
found = unresolved_speaker_names(db, {"Scum leader"})
|
|
assert found == [("scum leader", 1, 1)]
|
|
db.close()
|
|
|
|
def test_exact_match_is_not_reported(self, tmp_path: Path) -> None:
|
|
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "Hallr")])
|
|
assert unresolved_speaker_names(db, {"Hallr"}) == []
|
|
db.close()
|
|
|
|
def test_genuinely_unknown_name_is_not_reported(self, tmp_path: Path) -> None:
|
|
db = _db(tmp_path / "p.db", [(1, 0, "dialogue", "Manfred")])
|
|
assert unresolved_speaker_names(db, {"Hallr"}) == []
|
|
db.close()
|
|
|
|
|
|
def _project(tmp_path: Path, profiles: list[str], ledger: dict, rows: list[tuple],
|
|
refs: set[str] | None = None) -> Path:
|
|
data = tmp_path / ".umbra"
|
|
(data / "voices").mkdir(parents=True, exist_ok=True)
|
|
(data / "speakers.json").write_text(
|
|
json.dumps({"speakers": [{"primary_name": n} for n in profiles]})
|
|
)
|
|
(data / "voice_status.json").write_text(json.dumps(ledger))
|
|
for ref in refs or set():
|
|
(data / "voices" / f"{ref.replace(' ', '_')}_ref.wav").write_bytes(ref.encode())
|
|
_db(data / "project.db", rows).close()
|
|
return data
|
|
|
|
|
|
class TestMergeSafety:
|
|
def test_approval_is_carried_to_the_survivor(self, tmp_path: Path) -> None:
|
|
data = _project(
|
|
tmp_path,
|
|
profiles=["Keep", "Lose"],
|
|
ledger={"Lose": "approved"},
|
|
rows=[(1, 0, "dialogue", "Lose")],
|
|
)
|
|
con = sqlite3.connect(str(data / "project.db"))
|
|
result = merge_group(con, data, "Keep", ["Lose"])
|
|
con.close()
|
|
assert result["ok"] is True
|
|
ledger = json.loads((data / "voice_status.json").read_text())
|
|
assert ledger["Keep"] == "approved"
|
|
assert "Lose" not in ledger
|
|
assert result["approvals_carried"] == ["Lose"]
|
|
|
|
def test_repoint_never_touches_narration(self, tmp_path: Path) -> None:
|
|
"""Regression: an unscoped UPDATE moved 5,978 narration rows on Last Dwarf."""
|
|
data = _project(
|
|
tmp_path,
|
|
profiles=["Narrator"],
|
|
ledger={"Narrator": "approved"},
|
|
rows=[
|
|
(1, 0, "dialogue", "narrator"),
|
|
(1, 1, "narration", "narrator"),
|
|
(1, 2, "narration", "narrator"),
|
|
(1, 3, "scene_break", "Narrator"),
|
|
],
|
|
)
|
|
con = sqlite3.connect(str(data / "project.db"))
|
|
result = merge_group(con, data, "Narrator", ["narrator"])
|
|
remaining = con.execute(
|
|
"SELECT COUNT(*) FROM utterances WHERE speaker_name='narrator'"
|
|
).fetchone()[0]
|
|
moved = con.execute(
|
|
"SELECT COUNT(*) FROM utterances "
|
|
"WHERE speaker_name='Narrator' AND lower(type)='dialogue'"
|
|
).fetchone()[0]
|
|
con.close()
|
|
assert result["ok"] is True
|
|
assert result["repointed_only"] == ["narrator"]
|
|
assert moved == 1
|
|
assert remaining == 2, "narration rows must not be re-pointed"
|
|
|
|
def test_profile_merge_removes_the_duplicate_profile(self, tmp_path: Path) -> None:
|
|
data = _project(
|
|
tmp_path,
|
|
profiles=["Sergeant", "The Sergeant"],
|
|
ledger={},
|
|
rows=[(1, 0, "dialogue", "The Sergeant")],
|
|
)
|
|
con = sqlite3.connect(str(data / "project.db"))
|
|
result = merge_group(con, data, "Sergeant", ["The Sergeant"])
|
|
con.close()
|
|
assert result["merged"] == ["The Sergeant"]
|
|
names = {
|
|
p["primary_name"]
|
|
for p in json.loads((data / "speakers.json").read_text())["speakers"]
|
|
}
|
|
assert "The Sergeant" not in names
|
|
assert "Sergeant" in names
|
|
|
|
def test_conflicting_voices_are_refused_before_any_write(self, tmp_path: Path) -> None:
|
|
data = _project(
|
|
tmp_path,
|
|
profiles=["Officer", "The Officer"],
|
|
ledger={},
|
|
rows=[(1, 0, "dialogue", "Officer")],
|
|
refs={"Officer", "The Officer"},
|
|
)
|
|
before = (data / "speakers.json").read_text()
|
|
con = sqlite3.connect(str(data / "project.db"))
|
|
result = merge_group(con, data, "The Officer", ["Officer"])
|
|
con.close()
|
|
assert result["ok"] is False
|
|
assert result["reason"] == "conflicting voice references"
|
|
assert (data / "speakers.json").read_text() == before
|
|
|
|
def test_voice_files_are_never_deleted(self, tmp_path: Path) -> None:
|
|
"""A merge folds identity; it does not throw away a generated reference."""
|
|
data = _project(
|
|
tmp_path,
|
|
profiles=["Soldier", "One Soldier"],
|
|
ledger={},
|
|
rows=[(1, 0, "dialogue", "One Soldier")],
|
|
refs={"Soldier", "One Soldier"},
|
|
)
|
|
con = sqlite3.connect(str(data / "project.db"))
|
|
merge_group(con, data, "Soldier", ["One Soldier"])
|
|
con.close()
|
|
assert (data / "voices" / "One_Soldier_ref.wav").exists()
|
|
assert (data / "voices" / "Soldier_ref.wav").exists()
|
|
|
|
|
|
def test_load_speaker_facts_is_read_only(tmp_path: Path) -> None:
|
|
data = _project(tmp_path, ["Hallr"], {"Hallr": "approved"}, [(1, 0, "dialogue", "Hallr")])
|
|
before = (data / "speakers.json").read_text()
|
|
facts = load_speaker_facts(data)
|
|
assert facts["Hallr"].approved is True
|
|
assert facts["Hallr"].utterances == 1
|
|
assert (data / "speakers.json").read_text() == before
|
|
|
|
|
|
def test_spoken_only_name_becomes_a_fact_so_it_can_be_reported(tmp_path: Path) -> None:
|
|
data = _project(tmp_path, ["Hallr"], {}, [(1, 0, "dialogue", "scum leader")])
|
|
facts = load_speaker_facts(data)
|
|
assert "scum leader" in facts
|
|
assert facts["scum leader"].utterances == 1
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["", " ", "()"])
|
|
def test_degenerate_names_do_not_crash(name: str) -> None:
|
|
facts = _f(name)
|
|
assert normalize_name(name) == ""
|
|
assert is_designation(name) is False
|
|
assert similarity(facts, _f("Hallr"))[0] == 0.0
|