140 lines
5.6 KiB
Python
140 lines
5.6 KiB
Python
"""Acceptance tests for the shared Q1 alignment contract."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
Q1 = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(Q1))
|
|
from adapter import AlignmentError, Q1AlignmentAdapter # noqa: E402
|
|
from adapter.projection import project_intervals # noqa: E402
|
|
|
|
|
|
def cells(n: int) -> np.ndarray:
|
|
left = np.arange(n, dtype=np.float64) / n
|
|
return np.column_stack((left, left + 1 / n))
|
|
|
|
|
|
def record(text_length: int = 50, audio_length: int = 50, vision_length: int = 50):
|
|
text = np.zeros((50, 768), np.float32)
|
|
audio = np.zeros((500, 74), np.float32)
|
|
vision = np.zeros((500, 35), np.float32)
|
|
text[:text_length] = 1
|
|
audio[:audio_length] = 2
|
|
vision[:vision_length] = 3
|
|
return {"id": "video$_$clip", "sequence_order_verified": True,
|
|
"attention_mask": np.arange(50) < text_length,
|
|
"text": text, "audio": audio, "vision": vision,
|
|
"audio_length": audio_length, "vision_length": vision_length}
|
|
|
|
|
|
class UnifiedAdapterTests(unittest.TestCase):
|
|
def test_01_identity_L_equals_K(self):
|
|
x = np.arange(50, dtype=np.float32)[:, None]
|
|
result = project_intervals(x, cells(50), cells(50), np.ones(50, bool))
|
|
np.testing.assert_array_equal(result.x[:, 0], x[:, 0])
|
|
|
|
def test_02_expand_L_less_than_K(self):
|
|
x = np.arange(10, dtype=np.float32)[:, None]
|
|
result = project_intervals(x, cells(10), cells(50), np.ones(10, bool))
|
|
np.testing.assert_allclose(result.x[:, 0], np.repeat(x[:, 0], 5))
|
|
|
|
def test_03_aggregate_L_greater_than_K(self):
|
|
x = np.arange(100, dtype=np.float32)[:, None]
|
|
result = project_intervals(x, cells(100), cells(50), np.ones(100, bool))
|
|
np.testing.assert_allclose(result.x[:, 0], x.reshape(50, 2).mean(axis=1))
|
|
|
|
def test_04_partial_missing(self):
|
|
x = np.ones((50, 2), np.float32)
|
|
valid = np.ones((50, 2), bool)
|
|
valid[3, 0] = False
|
|
result = project_intervals(x, cells(50), cells(50), valid)
|
|
self.assertFalse(result.observed_dimensions[3, 0])
|
|
self.assertTrue(result.observed_dimensions[3, 1])
|
|
self.assertEqual(result.x[3, 0], 0)
|
|
|
|
def test_05_all_missing(self):
|
|
result = project_intervals(np.ones((10, 2), np.float32), cells(10), cells(50), np.zeros(10, bool))
|
|
self.assertFalse(result.observed.any())
|
|
self.assertFalse(result.x.any())
|
|
self.assertFalse(result.source_weights.nnz)
|
|
|
|
def test_06_text_padding_excluded(self):
|
|
r = record(text_length=10)
|
|
r["text"][10:] = 99
|
|
sample = Q1AlignmentAdapter().align(r)
|
|
self.assertTrue(np.all(sample.features["text"] == 1))
|
|
self.assertEqual(sample.provenance["text"].source_span, 10)
|
|
|
|
def test_07_audio_length_conflict_rejected(self):
|
|
r = record()
|
|
r["audio"][70] = 7
|
|
with self.assertRaisesRegex(AlignmentError, "audio contains observed"):
|
|
Q1AlignmentAdapter().align(r)
|
|
|
|
def test_08_vision_length_conflict_retained(self):
|
|
r = record()
|
|
r["vision"][70] = 7
|
|
p = Q1AlignmentAdapter().align(r).provenance["vision"]
|
|
self.assertTrue(p.length_conflict)
|
|
self.assertTrue(p.tail_ambiguous)
|
|
self.assertEqual(p.source_span, 71)
|
|
|
|
def test_09_provenance_conservation(self):
|
|
sample = Q1AlignmentAdapter().align(record())
|
|
for m, p in sample.provenance.items():
|
|
sums = np.asarray(p.source_weights.sum(axis=1)).ravel()
|
|
np.testing.assert_allclose(sums[sample.observed[m]], 1, atol=1e-6)
|
|
|
|
def test_10_physical_coordinate(self):
|
|
from q1_io import load_sample
|
|
sample = Q1AlignmentAdapter().align(load_sample("-iRBcNs9oI8/8"))
|
|
self.assertEqual(sample.metadata["coordinate_mode"], "physical")
|
|
self.assertTrue(sample.metadata["physical_time_alignment"])
|
|
self.assertAlmostEqual(sample.target_intervals[-1, 1], sample.metadata["duration_s"])
|
|
|
|
def test_11_relative_coordinate(self):
|
|
sample = Q1AlignmentAdapter().align(record())
|
|
self.assertEqual(sample.metadata["coordinate_unit"], "normalized_progress")
|
|
self.assertFalse(sample.metadata["physical_time_alignment"])
|
|
self.assertAlmostEqual(sample.target_intervals[-1, 1], 1.0)
|
|
|
|
def test_12_q2_shapes(self):
|
|
sample = Q1AlignmentAdapter().align(record())
|
|
features, mask = sample.q2_arrays()
|
|
self.assertEqual(mask.shape, (50, 3))
|
|
self.assertEqual({m: v.shape for m, v in features.items()},
|
|
{"text": (50, 768), "audio": (50, 74), "vision": (50, 35)})
|
|
|
|
def test_13_nonfinite_rejected(self):
|
|
r = record()
|
|
r["audio"][0, 0] = np.inf
|
|
with self.assertRaises(AlignmentError):
|
|
Q1AlignmentAdapter().align(r)
|
|
|
|
def test_14_old_new_full_equivalence(self):
|
|
report = json.loads((Q1 / "results" / "unified_adapter" / "full_audit.json").read_text())
|
|
for split in ("train", "valid", "test"):
|
|
self.assertTrue(all(report["relative"][split]["exact_feature_equivalence"].values()))
|
|
self.assertTrue(report["relative"][split]["exact_mask_equivalence"])
|
|
|
|
def test_15_auto_requires_evidence(self):
|
|
r = record()
|
|
r.pop("sequence_order_verified")
|
|
with self.assertRaisesRegex(AlignmentError, "requires"):
|
|
Q1AlignmentAdapter().align(r)
|
|
|
|
def test_16_invalid_physical_does_not_fall_back(self):
|
|
r = record()
|
|
r["_meta"] = {"duration_s": 2.0}
|
|
with self.assertRaisesRegex(AlignmentError, "verified media"):
|
|
Q1AlignmentAdapter().align(r)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|