265 lines
11 KiB
Python
265 lines
11 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from umbraaudio.analysis.orphan_resolution import (
|
|
ResolutionError,
|
|
assign_token_to_speaker,
|
|
create_speaker_from_token,
|
|
dismiss_token,
|
|
merge_speakers,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def env(tmp_path: Path) -> tuple[sqlite3.Connection, Path]:
|
|
db = sqlite3.connect(":memory:")
|
|
db.row_factory = sqlite3.Row
|
|
db.execute("CREATE TABLE project (id INTEGER PRIMARY KEY)")
|
|
db.execute("INSERT INTO project (id) VALUES (1)")
|
|
db.execute(
|
|
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, speaker_name TEXT, "
|
|
"type TEXT, text TEXT)"
|
|
)
|
|
data_dir = tmp_path / ".umbra"
|
|
data_dir.mkdir()
|
|
speakers = [
|
|
{
|
|
"primary_name": "Kainda",
|
|
"aliases": ["Princess Kainda"],
|
|
"demographics": {"type": "Human", "gender": "Female"},
|
|
"vocal_traits": ["calm"],
|
|
"chapters_appeared": [1, 2],
|
|
},
|
|
{
|
|
"primary_name": "Guard 1",
|
|
"aliases": [],
|
|
"demographics": {"type": "Human", "gender": "Unknown"},
|
|
"vocal_traits": ["gruff"],
|
|
"chapters_appeared": [3],
|
|
},
|
|
]
|
|
(data_dir / "speakers.json").write_text(json.dumps({"speakers": speakers}))
|
|
return db, data_dir
|
|
|
|
|
|
def _add_utt(db: sqlite3.Connection, speaker: str, text: str = "Hello") -> None:
|
|
db.execute(
|
|
"INSERT INTO utterances (speaker_name, type, text) VALUES (?, 'dialogue', ?)",
|
|
(speaker, text),
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def _load(data_dir: Path) -> list[dict]:
|
|
return json.loads((data_dir / "speakers.json").read_text())["speakers"]
|
|
|
|
|
|
class TestAssign:
|
|
def test_assign_adds_alias_and_renormalizes(self, env) -> None:
|
|
db, data_dir = env
|
|
_add_utt(db, "The first guard")
|
|
result = assign_token_to_speaker(db, data_dir, 1, "The first guard", "Guard 1")
|
|
assert result["ok"] and result["alias_added"] is True
|
|
assert result["utterances_changed"] == 1
|
|
|
|
speakers = _load(data_dir)
|
|
guard = next(s for s in speakers if s["primary_name"] == "Guard 1")
|
|
assert "The first guard" in guard["aliases"]
|
|
row = db.execute("SELECT speaker_name FROM utterances").fetchone()
|
|
assert row["speaker_name"] == "Guard 1"
|
|
|
|
def test_assign_is_case_insensitive(self, env) -> None:
|
|
db, data_dir = env
|
|
_add_utt(db, "the first guard")
|
|
assign_token_to_speaker(db, data_dir, 1, "The first guard", "Guard 1")
|
|
row = db.execute("SELECT speaker_name FROM utterances").fetchone()
|
|
assert row["speaker_name"] == "Guard 1"
|
|
|
|
def test_assign_unknown_target_raises(self, env) -> None:
|
|
db, data_dir = env
|
|
with pytest.raises(ResolutionError, match="not found"):
|
|
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Nobody")
|
|
|
|
def test_assign_idempotent_alias(self, env) -> None:
|
|
db, data_dir = env
|
|
_add_utt(db, "The Kid")
|
|
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
|
|
# Second assign should not duplicate the alias.
|
|
result = assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
|
|
guard = next(s for s in _load(data_dir) if s["primary_name"] == "Guard 1")
|
|
assert guard["aliases"].count("The Kid") == 1
|
|
assert result["alias_added"] is False
|
|
|
|
def test_assign_sets_token_status(self, env) -> None:
|
|
db, data_dir = env
|
|
_add_utt(db, "The Kid")
|
|
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
|
|
row = db.execute(
|
|
"SELECT status, target_speaker FROM name_tokens WHERE token='The Kid'"
|
|
).fetchone()
|
|
assert row["status"] == "assigned" and row["target_speaker"] == "Guard 1"
|
|
|
|
def test_assign_alias_evidence_is_string_list(self, env) -> None:
|
|
# Regression guard: discovery's _alias_has_evidence() calls .lower() on
|
|
# each alias_evidence item, so it must stay list[str] (not dicts).
|
|
db, data_dir = env
|
|
_add_utt(db, "The Kid")
|
|
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
|
|
guard = next(s for s in _load(data_dir) if s["primary_name"] == "Guard 1")
|
|
for item in guard["alias_evidence"]:
|
|
assert isinstance(item, str), f"non-string evidence: {item!r}"
|
|
item.lower() # must not raise
|
|
|
|
def test_assign_evidence_survives_discovery_alias_check(self, env) -> None:
|
|
# The evidence string must mention the alias so _alias_has_evidence()
|
|
# returns True for it on a later discovery pass.
|
|
from umbraaudio.analysis.character_discovery import CharacterDiscoverer
|
|
|
|
db, data_dir = env
|
|
_add_utt(db, "The Kid")
|
|
assign_token_to_speaker(db, data_dir, 1, "The Kid", "Guard 1")
|
|
guard = next(s for s in _load(data_dir) if s["primary_name"] == "Guard 1")
|
|
discoverer = CharacterDiscoverer.__new__(CharacterDiscoverer)
|
|
assert discoverer._alias_has_evidence(guard, "The Kid") is True
|
|
|
|
|
|
class TestCreate:
|
|
def test_create_from_token(self, env) -> None:
|
|
db, data_dir = env
|
|
_add_utt(db, "The Kid", "Where am I?")
|
|
result = create_speaker_from_token(db, data_dir, 1, "The Kid")
|
|
assert result["ok"]
|
|
speakers = _load(data_dir)
|
|
kid = next(s for s in speakers if s["primary_name"] == "The Kid")
|
|
assert kid["aliases"] == ["The Kid"]
|
|
assert kid["sample_dialogue"] == "Where am I?"
|
|
row = db.execute(
|
|
"SELECT status, target_speaker FROM name_tokens WHERE token='The Kid'"
|
|
).fetchone()
|
|
assert row["status"] == "created" and row["target_speaker"] == "The Kid"
|
|
|
|
def test_create_duplicate_raises(self, env) -> None:
|
|
db, data_dir = env
|
|
with pytest.raises(ResolutionError, match="already exists"):
|
|
create_speaker_from_token(db, data_dir, 1, "Kainda")
|
|
|
|
|
|
class TestDismiss:
|
|
def test_dismiss_records_status(self, env) -> None:
|
|
db, data_dir = env
|
|
result = dismiss_token(db, 1, "The Voice", reason="a sound effect")
|
|
assert result["ok"]
|
|
row = db.execute(
|
|
"SELECT status FROM name_tokens WHERE token='The Voice'"
|
|
).fetchone()
|
|
assert row["status"] == "dismissed"
|
|
# speakers.json and utterances must be untouched.
|
|
assert all("The Voice" not in s["primary_name"] for s in _load(data_dir))
|
|
|
|
def test_dismiss_preserves_scan_evidence(self, env) -> None:
|
|
from umbraaudio.analysis.name_tokens import (
|
|
NameToken,
|
|
ensure_name_tokens_table,
|
|
persist_name_tokens,
|
|
)
|
|
|
|
db, data_dir = env
|
|
ensure_name_tokens_table(db)
|
|
tok = NameToken(token="The Voice", kind="name_tag", occurrences=2, sample_texts=["a voice says,"])
|
|
persist_name_tokens(db, 1, [tok])
|
|
dismiss_token(db, 1, "The Voice", reason="not a character")
|
|
row = db.execute(
|
|
"SELECT evidence_json FROM name_tokens WHERE token='The Voice'"
|
|
).fetchone()
|
|
evidence = json.loads(row["evidence_json"])
|
|
# Original sample evidence survives; the reason is appended.
|
|
assert "a voice says," in evidence
|
|
assert any("dismissed: not a character" in e for e in evidence)
|
|
|
|
|
|
class TestMerge:
|
|
def test_merge_folds_aliases_and_renormalizes(self, env) -> None:
|
|
db, data_dir = env
|
|
# Register "The first guard" as an alias of Guard 1 (as an earlier
|
|
# token-assignment would have done), then merge.
|
|
speakers = _load(data_dir)
|
|
next(s for s in speakers if s["primary_name"] == "Guard 1")["aliases"] = [
|
|
"The first guard"
|
|
]
|
|
(data_dir / "speakers.json").write_text(
|
|
json.dumps({"speakers": speakers})
|
|
)
|
|
_add_utt(db, "Guard 1")
|
|
_add_utt(db, "The first guard")
|
|
result = merge_speakers(db, data_dir, "Guard 1", "Kainda")
|
|
assert result["merged"] is True
|
|
assert result["utterances_changed"] == 2
|
|
speakers = _load(data_dir)
|
|
assert all(s["primary_name"] != "Guard 1" for s in speakers)
|
|
kainda = next(s for s in speakers if s["primary_name"] == "Kainda")
|
|
assert "Guard 1" in kainda["aliases"]
|
|
assert "The first guard" in kainda["aliases"]
|
|
assert "calm" in kainda["vocal_traits"]
|
|
# Both utterances now on Kainda.
|
|
names = {r["speaker_name"] for r in db.execute("SELECT speaker_name FROM utterances")}
|
|
assert names == {"Kainda"}
|
|
|
|
def test_merge_idempotent(self, env) -> None:
|
|
db, data_dir = env
|
|
merge_speakers(db, data_dir, "Guard 1", "Kainda")
|
|
# Second merge: source is gone -> no-op, not an error.
|
|
result = merge_speakers(db, data_dir, "Guard 1", "Kainda")
|
|
assert result["merged"] is False
|
|
|
|
def test_merge_commits_for_a_fresh_connection(self, tmp_path) -> None:
|
|
# Regression guard: merge_speakers must COMMIT its utterance rename, so a
|
|
# later, independent connection sees the re-normalized speaker. Uses a
|
|
# file-backed DB because in-memory DBs are per-connection.
|
|
db_path = tmp_path / "p.db"
|
|
db = sqlite3.connect(db_path)
|
|
db.row_factory = sqlite3.Row
|
|
db.execute("CREATE TABLE project (id INTEGER PRIMARY KEY)")
|
|
db.execute("INSERT INTO project (id) VALUES (1)")
|
|
db.execute(
|
|
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, speaker_name TEXT, "
|
|
"type TEXT, text TEXT)"
|
|
)
|
|
db.execute("INSERT INTO utterances (speaker_name, type, text) VALUES ('Guard 1','dialogue','x')")
|
|
db.commit()
|
|
|
|
data_dir = tmp_path / ".umbra"
|
|
data_dir.mkdir()
|
|
(data_dir / "speakers.json").write_text(
|
|
json.dumps({"speakers": [
|
|
{"primary_name": "Kainda", "aliases": [], "chapters_appeared": []},
|
|
{"primary_name": "Guard 1", "aliases": [], "chapters_appeared": []},
|
|
]})
|
|
)
|
|
merge_speakers(db, data_dir, "Guard 1", "Kainda")
|
|
db.close()
|
|
|
|
# A completely separate connection must see the committed rename.
|
|
verify = sqlite3.connect(db_path)
|
|
verify.row_factory = sqlite3.Row
|
|
name = verify.execute("SELECT speaker_name FROM utterances").fetchone()["speaker_name"]
|
|
assert name == "Kainda"
|
|
verify.close()
|
|
|
|
def test_merge_into_self_raises(self, env) -> None:
|
|
db, data_dir = env
|
|
with pytest.raises(ResolutionError, match="itself"):
|
|
merge_speakers(db, data_dir, "Kainda", "kainda")
|
|
|
|
def test_merge_unknown_target_raises(self, env) -> None:
|
|
db, data_dir = env
|
|
with pytest.raises(ResolutionError, match="not found"):
|
|
merge_speakers(db, data_dir, "Guard 1", "Nobody")
|