File size: 5,742 Bytes
d61821a | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | from __future__ import annotations
from pathlib import Path
import unittest
from agent_harness.specs import (
load_agent_systems,
load_edit_interfaces,
load_experiments,
load_harnesses,
load_models,
load_repositories,
load_task_split,
load_tasks,
validate_configuration_tree,
)
ROOT = Path(__file__).resolve().parents[1]
class HarnessCatalogTests(unittest.TestCase):
def test_catalog_is_complete_and_hashes_are_unique(self) -> None:
harnesses = load_harnesses(ROOT)
self.assertEqual(list(harnesses), [f"H{number:03d}" for number in range(21)])
self.assertEqual(len({item.config_hash for item in harnesses.values()}), 21)
self.assertEqual(len({item.treatment_hash for item in harnesses.values()}), 21)
def test_core_retrieval_block_is_full_factorial(self) -> None:
harnesses = load_harnesses(ROOT)
observed = {
(
harnesses[f"H{number:03d}"].lexical,
harnesses[f"H{number:03d}"].syntax == "tree_sitter",
harnesses[f"H{number:03d}"].dense,
)
for number in range(8)
}
expected = {
(lexical, syntax, dense)
for lexical in (False, True)
for syntax in (False, True)
for dense in (False, True)
}
self.assertEqual(observed, expected)
def test_control_harnesses_have_no_retrieval_capabilities(self) -> None:
harnesses = load_harnesses(ROOT)
for identifier in ("H016", "H017", "H018", "H019"):
harness = harnesses[identifier]
self.assertFalse(harness.exact_search)
self.assertFalse(harness.lexical)
self.assertFalse(harness.dense)
self.assertEqual(harness.syntax, "raw")
self.assertEqual(harness.graph_hops, 0)
def test_configuration_tree_is_valid_and_embedding_is_ready(self) -> None:
errors, warnings = validate_configuration_tree(ROOT)
self.assertEqual(errors, [])
self.assertFalse(any("dense retrieval" in warning for warning in warnings))
self.assertEqual(warnings, [])
def test_fixed_model_and_end_to_end_plan(self) -> None:
models = load_models(ROOT)
experiments = load_experiments(ROOT)
self.assertEqual(models["M001"].canonical_name, "Qwen3.6-35B-A3B")
self.assertEqual(experiments["E03"].cells_per_task(), 10)
self.assertEqual(experiments["E03"].model_ids, ("M001",))
self.assertEqual(models["M002"].context_length, 65536)
self.assertEqual(experiments["E07"].cells_per_task(), 7)
self.assertEqual(experiments["E07"].model_ids, ("M002",))
def test_task_template_has_a_valid_schema_but_is_not_in_a_split(self) -> None:
tasks = load_tasks(ROOT)
self.assertEqual(len(tasks), 108)
self.assertEqual(
{item for item in tasks if item.startswith("TASK_CR_")},
{f"TASK_CR_{number:03d}" for number in range(1, 16)},
)
self.assertEqual(tasks["TASK_EXAMPLE"].base_commit, "0" * 40)
for task_id in ("TASK_GR_001", "TASK_GR_002", "TASK_GR_003", "TASK_GR_004", "TASK_GR_005"):
self.assertEqual(tasks[task_id].validation_status, "retrieval_ready")
self.assertEqual(tasks[task_id].repository_url, "https://gitlab.com/gitlab-org/gitlab-runner.git")
end_to_end = {"TASK_CR_001", "TASK_CR_002", "TASK_CR_003", "TASK_CR_005", "TASK_CR_006", "TASK_CR_007", "TASK_CR_008", "TASK_CR_009", "TASK_CR_012", "TASK_CR_013"}
for task_id in end_to_end:
self.assertEqual(tasks[task_id].validation_status, "end_to_end_ready")
self.assertTrue(tasks[task_id].gold_patch)
self.assertTrue(tasks[task_id].test_patch)
def test_study2_catalog_and_powered_matrix_are_frozen(self) -> None:
systems = load_agent_systems(ROOT)
repositories = load_repositories(ROOT)
experiments = load_experiments(ROOT)
split = load_task_split(ROOT / "tasks" / "splits" / "study2_confirmatory.txt")
self.assertEqual(set(systems), {"A001", "A002"})
self.assertEqual(set(repositories), {"R001", "R002", "R003"})
self.assertEqual(len(split), 60)
self.assertEqual(experiments["E08"].cells_per_task(), 14)
self.assertEqual(experiments["E08"].cells_per_task() * len(split), 840)
def test_study3_edit_interfaces_are_separate_and_frozen(self) -> None:
interfaces = load_edit_interfaces(ROOT)
self.assertEqual(set(interfaces), {"P001", "P002", "P003"})
self.assertEqual(
{item.edit_tool for item in interfaces.values()},
{"apply_patch", "replace_text", "write_file"},
)
self.assertEqual(len({item.treatment_hash for item in interfaces.values()}), 3)
def test_study4_fresh_task_matrix_is_frozen(self) -> None:
tasks = load_tasks(ROOT)
experiments = load_experiments(ROOT)
split = load_task_split(ROOT / "tasks" / "splits" / "study4_fresh.txt")
self.assertEqual(len(split), 20)
self.assertEqual(
{task_id.split("_")[2] for task_id in split}, {"R001", "R002", "R003"}
)
self.assertEqual(
[sum(task_id.startswith(f"TASK_S4_{repo}_") for task_id in split) for repo in ("R001", "R002", "R003")],
[3, 7, 10],
)
self.assertTrue(all(tasks[task_id].validation_status == "end_to_end_ready" for task_id in split))
experiment = experiments["E10"]
self.assertEqual(experiment.harness_ids, ("H000", "H007", "H018"))
self.assertEqual(experiment.model_ids, ("M002", "M003", "M004"))
if __name__ == "__main__":
unittest.main()
|