File size: 2,585 Bytes
784f572 | 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 74 75 | from pathlib import Path
import pytest
from dlb.matrix import (
MATRIX_COLUMNS,
MATRIX_SCHEMA,
build_matrix,
read_matrix,
unsupported_inventory,
validate_matrix,
write_matrix,
write_unsupported_inventory,
)
from dlb.registry import load_registry
ROOT = Path(__file__).parents[1]
@pytest.fixture
def registry():
return load_registry(ROOT / "configs/experiments.yaml")
def test_matrix_cardinality_and_declared_steps(registry):
tasks = build_matrix(registry)
assert len(tasks) == 132
assert sum(task.category == "many" for task in tasks) == 72
assert sum(task.category == "few" for task in tasks) == 60
assert sum(task.category == "fixed_1024" for task in tasks) == 0
assert [
task.steps for task in tasks if (task.model, task.dataset) == ("rdlm", "lm1b")
] == [1000, 1024]
for task in tasks:
model = registry.models[task.model]
assert task.steps in (model.step_override or registry.step_grids[task.category])
assert task.sample_count == 1024
assert task.seed == 42
def test_matrix_is_stably_sorted_and_round_trips_without_eval(registry, tmp_path):
tasks = build_matrix(registry, root=tmp_path)
output = write_matrix(tmp_path / "generation.tsv", tasks)
content = output.read_text(encoding="utf-8")
assert content.splitlines()[0] == f"# schema={MATRIX_SCHEMA}"
assert content.splitlines()[1].split("\t") == list(MATRIX_COLUMNS)
restored = read_matrix(output)
assert restored == tasks
assert validate_matrix(output, registry) == tasks
assert len({task.task_id for task in restored}) == 132
assert "eval(" not in content
def test_matrix_rejects_duplicate_task_ids(tmp_path, registry):
tasks = build_matrix(registry, root=tmp_path)
output = write_matrix(tmp_path / "generation.tsv", tasks)
lines = output.read_text(encoding="utf-8").splitlines()
lines.append(lines[2])
output.write_text("\n".join(lines) + "\n", encoding="utf-8")
with pytest.raises(ValueError, match="duplicate"):
read_matrix(output)
def test_unsupported_inventory_is_not_expanded(registry, tmp_path):
records = unsupported_inventory(registry)
assert {(item["model"], item["dataset"]) for item in records} == {
("rdlm", "owt"),
}
assert all(item["reason"] for item in records)
output = write_unsupported_inventory(tmp_path / "unsupported.tsv", records)
lines = output.read_text(encoding="utf-8").splitlines()
assert lines[0] == f"# schema={MATRIX_SCHEMA}-unsupported"
assert len(lines) == 3
|