251 lines
9.2 KiB
Python
251 lines
9.2 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for content-addressed source-unit identity (``source_unit_key``).
|
|
|
|
The contract under test is the one the controller depends on (spec 21 §9.1 M4-04):
|
|
a source unit must be resolvable **without** ``utterance_index``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from umbraaudio.source_units import (
|
|
COLUMN,
|
|
KEY_VERSION,
|
|
assign_keys_for_chapter,
|
|
backfill,
|
|
chapter_keys,
|
|
chapter_text_sha256,
|
|
compute_source_unit_key,
|
|
ensure_chapter_keys,
|
|
ensure_column,
|
|
key_report,
|
|
stamp_chapter,
|
|
text_sha256,
|
|
)
|
|
|
|
|
|
def _db(chapters: dict[int, str], utterances: dict[int, list[str]]) -> sqlite3.Connection:
|
|
"""Build a minimal chapters/utterances database."""
|
|
con = sqlite3.connect(":memory:")
|
|
con.executescript(
|
|
"CREATE TABLE chapters (id INTEGER PRIMARY KEY, source_text TEXT, "
|
|
"chapter_index INTEGER);"
|
|
"CREATE TABLE utterances (id INTEGER PRIMARY KEY, chapter_id INTEGER, "
|
|
"utterance_index INTEGER, text TEXT, type TEXT DEFAULT 'narration');"
|
|
)
|
|
for cid, text in chapters.items():
|
|
con.execute(
|
|
"INSERT INTO chapters (id, source_text, chapter_index) VALUES (?, ?, ?)",
|
|
(cid, text, cid),
|
|
)
|
|
utt_id = 0
|
|
for cid, texts in utterances.items():
|
|
for idx, text in enumerate(texts):
|
|
utt_id += 1
|
|
con.execute(
|
|
"INSERT INTO utterances (id, chapter_id, utterance_index, text) VALUES (?,?,?,?)",
|
|
(utt_id, cid, idx, text),
|
|
)
|
|
con.commit()
|
|
return con
|
|
|
|
|
|
class TestKeyDerivation:
|
|
def test_deterministic(self) -> None:
|
|
a = compute_source_unit_key("abc", "Hello.", 0)
|
|
b = compute_source_unit_key("abc", "Hello.", 0)
|
|
assert a == b
|
|
|
|
def test_key_is_hex_sha256(self) -> None:
|
|
key = compute_source_unit_key("abc", "Hello.", 0)
|
|
assert len(key) == 64
|
|
assert all(c in "0123456789abcdef" for c in key)
|
|
|
|
def test_occurrence_changes_key(self) -> None:
|
|
assert compute_source_unit_key("abc", "Yes.", 0) != compute_source_unit_key(
|
|
"abc", "Yes.", 1
|
|
)
|
|
|
|
def test_chapter_hash_changes_key(self) -> None:
|
|
assert compute_source_unit_key("a" * 64, "Hi.", 0) != compute_source_unit_key(
|
|
"b" * 64, "Hi.", 0
|
|
)
|
|
|
|
def test_key_version_is_part_of_the_payload(self) -> None:
|
|
"""A version bump must change every key, so old keys stay distinguishable."""
|
|
expected = hashlib.sha256(
|
|
b"\x00".join(
|
|
(
|
|
KEY_VERSION.encode(),
|
|
("c" * 64).encode(),
|
|
b"0",
|
|
text_sha256("Hi.").encode(),
|
|
)
|
|
)
|
|
).hexdigest()
|
|
assert compute_source_unit_key("c" * 64, "Hi.", 0) == expected
|
|
|
|
def test_chapter_and_text_hashes_are_utf8_stable(self) -> None:
|
|
assert chapter_text_sha256("café") == chapter_text_sha256("café")
|
|
# A chapter with no captured source text must still hash deterministically,
|
|
# and it must be the SAME as an empty string - not random.
|
|
assert chapter_text_sha256(None) == chapter_text_sha256("")
|
|
assert text_sha256(None) == text_sha256("")
|
|
|
|
|
|
class TestRepeatedText:
|
|
def test_identical_lines_get_distinct_keys(self) -> None:
|
|
keys = assign_keys_for_chapter("s" * 64, ["Yes.", "No.", "Yes.", "Yes."])
|
|
assert len(set(keys)) == 4
|
|
|
|
def test_occurrence_is_document_order(self) -> None:
|
|
first = compute_source_unit_key("s" * 64, "Yes.", 0)
|
|
second = compute_source_unit_key("s" * 64, "Yes.", 1)
|
|
keys = assign_keys_for_chapter("s" * 64, ["Yes.", "Yes."])
|
|
assert keys == [first, second]
|
|
|
|
def test_different_chapters_get_different_keys(self) -> None:
|
|
a = assign_keys_for_chapter("a" * 64, ["Same line."])
|
|
b = assign_keys_for_chapter("b" * 64, ["Same line."])
|
|
assert a != b
|
|
|
|
|
|
class TestIndexIndependence:
|
|
def test_renumbering_does_not_change_keys(self) -> None:
|
|
"""The load-bearing property: `utterance_index` must not be load-bearing."""
|
|
chapter = "The dwarf woke. The dwarf rose."
|
|
texts = ["The dwarf woke.", "The dwarf rose."]
|
|
|
|
con = _db({1: chapter}, {1: texts})
|
|
assert ensure_column(con)
|
|
stamp_chapter(con, 1)
|
|
before = dict(con.execute(f"SELECT id, {COLUMN} FROM utterances").fetchall())
|
|
|
|
# Re-segmentation: a new utterance is inserted at position 0 and every
|
|
# index shifts. The original two units keep their identity.
|
|
con.execute(
|
|
"INSERT INTO utterances (id, chapter_id, utterance_index, text) VALUES (99,1,0,?)",
|
|
("Inserted first.",),
|
|
)
|
|
con.execute("UPDATE utterances SET utterance_index = utterance_index + 1 WHERE id != 99")
|
|
con.execute("UPDATE utterances SET utterance_index = 0 WHERE id = 99")
|
|
con.commit()
|
|
|
|
stamp_chapter(con, 1)
|
|
after = dict(con.execute(f"SELECT id, {COLUMN} FROM utterances").fetchall())
|
|
|
|
for utt_id, key in before.items():
|
|
assert after[utt_id] == key
|
|
|
|
def test_chapter_text_edit_invalidates_that_chapters_keys(self) -> None:
|
|
con = _db({1: "Original text."}, {1: ["A line."]})
|
|
stamp_chapter(con, 1)
|
|
first = con.execute(f"SELECT {COLUMN} FROM utterances").fetchone()[0]
|
|
|
|
con.execute("UPDATE chapters SET source_text = 'Edited text.' WHERE id = 1")
|
|
stamp_chapter(con, 1)
|
|
second = con.execute(f"SELECT {COLUMN} FROM utterances").fetchone()[0]
|
|
assert first != second
|
|
|
|
def test_chapter_without_source_text_still_hashes(self) -> None:
|
|
con = _db({1: None}, {1: ["A line."]})
|
|
stamp_chapter(con, 1)
|
|
assert con.execute(f"SELECT {COLUMN} FROM utterances").fetchone()[0]
|
|
|
|
def test_chapter_keys_do_not_need_the_index(self) -> None:
|
|
"""Resolution is by key, exactly as controller validation phase 3 requires."""
|
|
con = _db({1: "Chapter body."}, {1: ["One.", "Two."]})
|
|
ensure_column(con)
|
|
stamp_chapter(con, 1)
|
|
keys = [k for _, k in chapter_keys(con, 1)]
|
|
|
|
resolved = con.execute(
|
|
f"SELECT text FROM utterances WHERE {COLUMN} = ?", (keys[1],)
|
|
).fetchone()
|
|
assert resolved[0] == "Two."
|
|
|
|
|
|
class TestIdempotence:
|
|
def test_stamp_twice_changes_nothing_the_second_time(self) -> None:
|
|
con = _db({1: "Body."}, {1: ["A.", "B."]})
|
|
ensure_column(con)
|
|
stamp_chapter(con, 1)
|
|
assert stamp_chapter(con, 1) == 0
|
|
|
|
def test_ensure_only_fills_missing(self) -> None:
|
|
con = _db({1: "Body."}, {1: ["A.", "B."]})
|
|
ensure_column(con)
|
|
assert ensure_chapter_keys(con, 1) == 2
|
|
assert ensure_chapter_keys(con, 1) == 0
|
|
|
|
def test_column_add_is_idempotent(self) -> None:
|
|
con = _db({1: "Body."}, {1: ["A."]})
|
|
assert ensure_column(con) is True
|
|
assert ensure_column(con) is False
|
|
|
|
def test_backfill_reports_coverage(self) -> None:
|
|
con = _db({1: "Body.", 2: "Body two."}, {1: ["A."], 2: ["B."]})
|
|
result = backfill(con)
|
|
assert result["chapters"] == 2
|
|
assert result["utterances_total"] == 2
|
|
assert result["utterances_with_key"] == 2
|
|
assert result["utterances_without_key"] == 0
|
|
|
|
def test_backfill_can_target_chapters(self) -> None:
|
|
con = _db({1: "A.", 2: "B."}, {1: ["x"], 2: ["y"]})
|
|
result = backfill(con, chapter_ids=[1])
|
|
assert result["utterances_with_key"] == 1
|
|
|
|
|
|
class TestKeyReport:
|
|
def test_absent_column_reports_false(self) -> None:
|
|
con = _db({1: "Body."}, {1: ["A."]})
|
|
assert key_report(con) == {"column_present": False}
|
|
|
|
def test_collisions_are_counted(self) -> None:
|
|
"""Collisions must be visible, not silently accepted."""
|
|
con = _db({1: "Body."}, {1: ["A.", "B."]})
|
|
ensure_column(con)
|
|
stamp_chapter(con, 1)
|
|
# Force a duplicate to prove the report detects it rather than hiding it.
|
|
con.execute(
|
|
"UPDATE utterances SET "
|
|
f"{COLUMN} = (SELECT {COLUMN} FROM utterances ORDER BY id LIMIT 1)"
|
|
)
|
|
con.commit()
|
|
report = key_report(con)
|
|
assert report["column_present"] is True
|
|
assert report["utterances_total"] == 2
|
|
assert report["distinct_keys"] == 1
|
|
assert report["collisions"] == 1
|
|
|
|
|
|
def test_no_index_column_is_written_by_stamp() -> None:
|
|
"""Guard: stamping must not touch utterance_index or any other column."""
|
|
con = _db({1: "Body."}, {1: ["A.", "B."]})
|
|
ensure_column(con)
|
|
con.execute(f"UPDATE utterances SET {COLUMN} = 'stale'")
|
|
con.commit()
|
|
cols = "SELECT id, utterance_index, text, type FROM utterances ORDER BY id"
|
|
before = con.execute(cols).fetchall()
|
|
stamp_chapter(con, 1)
|
|
after = con.execute(cols).fetchall()
|
|
assert before == after
|
|
|
|
|
|
@pytest.mark.parametrize("path_exists", [True, False])
|
|
def test_file_sha256_handles_missing_files(tmp_path, path_exists: bool) -> None:
|
|
from umbraaudio.generation.staleness import file_sha256
|
|
|
|
target = tmp_path / "x.wav"
|
|
if path_exists:
|
|
target.write_bytes(b"abc")
|
|
assert file_sha256(target) == hashlib.sha256(b"abc").hexdigest()
|
|
else:
|
|
assert file_sha256(target) is None
|