| import pytest
|
|
|
| from crosswire.study import RunConfig, run, run_study, summarize
|
|
|
|
|
| def test_configuration_rejects_nonfinite_unbounded_and_unknown_inputs():
|
| for values in ({'budget': True}, {'budget': -1}, {'horizon': 7}, {'seed': -1},
|
| {'policy': 'secret_oracle'}, {'mystery': 2}, {'anchors': 'true'}):
|
| with pytest.raises(ValueError):
|
| RunConfig.from_dict(values)
|
|
|
|
|
| def test_zero_budget_robust_refuses_semantically_ambiguous_connection():
|
| result = run(RunConfig(family='symmetric', budget=0))
|
| assert result['decision']['admitted'] is False
|
| assert result['audit']['contract_satisfied'] is None
|
| assert result['metrics']['target_steps'] == 0
|
| assert result['audit']['oracle_feasible'] is True
|
| assert result['decision_frozen_hash']
|
|
|
|
|
| def test_anchors_and_probes_do_not_drop_true_map_on_supported_case():
|
| result = run(RunConfig(family='latch', budget=16, anchors=True))
|
| assert result['audit']['true_map_retained'] is True
|
| assert result['metrics']['target_steps'] <= 16
|
| assert result['metrics']['target_calls'] == result['metrics']['resets'] + result['metrics']['target_steps']
|
| if result['decision']['admitted']:
|
| assert result['audit']['contract_satisfied'] is True
|
|
|
|
|
| def test_related_methods_share_probe_transcript_when_only_admission_changes():
|
| robust = run(RunConfig(family='latch', budget=8, admission='robust'))
|
| one = run(RunConfig(family='latch', budget=8, admission='one_best'))
|
| assert robust['probes'] == one['probes']
|
|
|
|
|
| def test_semantic_corruption_uses_actual_private_events():
|
| oracle = run(RunConfig(family='semantic_corruption', admission='oracle', budget=0))
|
| assert oracle['baseline']['privileged'] is True
|
| assert oracle['audit']['contract_satisfied'] is True
|
| assert oracle['audit']['raw_equivalent_maps_count'] > 0
|
|
|
|
|
| def test_summary_keeps_abstention_risk_undefined_and_clusters_related_cases():
|
| row = run(RunConfig(family='symmetric', budget=0))
|
| summary = summarize([row, row])[0]
|
| assert summary['requests'] == 2
|
| assert summary['parent_clusters'] == 1
|
| assert summary['selective_harm'] is None
|
| assert summary['useful_coverage'] == 0
|
|
|
|
|
| def test_study_enforces_size_limit_before_work():
|
| with pytest.raises(ValueError):
|
| run_study({'seeds': list(range(10001))})
|
|
|
|
|
| def test_calibration_budget_counts_resets_as_well_as_actions():
|
| result = run(RunConfig(family='symmetric', policy='fixed', budget=8))
|
| assert result['metrics']['target_calls'] <= 8
|
|
|
|
|
| def test_failed_deployment_never_audits_an_unexecuted_suffix():
|
| result = run(RunConfig(family='malformed', seed=1, budget=0, admission='identity'))
|
| assert result['deployment']['complete'] is False
|
| assert result['audit']['semantic_events'] == []
|
| assert result['audit']['hazard'] is None
|
| summary = summarize([result])[0]
|
| assert summary['interface_errors'] == 1
|
| assert summary['hazardous'] == 0
|
|
|
|
|
| def test_forced_fit_exposes_residual_instead_of_implying_consistency():
|
| result = run(RunConfig(family='incompatible', budget=16, policy='fixed', admission='fit'))
|
| fit = result['baseline']['fit_diagnostics']
|
| assert fit['chosen_adapter_id'] is not None
|
| assert fit['raw_mismatches'] > 0
|
|
|
|
|
| def test_paired_comparison_never_matches_different_budget_conditions():
|
| from crosswire.study import paired_comparisons
|
| def row(policy, budget, success):
|
| return {'configuration': {'family':'latch','policy':policy,'admission':'robust',
|
| 'seed':17,'budget':budget,'anchors':False,'horizon':4,
|
| 'split':'evaluation','implementation':'procedural'},
|
| 'parent_instance_id':'p1','audit':{'contract_satisfied':success}}
|
| assert paired_comparisons([row('task',4,True),row('random',8,False)]) == []
|
|
|
|
|
| def test_paired_comparison_evaluates_admission_contrasts():
|
| from crosswire.study import paired_comparisons
|
| def row(admission, success):
|
| return {'configuration': {'family':'latch','policy':'task','admission':admission,
|
| 'seed':17,'budget':4,'anchors':False,'horizon':4,
|
| 'execution_mode':'shielded','temporal':False,
|
| 'split':'evaluation','implementation':'procedural'},
|
| 'parent_instance_id':'p1','audit':{'contract_satisfied':success}}
|
| pairs = paired_comparisons([row('robust', False), row('adaptive', True)])
|
| assert len(pairs) == 1
|
| assert pairs[0]['contrast'] == 'adaptive - robust'
|
| assert pairs[0]['mean_delta'] == 1.0
|
|
|
|
|
| def test_unsupported_aalpy_combinations_rejected():
|
| from crosswire.study import RunConfig
|
| with pytest.raises(ValueError, match='AALpy baseline does not support temporal'):
|
| RunConfig(policy='aalpy', temporal=True).check()
|
| with pytest.raises(ValueError, match='AALpy baseline does not support adaptive'):
|
| RunConfig(policy='aalpy', admission='adaptive', execution_mode='shielded').check()
|
|
|
|
|
| def _analysis_row(**changes):
|
| from dataclasses import asdict
|
| configuration = asdict(RunConfig())
|
| configuration.update(changes)
|
| return {'configuration': configuration, 'parent_instance_id': 'parent-latch',
|
| 'decision': {'admitted': True, 'status': 'TASK_ADMISSIBLE'},
|
| 'audit': {'contract_satisfied': True, 'oracle_feasible': True,
|
| 'hazard': False, 'calibration_hazardous_probes': 0},
|
| 'metrics': {'target_calls': 0, 'target_steps': 0, 'resets': 0},
|
| 'deployment': {'target_calls': 0}}
|
|
|
|
|
| @pytest.mark.parametrize('field,other', [('anchors', True), ('split', 'evaluation'),
|
| ('horizon', 5), ('execution_mode', 'shielded'),
|
| ('temporal', True), ('escalate', True)])
|
| def test_summary_separates_experimental_conditions(field, other):
|
| rows = [_analysis_row(), _analysis_row(**{field: other})]
|
| summary = summarize(rows)
|
| assert len(summary) == 2
|
| assert all(item['requests'] == 1 for item in summary)
|
| assert {item[field] for item in summary} == {rows[0]['configuration'][field], other}
|
|
|
|
|
| def test_paired_comparison_never_matches_different_escalation_conditions():
|
| from crosswire.study import paired_comparisons
|
| rows = [_analysis_row(policy='task', escalate=True),
|
| _analysis_row(policy='random', escalate=False)]
|
| assert paired_comparisons(rows) == []
|
|
|
|
|
| def test_paired_comparisons_report_separate_matching_condition_strata():
|
| from crosswire.study import paired_comparisons
|
| rows = [_analysis_row(policy=policy, escalate=escalate)
|
| for policy in ('task', 'random') for escalate in (False, True)]
|
| pairs = paired_comparisons(rows)
|
| assert len(pairs) == 2
|
| assert {pair['escalate'] for pair in pairs} == {False, True}
|
| assert all(pair['matched_cells'] == 1 for pair in pairs)
|
|
|
|
|
| def test_paired_comparisons_reject_duplicate_cells_before_silent_overwrite():
|
| from crosswire.study import paired_comparisons
|
| first, duplicate = _analysis_row(), _analysis_row()
|
| duplicate['audit']['contract_satisfied'] = False
|
| with pytest.raises(ValueError, match='Duplicate'):
|
| paired_comparisons([first, duplicate, _analysis_row(policy='random')])
|
|
|
|
|
| def test_model_escalation_preserves_discarded_class_computation_in_metrics():
|
| from crosswire.core import CandidateSpace
|
| from crosswire.fixtures import make_case, make_engine
|
| from crosswire.study import _acquire, fit_evidence
|
| case = make_case('delay', seed=17)
|
|
|
| def acquire(budget, escalate):
|
| engine, counts = make_engine(case.private), {'target_calls': 0}
|
| def query(word):
|
| engine.reset()
|
| counts['target_calls'] += 1 + len(word)
|
| return [engine.step(action) for action in word]
|
| result = _acquire(case.public, 'task', budget, query, counts, None, None,
|
| escalate=escalate)
|
| engine.close()
|
| return result, counts['target_calls']
|
|
|
| (_, survivors, _, _, _), used = acquire(16, False)
|
| assert not survivors
|
| (static, _, probes, _, _), _ = acquire(used, False)
|
| (escalated, _, replayed, _, info), _ = acquire(used, True)
|
| assert replayed == [dict(probe, survivors_after=replayed[i]['survivors_after'])
|
| for i, probe in enumerate(probes)]
|
| temporal = CandidateSpace(static.machine, static.contract, temporal=True)
|
| temporal.survivors(fit_evidence(probes), {})
|
| assert info['escalated'] is True
|
| assert escalated.work == {key: value + temporal.work[key]
|
| for key, value in static.work.items()}
|
|
|
|
|
