| import copy |
| from pathlib import Path |
| import pytest |
| from common import read_json,write_json,is_validation_group |
| from benchmarks.bfcl_tasks import validate_sample,replay_sample,parent_correction_sample |
| from benchmarks.bfcl_data import add_replay,readiness |
| from learning.trajectories import validate_pool |
| from helpers import generated_fixture,make_sample,row |
|
|
|
|
| def test_guided_success_preserves_bound_failure(monkeypatch,tmp_path,config): |
| f=generated_fixture(monkeypatch,tmp_path,config) |
| s=make_sample(f) |
| assert s['pair_kind']=='guided_positive' |
| assert s['failure_labels'] |
| assert s['verification']['first_failure']['episode']==str(f['bad']) |
| assert 'rejected' not in s |
| assert validate_sample(s,{('bfcl',f['parent']['task_id']):f['parent']}) |
| validate_pool([s],[f['parent']]) |
| assert readiness([s],set(),config)['correction_samples']==0 |
|
|
|
|
| def test_same_prompt_is_a_correction_pair(monkeypatch,tmp_path,config): |
| f=generated_fixture(monkeypatch,tmp_path,config,source='student_retry',same_prompt=True) |
| s=make_sample(f) |
| assert s['pair_kind']=='same_prompt' and s['rejected'] |
| assert validate_sample(s) |
| assert readiness([s],set(),config)['correction_samples']==1 |
|
|
|
|
| def test_non_harness_guidance_change_cannot_be_laundered(monkeypatch,tmp_path,config): |
| f=generated_fixture(monkeypatch,tmp_path,config,source='student_retry') |
| with pytest.raises(ValueError,match='Only an explicit temporary Harness'): |
| make_sample(f) |
|
|
|
|
| @pytest.mark.parametrize('mutation',['target','first_failure','category','group']) |
| def test_sample_tampering_rejected(monkeypatch,tmp_path,config,mutation): |
| f=generated_fixture(monkeypatch,tmp_path,config) |
| s=make_sample(f) |
| if mutation=='target':s['target']['content']='[]' |
| elif mutation=='first_failure':s['verification']['first_failure']=None |
| elif mutation=='category':s['category']='parallel' |
| else:s['split_group']='0'*64 |
| with pytest.raises(ValueError):validate_sample(s) |
|
|
|
|
| def test_first_failure_integrity_is_not_silently_dropped(monkeypatch,tmp_path,config): |
| f=generated_fixture(monkeypatch,tmp_path,config) |
| bad=read_json(f['bad']/'outcome.json');bad['task_id']='unrelated' |
| write_json(f['bad']/'outcome.json',bad) |
| with pytest.raises(ValueError,match='real failed first answer'):make_sample(f) |
|
|
|
|
| @pytest.mark.parametrize('mutation',['teacher_backend','correct_response']) |
| def test_student_failure_cannot_be_fabricated(monkeypatch,tmp_path,config,mutation): |
| f=generated_fixture(monkeypatch,tmp_path,config) |
| generation=read_json(f['bad']/'static-generation.json') |
| if mutation=='teacher_backend':generation['execution_backend']='teacher' |
| else:generation['response']=read_json(f['good']/'static-generation.json')['response'] |
| write_json(f['bad']/'static-generation.json',generation) |
| with pytest.raises(ValueError):make_sample(f) |
|
|
|
|
| def test_replay_is_explicit_and_cannot_cross_roles(): |
| r=row();s=replay_sample(r) |
| assert s['verification']['executed_success_claim'] is False |
| assert s['failure_labels']==[] and 'assistance' not in s |
| assert validate_sample(s,{('bfcl',r['task_id']):r}) |
| validate_pool([s],[r]) |
| for role in ['selection','audit','meta_eval','sealed_test']: |
| with pytest.raises(ValueError):replay_sample(row(role=role)) |
| forged=copy.deepcopy(s);forged['target']['content']='[]' |
| with pytest.raises(ValueError):validate_sample(forged) |
|
|
|
|
| def test_replay_fills_healthy_categories_without_inventing_failure_groups(config): |
| rows=[row(i,c) for c in ('simple_python','multiple','parallel','parallel_multiple') |
| for i in range(100)] |
| failure=rows[0] |
| samples=add_replay([],rows,{('bfcl',failure['task_id'])},config) |
| assert len(samples)==64 |
| status=readiness(samples,set(),config) |
| for coverage in status['category_coverage'].values(): |
| assert coverage['samples']>=8 and coverage['groups']>=4 |
| assert coverage['train_groups'] and coverage['validation_groups'] |
| assert all(s['task']['task_id']!=failure['task_id'] for s in samples) |
| assert not status['ready'] |
| assert status['correction_groups']==0 |
| assert add_replay(samples,rows,{('bfcl',failure['task_id'])},config)==[] |
|
|
