File size: 3,037 Bytes
35d483e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | from __future__ import annotations
import sys
import unittest
from copy import deepcopy
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from turn_detection.data import GroupingConfig, attach_group_ids, derive_group_keys # noqa: E402
class LeakageGroupingTests(unittest.TestCase):
def test_transitive_audio_and_speaker_links(self) -> None:
raw = [
{"id": "a", "dataset": "d", "speaker_id": "speaker-one", "audio_sha256": "x"},
{"id": "b", "dataset": "d", "speaker_id": "speaker-two", "audio_sha256": "x"},
{"id": "c", "dataset": "d", "speaker_id": "speaker-two", "audio_sha256": "y"},
{"id": "d", "dataset": "d", "speaker_id": "speaker-three", "audio_sha256": "z"},
]
rows = [
{
**record,
"group_keys": derive_group_keys(record, config=GroupingConfig(include_text=False)),
}
for record in raw
]
attach_group_ids(rows)
self.assertEqual(rows[0]["group_id"], rows[1]["group_id"])
self.assertEqual(rows[1]["group_id"], rows[2]["group_id"])
self.assertNotEqual(rows[2]["group_id"], rows[3]["group_id"])
def test_group_ids_are_order_independent(self) -> None:
rows = [
{"record_id": "a", "group_keys": ["audio:x", "speaker:1"]},
{"record_id": "b", "group_keys": ["speaker:1"]},
{"record_id": "c", "group_keys": ["audio:z"]},
]
forward = attach_group_ids(deepcopy(rows))
reverse = attach_group_ids(list(reversed(deepcopy(rows))))
forward_ids = {row["record_id"]: row["group_id"] for row in forward}
reverse_ids = {row["record_id"]: row["group_id"] for row in reverse}
self.assertEqual(forward_ids, reverse_ids)
def test_identifiers_are_hashed_in_manifest_keys(self) -> None:
keys = derive_group_keys(
{"id": "record-1", "dataset": "source", "speaker_id": "private-speaker"}
)
joined = " ".join(keys)
self.assertNotIn("private-speaker", joined)
self.assertTrue(any(key.startswith("speaker:") for key in keys))
def test_repeated_nontrivial_text_links_prompts(self) -> None:
first = derive_group_keys(
{"id": "1", "dataset": "a", "spoken_text": "mera order kidhar hai please"}
)
second = derive_group_keys(
{"id": "2", "dataset": "b", "spoken_text": "Mera order kidhar hai, please?"}
)
self.assertTrue(set(key for key in first if key.startswith("text:")) & set(second))
def test_empty_keys_get_unique_stable_fallbacks(self) -> None:
rows = [
{"record_id": "a", "source_file": "x", "source_row": 0, "group_keys": []},
{"record_id": "b", "source_file": "x", "source_row": 1, "group_keys": []},
]
attach_group_ids(rows)
self.assertNotEqual(rows[0]["group_id"], rows[1]["group_id"])
if __name__ == "__main__":
unittest.main()
|