bf83bb35cc
The remote had one commit this branch did not (03f8a85), which adds the
Apache-2.0 LICENSE, the NOTICE and SPDX headers to every .py and .sh
file. The merge was clean - the headers sit at the top of each file and
my changes are further down.
The new files this branch added (languages.py, translations.py, the
cluster package, the new tests and scripts) were written after the
merge base, so the licence commit never saw them. They now carry the
same header, so the convention holds across the whole tree.
Nothing else changed. 883 tests passed / 8 failed - the same 8
pre-existing failures.
370 lines
13 KiB
Python
370 lines
13 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for source-language tagging and per-language translations (spec 31).
|
|
|
|
The behaviours pinned here are the ones that would silently corrupt audio or
|
|
lose alignment if they regressed: the source/target axis split, the per-language
|
|
voiced flag, idempotent re-delivery, and refusal to overwrite approved text.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from umbraaudio import translations as tr
|
|
|
|
|
|
def _seed(conn: sqlite3.Connection) -> None:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE project (id INTEGER PRIMARY KEY, name TEXT);
|
|
CREATE TABLE chapters (id INTEGER PRIMARY KEY, chapter_index INTEGER);
|
|
CREATE TABLE utterances (
|
|
id INTEGER PRIMARY KEY,
|
|
chapter_id INTEGER,
|
|
utterance_index INTEGER,
|
|
speaker_name TEXT,
|
|
text TEXT,
|
|
context_before TEXT,
|
|
context_after TEXT,
|
|
language_code TEXT,
|
|
language_origin TEXT
|
|
);
|
|
INSERT INTO project (id, name) VALUES (1, 'Test');
|
|
INSERT INTO chapters (id, chapter_index) VALUES (1, 1), (2, 2);
|
|
INSERT INTO utterances (id, chapter_id, utterance_index, speaker_name, text) VALUES
|
|
(1, 1, 0, 'narrator', 'The ship landed.'),
|
|
(2, 1, 1, 'Kabir', 'Hola, amigo.'),
|
|
(3, 1, 2, 'Kabir', 'We should go.'),
|
|
(4, 2, 0, 'narrator', 'Later that day.'),
|
|
(5, 2, 1, 'Nanny', 'Danger.');
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
def conn() -> sqlite3.Connection:
|
|
c = sqlite3.connect(":memory:")
|
|
c.row_factory = sqlite3.Row
|
|
_seed(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def plain_conn() -> sqlite3.Connection:
|
|
"""No row_factory - proves the accessors do not assume sqlite3.Row."""
|
|
c = sqlite3.connect(":memory:")
|
|
_seed(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
# --- source axis ---------------------------------------------------------
|
|
|
|
|
|
def test_source_language_defaults_to_english(conn: sqlite3.Connection) -> None:
|
|
assert tr.source_language(conn) == "en"
|
|
|
|
|
|
def test_source_language_reads_project_column() -> None:
|
|
c = sqlite3.connect(":memory:")
|
|
c.execute("CREATE TABLE project (id INTEGER PRIMARY KEY, source_language TEXT)")
|
|
c.execute("INSERT INTO project VALUES (1, 'ZH')")
|
|
assert tr.source_language(c) == "zh"
|
|
c.close()
|
|
|
|
|
|
def test_set_source_language_normalises_and_reports_change(conn: sqlite3.Connection) -> None:
|
|
assert tr.set_source_language(conn, 2, "Spanish") is True
|
|
row = conn.execute(
|
|
"SELECT language_code, language_origin FROM utterances WHERE id=2"
|
|
).fetchone()
|
|
assert row["language_code"] == "es"
|
|
assert row["language_origin"] == "umbratrans"
|
|
# Setting the same value again is not a change.
|
|
assert tr.set_source_language(conn, 2, "es") is False
|
|
|
|
|
|
def test_set_source_language_unknown_id_is_not_an_error(conn: sqlite3.Connection) -> None:
|
|
assert tr.set_source_language(conn, 9999, "es") is False
|
|
|
|
|
|
def test_set_source_language_rejects_bad_origin(conn: sqlite3.Connection) -> None:
|
|
with pytest.raises(ValueError):
|
|
tr.set_source_language(conn, 1, "es", "nonsense")
|
|
|
|
|
|
def test_operator_tag_outranks_machine_reidentification(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es", tr.lang.ORIGIN_OPERATOR)
|
|
assert tr.set_source_language(conn, 2, "fr", tr.lang.ORIGIN_UMBRATRANS) is False
|
|
row = conn.execute("SELECT language_code FROM utterances WHERE id=2").fetchone()
|
|
assert row["language_code"] == "es"
|
|
|
|
|
|
def test_force_overrides_operator_tag(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es", tr.lang.ORIGIN_OPERATOR)
|
|
assert tr.set_source_language(conn, 2, "fr", tr.lang.ORIGIN_UMBRATRANS, force=True) is True
|
|
row = conn.execute("SELECT language_code FROM utterances WHERE id=2").fetchone()
|
|
assert row["language_code"] == "fr"
|
|
|
|
|
|
def test_foreign_lines_are_excluded_from_translation(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es")
|
|
assert tr.foreign_utterance_ids(conn) == {2}
|
|
to_translate = tr.utterances_to_translate(conn)
|
|
assert 2 not in to_translate
|
|
assert to_translate == [1, 3, 4, 5]
|
|
|
|
|
|
def test_untagged_rows_count_as_source_language(conn: sqlite3.Connection) -> None:
|
|
# NULL tag = the project's primary language (spec 31 section 6.1).
|
|
assert tr.foreign_utterance_ids(conn) == set()
|
|
assert tr.utterances_to_translate(conn) == [1, 2, 3, 4, 5]
|
|
|
|
|
|
def test_effective_language_resolves_null_to_primary(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es")
|
|
assert tr.effective_language(conn, 1) == "en"
|
|
assert tr.effective_language(conn, 2) == "es"
|
|
|
|
|
|
def test_unknown_tag_never_falls_back_to_the_primary_language(conn: sqlite3.Connection) -> None:
|
|
"""A Klingon line must not be spoken as English.
|
|
|
|
This is the fail-loud guarantee (spec 31 section 4.3). If an unrecognised tag
|
|
collapsed to NULL, the line would look untagged, resolve to 'en', and be
|
|
voiced with English phonology - wrong, and silent about being wrong.
|
|
"""
|
|
tr.set_source_language(conn, 1, "klingon")
|
|
assert tr.effective_language(conn, 1) == "klingon"
|
|
assert 1 in tr.foreign_utterance_ids(conn), "an unreadable language is not the primary one"
|
|
assert not tr.lang.is_voiceable(tr.effective_language(conn, 1))
|
|
|
|
|
|
# --- target axis ---------------------------------------------------------
|
|
|
|
|
|
def test_import_stores_a_translation(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(
|
|
conn,
|
|
1,
|
|
{
|
|
"target_language": "de",
|
|
"utterances": [
|
|
{"utterance_id": 1, "source_language": "en", "text": "Das Schiff landete."}
|
|
],
|
|
},
|
|
)
|
|
assert report.accepted == 1
|
|
assert report.source_tagged == 1
|
|
row = conn.execute(
|
|
"SELECT language_name, text, status, origin, voiced"
|
|
" FROM utterance_translations WHERE utterance_id=1"
|
|
).fetchone()
|
|
assert row["language_name"] == "German"
|
|
assert row["text"] == "Das Schiff landete."
|
|
assert row["status"] == "draft"
|
|
assert row["origin"] == "umbratrans"
|
|
assert row["voiced"] == 1, "a normal translation is voiced by default"
|
|
|
|
|
|
def test_import_is_idempotent_on_redelivery(conn: sqlite3.Connection) -> None:
|
|
payload = {"target_language": "de", "utterances": [{"utterance_id": 1, "text": "Erste"}]}
|
|
tr.import_translations(conn, 1, payload)
|
|
payload["utterances"][0]["text"] = "Zweite"
|
|
tr.import_translations(conn, 1, payload)
|
|
rows = conn.execute("SELECT text FROM utterance_translations WHERE utterance_id=1").fetchall()
|
|
assert len(rows) == 1, "re-delivery updates in place, it does not accumulate"
|
|
assert rows[0]["text"] == "Zweite"
|
|
|
|
|
|
def test_confirmed_translation_is_not_overwritten(conn: sqlite3.Connection) -> None:
|
|
tr.import_translations(
|
|
conn,
|
|
1,
|
|
{"target_language": "de", "utterances": [{"utterance_id": 1, "text": "Approved"}]},
|
|
confirm=True,
|
|
)
|
|
report = tr.import_translations(
|
|
conn, 1, {"target_language": "de", "utterances": [{"utterance_id": 1, "text": "Sneaky"}]}
|
|
)
|
|
assert report.skipped_confirmed == 1
|
|
assert report.accepted == 0
|
|
row = conn.execute(
|
|
"SELECT text, status FROM utterance_translations WHERE utterance_id=1"
|
|
).fetchone()
|
|
assert row["text"] == "Approved"
|
|
assert row["status"] == "confirmed"
|
|
|
|
|
|
def test_foreign_line_returns_identification_only(conn: sqlite3.Connection) -> None:
|
|
"""The whole point of the source axis: a Spanish line is not translated."""
|
|
report = tr.import_translations(
|
|
conn,
|
|
1,
|
|
{"target_language": "de", "utterances": [{"utterance_id": 2, "source_language": "es"}]},
|
|
)
|
|
assert report.source_tagged == 1
|
|
assert report.foreign_untouched == 1
|
|
assert report.accepted == 0
|
|
assert (
|
|
conn.execute("SELECT COUNT(*) FROM utterance_translations WHERE utterance_id=2").fetchone()[
|
|
0
|
|
]
|
|
== 0
|
|
)
|
|
assert tr.effective_language(conn, 2) == "es"
|
|
|
|
|
|
def test_collapse_orphan_is_kept_with_a_reason(conn: sqlite3.Connection) -> None:
|
|
"""Three source sentences collapsing into two leaves one with no target."""
|
|
report = tr.import_translations(
|
|
conn,
|
|
1,
|
|
{
|
|
"target_language": "de",
|
|
"utterances": [
|
|
{"utterance_id": 1, "text": "Das Schiff landete. Wir gingen."},
|
|
{"utterance_id": 3, "voiced": False, "reason": "absorbed into utterance 1"},
|
|
],
|
|
},
|
|
)
|
|
assert report.accepted == 2
|
|
assert report.not_voiced == 1
|
|
row = conn.execute(
|
|
"SELECT voiced, review_reason FROM utterance_translations WHERE utterance_id=3"
|
|
).fetchone()
|
|
assert row["voiced"] == 0
|
|
assert row["review_reason"] == "absorbed into utterance 1"
|
|
orphans = tr.not_voiced_rows(conn, "de")
|
|
assert [o["utterance_id"] for o in orphans] == [3]
|
|
|
|
|
|
def test_gap_language_tag_is_stored_and_render_refused(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(
|
|
conn,
|
|
1,
|
|
{"target_language": "de", "utterances": [{"utterance_id": 1, "source_language": "nl"}]},
|
|
)
|
|
assert report.refused_voice == 1
|
|
assert tr.effective_language(conn, 1) == "nl"
|
|
refused = tr.refused_render_rows(conn)
|
|
assert len(refused) == 1
|
|
assert refused[0]["language_code"] == "nl"
|
|
assert "Dutch" in refused[0]["reason"]
|
|
|
|
|
|
def test_unknown_language_is_reported_not_crashed(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(
|
|
conn,
|
|
1,
|
|
{
|
|
"target_language": "de",
|
|
"utterances": [{"utterance_id": 1, "source_language": "klingon"}],
|
|
},
|
|
)
|
|
assert "klingon" in report.unknown_language
|
|
assert tr.refused_render_rows(conn)[0]["reason"].startswith("unrecognised")
|
|
|
|
|
|
def test_identification_only_payload_needs_no_target_language(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(
|
|
conn, 1, {"utterances": [{"utterance_id": 2, "source_language": "es"}]}
|
|
)
|
|
assert report.source_tagged == 1
|
|
assert report.problems == []
|
|
|
|
|
|
def test_text_without_target_language_is_a_problem(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(conn, 1, {"utterances": [{"utterance_id": 1, "text": "Hallo"}]})
|
|
assert report.problems and "target_language" in report.problems[0]
|
|
|
|
|
|
def test_row_without_utterance_id_is_reported(conn: sqlite3.Connection) -> None:
|
|
report = tr.import_translations(
|
|
conn, 1, {"target_language": "de", "utterances": [{"text": "x"}]}
|
|
)
|
|
assert report.problems and "utterance_id" in report.problems[0]
|
|
|
|
|
|
def test_utterances_must_be_a_list(conn: sqlite3.Connection) -> None:
|
|
with pytest.raises(ValueError):
|
|
tr.import_translations(conn, 1, {"target_language": "de", "utterances": {"a": 1}})
|
|
|
|
|
|
def test_plain_tuple_connection_is_supported(plain_conn: sqlite3.Connection) -> None:
|
|
"""No row_factory set - the accessors must still work."""
|
|
tr.set_source_language(plain_conn, 2, "es")
|
|
assert tr.effective_language(plain_conn, 2) == "es"
|
|
report = tr.import_translations(
|
|
plain_conn,
|
|
1,
|
|
{"target_language": "de", "utterances": [{"utterance_id": 1, "text": "Hallo"}]},
|
|
)
|
|
assert report.accepted == 1
|
|
|
|
|
|
# --- coverage and export -------------------------------------------------
|
|
|
|
|
|
def test_coverage_counts_the_gap_honestly(conn: sqlite3.Connection) -> None:
|
|
tr.import_translations(
|
|
conn,
|
|
1,
|
|
{
|
|
"target_language": "de",
|
|
"utterances": [
|
|
{"utterance_id": 1, "text": "Eins"},
|
|
{"utterance_id": 3, "voiced": False, "reason": "merged"},
|
|
],
|
|
},
|
|
)
|
|
cov = tr.coverage(conn, "de")
|
|
assert cov["total"] == 5
|
|
assert cov["delivered"] == 2
|
|
assert cov["voiced"] == 1
|
|
assert cov["not_voiced"] == 1
|
|
assert cov["missing"] == 3
|
|
|
|
|
|
def test_coverage_of_an_untouched_language_is_all_missing(conn: sqlite3.Connection) -> None:
|
|
cov = tr.coverage(conn, "fr")
|
|
assert cov["delivered"] == 0 and cov["missing"] == 5 and cov["not_voiced"] == 0
|
|
|
|
|
|
def test_export_payload_shape(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es")
|
|
payload = tr.build_export_payload(conn)
|
|
assert payload["book_language"] == "en"
|
|
assert payload["utterance_count"] == 5
|
|
assert payload["foreign_count"] == 1
|
|
first = payload["utterances"][0]
|
|
assert first["source_text"] == "The ship landed."
|
|
assert first["language_code"] is None, "untagged lines are null on the first pass"
|
|
spanish = [u for u in payload["utterances"] if u["utterance_id"] == 2][0]
|
|
assert spanish["language_code"] == "es", "a known tag is sent so it is not re-identified"
|
|
|
|
|
|
def test_export_can_exclude_foreign_lines(conn: sqlite3.Connection) -> None:
|
|
tr.set_source_language(conn, 2, "es")
|
|
payload = tr.build_export_payload(conn, include_foreign=False)
|
|
assert payload["utterance_count"] == 4
|
|
assert 2 not in [u["utterance_id"] for u in payload["utterances"]]
|
|
assert payload["foreign_count"] == 1, "the count is still reported even when omitted"
|
|
|
|
|
|
def test_ensure_tables_is_idempotent(conn: sqlite3.Connection) -> None:
|
|
tr.ensure_tables(conn)
|
|
tr.ensure_tables(conn)
|
|
assert (
|
|
conn.execute(
|
|
"SELECT COUNT(*) FROM sqlite_master WHERE name='utterance_translations'"
|
|
).fetchone()[0]
|
|
== 1
|
|
)
|