141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(MODULE_DIR))
|
|
|
|
import purpose_data
|
|
|
|
|
|
def example(index: int, **overrides):
|
|
value = {
|
|
"prompt": f"Implement sample endpoint number {index} with stable pagination",
|
|
"purpose": purpose_data.LABELS[index % len(purpose_data.LABELS)],
|
|
"secondary": None,
|
|
"mixed": False,
|
|
"difficulty": 0.4,
|
|
"slice": "core",
|
|
"lang": "en",
|
|
}
|
|
value.update(overrides)
|
|
return value
|
|
|
|
|
|
def source(value, line=1):
|
|
return purpose_data.SourceRecord(value, Path("source.jsonl"), line)
|
|
|
|
|
|
class NormalizationTests(unittest.TestCase):
|
|
def test_normalization_matches_runtime_whitespace_contract(self):
|
|
self.assertEqual(
|
|
"Café deploy now",
|
|
purpose_data.normalize_prompt(" Cafe\u0301\tdeploy\nnow "),
|
|
)
|
|
self.assertEqual(
|
|
purpose_data.normalized_key("FIX spacing"),
|
|
purpose_data.normalized_key(" fix spacing "),
|
|
)
|
|
|
|
|
|
class CurationTests(unittest.TestCase):
|
|
def test_excludes_exact_fixture_overlap(self):
|
|
record = source(example(0, prompt="Make the toolbar nicer", purpose="frontendImpl"))
|
|
result = purpose_data.curate_records(
|
|
[record],
|
|
[{"prompt": " make the toolbar nicer ", "purpose": "frontendImpl"}],
|
|
)
|
|
self.assertEqual([], result.records)
|
|
self.assertEqual("exact", result.duplicates[0].kind)
|
|
|
|
def test_excludes_high_overlap_generated_template(self):
|
|
words = [f"token{index}" for index in range(100)]
|
|
first = " ".join(words)
|
|
words[50] = "replacement"
|
|
second = " ".join(words)
|
|
result = purpose_data.curate_records(
|
|
[
|
|
source(example(0, prompt=first, purpose="backendImpl"), 1),
|
|
source(example(1, prompt=second, purpose="backendImpl"), 2),
|
|
],
|
|
[],
|
|
near_duplicate_threshold=0.92,
|
|
)
|
|
self.assertEqual(1, len(result.records))
|
|
self.assertEqual(1, len(result.duplicates))
|
|
self.assertEqual("near", result.duplicates[0].kind)
|
|
self.assertGreaterEqual(result.duplicates[0].similarity, 0.92)
|
|
|
|
def test_near_duplicate_label_conflict_requires_review(self):
|
|
words = [f"token{index}" for index in range(100)]
|
|
first = " ".join(words)
|
|
words[50] = "replacement"
|
|
second = " ".join(words)
|
|
with self.assertRaisesRegex(
|
|
purpose_data.DataError, "label conflicts require review"
|
|
):
|
|
purpose_data.curate_records(
|
|
[
|
|
source(example(0, prompt=first, purpose="backendImpl"), 1),
|
|
source(example(1, prompt=second, purpose="writing"), 2),
|
|
],
|
|
[],
|
|
near_duplicate_threshold=0.92,
|
|
)
|
|
|
|
def test_reviewed_semantic_exclusions_fail_closed_on_corpus_drift(self):
|
|
first = source(example(0, prompt="Plan the cache migration", purpose="planning"), 1)
|
|
second = source(
|
|
example(1, prompt="Propose a cache migration plan", purpose="planning"), 2
|
|
)
|
|
decision = {
|
|
"droppedPromptHash": purpose_data.prompt_hash(second.value["prompt"]),
|
|
"matchedPromptHash": purpose_data.prompt_hash(first.value["prompt"]),
|
|
"similarity": 0.98,
|
|
}
|
|
result = purpose_data.exclude_reviewed_duplicates(
|
|
[first, second], [decision]
|
|
)
|
|
self.assertEqual([first], result.records)
|
|
self.assertEqual("semantic-reviewed", result.duplicates[0].kind)
|
|
with self.assertRaisesRegex(
|
|
purpose_data.DataError, "no longer matches the curated corpus"
|
|
):
|
|
purpose_data.exclude_reviewed_duplicates([first], [decision])
|
|
|
|
|
|
class SplitTests(unittest.TestCase):
|
|
def test_split_is_deterministic_stratified_and_keeps_vague_out_of_train(self):
|
|
records = []
|
|
for index in range(1_000):
|
|
slice_name = "vague-eval" if index < 50 else (
|
|
"boundary" if index % 5 == 0 else "core"
|
|
)
|
|
records.append(source(example(index, slice=slice_name), index + 1))
|
|
|
|
first = purpose_data.split_records(records, fixture_count=10, seed=17)
|
|
second = purpose_data.split_records(records, fixture_count=10, seed=17)
|
|
|
|
self.assertEqual(
|
|
[row.value["prompt"] for row in first.train],
|
|
[row.value["prompt"] for row in second.train],
|
|
)
|
|
self.assertEqual(808, len(first.train))
|
|
self.assertEqual(101, len(first.validation))
|
|
self.assertEqual(101, first.logical_test_count)
|
|
self.assertFalse(
|
|
any(row.value["slice"] == "vague-eval" for row in first.train)
|
|
)
|
|
split_prompts = [
|
|
{row.value["prompt"] for row in split}
|
|
for split in (first.train, first.validation, first.test)
|
|
]
|
|
self.assertFalse(split_prompts[0] & split_prompts[1])
|
|
self.assertFalse(split_prompts[0] & split_prompts[2])
|
|
self.assertFalse(split_prompts[1] & split_prompts[2])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|