tboudreaux/MeanField
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Validate preconditioner evidence before claiming speedups

Browse Source
This commit is contained in:
Emily Boudreaux
2026-09-17 16:24:42 -04:00
parent bc3c39dcfd
commit 61a44fc853
2 changed files with 496 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

251
experiments/check_preconditioner_benchmark.py Normal file
View File

@@ -0,0 +1,251 @@
#!/usr/bin/env python3
"""Audit frozen preconditioner artifacts and compare equal-accuracy solve costs.
A completed experiment is valid even when neither inverse reaches the requested
linear tolerance. Those observations must survive the audit, without acquiring
an invented time-to-convergence speedup.
"""
import argparse
import csv
import hashlib
import json
import math
from pathlib import Path
REQUIRED_ARTIFACTS = {
'metadata.txt', 'layout.csv', 'preconditioner_summary.csv', 'preconditioner_actions.csv',
'preconditioner_blocks.csv', 'preconditioner_correction_blocks.csv', 'preconditioner_history.csv',
'preconditioner_invariants.csv', 'preconditioner_reference.csv', 'preconditioner_scope.txt',
'benchmark_physical_base.txt', 'benchmark_normalized_base.txt', 'benchmark_normalized_rhs.txt',
'benchmark_state_factors.txt', 'benchmark_residual_factors.txt',
'benchmark_invariant_direction.txt', 'benchmark_invariant_action.txt',
'context_reference_normalized_correction.txt',
} | {role + suffix for role in ('baseline', 'candidate') for suffix in (
'_normalized_correction.txt', '_physical_correction.txt',
'_normalized_linear_defect.txt', '_physical_linear_defect.txt')}
def digest(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def hash_matches(path, expected):
return path.is_file() and digest(path) == expected
def rows(path):
with path.open() as stream:
return list(csv.DictReader(stream))
def vector(path):
values = path.read_text().split()
if not values:
raise ValueError(f'Empty vector: {path}')
count = int(values[0])
result = [float(v) for v in values[1:]]
if count <= 0 or count != len(result) or not all(math.isfinite(v) for v in result):
raise ValueError(f'Invalid finite vector: {path}')
return result
def norm(values):
return math.sqrt(math.fsum(v * v for v in values))
def close(left, right):
return math.isfinite(left) and math.isfinite(right) and math.isclose(left, right, rel_tol=2e-12, abs_tol=1e-28)
def ratio(numerator, denominator):
# Exact residual zero is a valid outcome. Do not invent a finite ratio or
# serialize infinity when the comparison denominator is exactly zero.
return numerator / denominator if denominator > 0 else None
def invalidate_comparison(result):
comparison = result['comparison']
comparison['comparison_valid'] = False
comparison['outcome'] = 'invalid_artifacts'
for key in ('candidate_over_baseline_defect', 'candidate_over_baseline_measured_cost', 'equal_gate_speedup'):
comparison[key] = None
def invalid_case(error):
result = {'passed': False, 'checks': [{'check': 'readable_complete_artifacts', 'passed': False,
'error': str(error)}], 'comparison': {}}
invalidate_comparison(result)
return result
def inspect_case(folder):
try:
return inspect_case_artifacts(folder)
except (OSError, ValueError, KeyError, TypeError, OverflowError, ZeroDivisionError) as error:
return invalid_case(error)
def inspect_case_artifacts(folder):
checks = []
def check(name, result):
checks.append({'check': name, 'passed': bool(result)})
summary = rows(folder / 'preconditioner_summary.csv')
check('exactly_one_baseline_and_candidate', len(summary) == 2 and {r['role'] for r in summary} == {'baseline', 'candidate'})
by_role = {r['role']: r for r in summary}
rhs = vector(folder / 'benchmark_normalized_rhs.txt')
rhs_norm = norm(rhs)
check('finite_nonzero_rhs', math.isfinite(rhs_norm) and rhs_norm > 0)
layout = rows(folder / 'layout.csv')
residual_blocks = [r for r in layout if r['kind'] == 'residual']
covered = []
for row in residual_blocks:
covered.extend(range(int(row['offset']), int(row['offset']) + int(row['size'])))
check('residual_layout_covers_vector_once', covered == list(range(len(rhs))))
invariants = rows(folder / 'preconditioner_invariants.csv')
required = {'accepted_physical_bitwise_identical', 'accepted_normalized_bitwise_identical',
'accepted_residual_bitwise_identical', 'correction_bitwise_identical',
'trial_state_bitwise_identical', 'trial_residual_bitwise_identical',
'candidate_physical_bitwise_identical', 'state_factors_bitwise_identical',
'residual_factors_bitwise_identical', 'preparation_generation_unchanged',
'normalized_operator_current', 'jacobian_action_bitwise_identical'}
for role in ('context_reference', 'baseline', 'candidate'):
role_rows = [r for r in invariants if r['role'] == role]
metrics = {r['metric']: float(r['value']) for r in role_rows}
check(role + '_unique_invariants', len(role_rows) == len(metrics))
check(role + '_all_invariants_recorded', required <= metrics.keys())
check(role + '_all_invariants_pass', all(metrics.get(name) == 1.0 for name in required))
check(role + '_exact_jacobian_unchanged', metrics.get('jacobian_action_difference_norm') == 0.0)
action_rows = rows(folder / 'preconditioner_actions.csv')
check('action_checks_present', len(action_rows) == 2 and
{row['role'] for row in action_rows} == {'baseline', 'candidate'})
for row in action_rows:
expected = by_role[row['role']]['variant'] == 'independent'
check(row['role'] + '_action_variant_matches', row['variant'] == by_role[row['role']]['variant'])
check(row['role'] + '_equivalence_requirement_recorded',
row['context_equivalence_required'] == ('1' if expected else '0'))
if expected:
difference = float(row['context_action_relative_difference'])
check(row['role'] + '_fresh_inverse_matches_default',
math.isfinite(difference) and 0 <= difference <= 256 * 2.220446049250313e-16)
check('baseline_is_independent', by_role['baseline']['variant'] == 'independent')
block_rows = rows(folder / 'preconditioner_blocks.csv')
for role, row in by_role.items():
defect = vector(folder / (role + '_normalized_linear_defect.txt'))
solution = vector(folder / (role + '_normalized_correction.txt'))
true_norm = norm(defect)
check(role + '_compatible_vector_sizes', len(defect) == len(rhs) == len(solution))
check(role + '_rhs_norm', close(float(row['rhs_norm']), rhs_norm))
check(role + '_defect_norm', close(float(row['true_residual_norm']), true_norm))
check(role + '_relative_defect', close(float(row['true_relative']), true_norm / rhs_norm))
check(role + '_solution_norm', close(float(row['normalized_correction_norm']), norm(solution)))
target = float(row['relative_tolerance']) * rhs_norm
check(role + '_valid_positive_tolerance', math.isfinite(target) and target > 0)
check(role + '_unchanged_target', close(float(row['absolute_target']), target))
check(role + '_valid_gate_flag', row['true_gate_pass'] in ('0', '1'))
check(role + '_honest_gate', (row['true_gate_pass'] == '1') == (true_norm <= target))
check(role + '_cold_start', close(float(row['reported_initial_norm']), rhs_norm))
check(role + '_iteration_count', int(row['iterations']) == int(row['preconditioner_applications'])
and 0 < int(row['iterations']) <= int(row['maximum_iterations']))
timing_names = ['construction_seconds', 'normalization_binding_seconds', 'krylov_setup_seconds', 'krylov_seconds']
check(role + '_valid_timing', all(math.isfinite(float(row[k])) and float(row[k]) >= 0 for k in timing_names))
check(role + '_complete_setup_solve_cost', close(float(row['setup_plus_solve_seconds']), sum(float(row[k]) for k in timing_names)))
extra_timings = ['true_check_seconds', 'jacobian_seconds', 'preconditioner_seconds']
check(role + '_valid_verification_application_timing',
all(math.isfinite(float(row[k])) and float(row[k]) >= 0 for k in extra_timings))
full_cost = float(row['setup_solve_and_true_check_seconds'])
check(role + '_complete_verified_solve_cost', full_cost > 0 and
close(full_cost, float(row['setup_plus_solve_seconds']) + float(row['true_check_seconds'])))
check(role + '_one_independent_true_check', int(row['true_check_jacobian_applications']) == 1)
check(role + '_nonoverlapping_application_timers',
float(row['jacobian_seconds']) + float(row['preconditioner_seconds']) <= float(row['krylov_seconds']) + 1e-8)
role_blocks = [r for r in block_rows if r['role'] == role]
recorded = {r['block']: r for r in role_blocks}
check(role + '_unique_residual_blocks', len(role_blocks) == len(recorded))
check(role + '_all_residual_blocks_present', set(recorded) == {b['block'] for b in residual_blocks})
for block in residual_blocks:
start, count = int(block['offset']), int(block['size'])
measured = recorded[block['block']]
check(role + '_' + block['block'] + '_defect_norm',
close(float(measured['normalized_defect_norm']), norm(defect[start:start + count])))
check(role + '_' + block['block'] + '_rhs_norm',
close(float(measured['normalized_rhs_norm']), norm(rhs[start:start + count])))
baseline, candidate = by_role['baseline'], by_role['candidate']
for key in ('restart', 'maximum_iterations', 'relative_tolerance', 'rhs_norm', 'absolute_target'):
check('matched_' + key, baseline[key] == candidate[key])
both_pass = baseline['true_gate_pass'] == candidate['true_gate_pass'] == '1'
outcome = 'both_pass' if both_pass else ('candidate_only_pass' if candidate['true_gate_pass'] == '1'
else 'baseline_only_pass' if baseline['true_gate_pass'] == '1' else 'neither_passes')
comparison = {'outcome': outcome, 'recorded_gate_outcome': outcome, 'comparison_valid': True,
'candidate_over_baseline_defect': ratio(float(candidate['true_relative']), float(baseline['true_relative'])),
'candidate_over_baseline_measured_cost': ratio(float(candidate['setup_solve_and_true_check_seconds']), float(baseline['setup_solve_and_true_check_seconds'])),
'equal_gate_speedup': float(baseline['setup_solve_and_true_check_seconds']) / float(candidate['setup_solve_and_true_check_seconds']) if both_pass else None,
'baseline': baseline, 'candidate': candidate}
result = {'passed': all(c['passed'] for c in checks), 'checks': checks, 'comparison': comparison}
if not result['passed']:
invalidate_comparison(result)
return result
def inspect_campaign(campaign):
provenance = json.loads((campaign / 'provenance.json').read_text())
checks = [{'check': 'immutable_executable_hash', 'passed': hash_matches(campaign / 'plateau_diagnostic', provenance['binary_sha256'])},
{'check': 'mesh_hash', 'passed': hash_matches(Path(provenance['mesh']), provenance['mesh_sha256'])},
{'check': 'nonempty_unique_cases', 'passed': bool(provenance['cases']) and
len(provenance['cases']) == len({case['id'] for case in provenance['cases']})}]
results = {}
for case in provenance['cases']:
name = case['id']
record = json.loads((campaign / (name + '_run.json')).read_text())
checks.append({'check': name + '_completed', 'passed': bool(record.get('finished_utc')) and
record.get('exit_code') == 0 and not record.get('timed_out', False)})
checks.append({'check': name + '_required_outputs_hashed',
'passed': REQUIRED_ARTIFACTS <= record.get('output_hashes', {}).keys()})
arguments = case.get('arguments', [])
expected_inputs = set()
for index, argument in enumerate(arguments):
if argument in ('--probe-state', '--probe-direction', '--replay-candidate'):
path = Path(arguments[index + 1])
if not path.is_absolute(): path = Path(provenance['source']) / path
expected_inputs.add(str(path.resolve()))
checks.append({'check': name + '_required_inputs_hashed',
'passed': expected_inputs <= record.get('input_state_hashes', {}).keys()})
for path, expected in record.get('input_state_hashes', {}).items():
checks.append({'check': name + '_input_hash_' + path, 'passed': hash_matches(Path(path), expected)})
for path, expected in record.get('output_hashes', {}).items():
checks.append({'check': name + '_output_hash_' + path, 'passed': hash_matches(campaign / name / path, expected)})
results[name] = inspect_case(campaign / name)
if '--preconditioner-variant' in arguments:
expected_variant = arguments[arguments.index('--preconditioner-variant') + 1]
candidate = results[name]['comparison'].get('candidate', {})
checks.append({'check': name + '_requested_variant', 'passed': candidate.get('variant') == expected_variant})
passed = all(c['passed'] for c in checks) and all(r['passed'] for r in results.values())
if not all(c['passed'] for c in checks):
# A valid-looking CSV cannot substantiate a speedup when its producing
# executable or input/output provenance failed verification.
for result in results.values():
result['passed'] = False
invalidate_comparison(result)
assessment = {'passed': passed, 'source_commit': provenance['commit'], 'provenance_checks': checks, 'cases': results,
'scope': 'Frozen linear-solve comparison only. A failed baseline has no finite equal-gate speedup.'}
return assessment
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('campaign', type=Path)
parser.add_argument('--output', type=Path)
args = parser.parse_args()
campaign = args.campaign.resolve()
assessment = inspect_campaign(campaign)
path = args.output or campaign / 'preconditioner_assessment.json'
path.write_text(json.dumps(assessment, indent=2, allow_nan=False) + '\n')
for name, result in assessment['cases'].items():
c = result['comparison']
print(name, 'PASS' if result['passed'] else 'FAIL', c['outcome'],
'defect_ratio=', c['candidate_over_baseline_defect'], 'equal_gate_speedup=', c['equal_gate_speedup'])
return 0 if assessment['passed'] else 1
if __name__ == '__main__':
raise SystemExit(main())

245
experiments/test_preconditioner_benchmark_checker.py Normal file
View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""Small synthetic linear experiments testing whether a speedup is supportable.
No executable, FEM setup, or solver is launched. The fixture uses J=I and keeps
the exact solution/defect relationship while varying convergence and evidence.
"""
import csv
import json
from pathlib import Path
import tempfile
import unittest
import check_preconditioner_benchmark as checker
def write_csv(path, records):
with path.open('w', newline='') as stream:
writer = csv.DictWriter(stream, fieldnames=list(records[0]))
writer.writeheader()
writer.writerows(records)
def write_vector(path, values):
path.write_text(str(len(values)) + '\n' + '\n'.join(repr(v) for v in values) + '\n')
def write_json(path, value):
path.write_text(json.dumps(value, indent=2) + '\n')
class BenchmarkCheckerTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.campaign = Path(self.temporary.name).resolve()
self.folder = self.campaign / 'screen'
self.folder.mkdir()
self.rhs = [3.0, 4.0]
self.make_case()
def make_case(self, baseline_defect=0.1, candidate_defect=0.05):
layout = [{'kind': 'residual', 'block': 'equilibrium', 'offset': 0, 'size': 2, 'scale': 1},
{'kind': 'value', 'block': 'state', 'offset': 0, 'size': 2, 'scale': 1}]
write_csv(self.folder / 'layout.csv', layout)
summary, blocks, actions, correction_blocks, history = [], [], [], [], []
for role, variant, defect, cost in (
('baseline', 'independent', baseline_defect, 10.0),
('candidate', 'stellar-ldu', candidate_defect, 5.0)):
# The saved defect convention is Jp-rhs, hence p=rhs+defect for J=I.
solution = [self.rhs[0] + defect, self.rhs[1]]
exact_defect = [defect, 0.0]
for coordinates in ('normalized', 'physical'):
write_vector(self.folder / f'{role}_{coordinates}_correction.txt', solution)
write_vector(self.folder / f'{role}_{coordinates}_linear_defect.txt', exact_defect)
summary.append({
'role': role, 'variant': variant, 'restart': 10, 'maximum_iterations': 20,
'relative_tolerance': 0.03, 'rhs_norm': 5.0, 'absolute_target': 0.15,
'true_residual_norm': defect, 'true_relative': defect / 5.0,
'true_gate_pass': int(defect <= 0.15), 'solver_converged': int(defect <= 0.15),
'reported_initial_norm': 5.0, 'normalized_correction_norm': checker.norm(solution),
'iterations': 4, 'preconditioner_applications': 4,
'construction_seconds': 1.0, 'normalization_binding_seconds': 0.1,
'krylov_setup_seconds': 0.2, 'krylov_seconds': cost - 1.3,
'setup_plus_solve_seconds': cost, 'true_check_seconds': 0.1,
'setup_solve_and_true_check_seconds': cost + 0.1,
'jacobian_seconds': 1.0, 'preconditioner_seconds': 1.0,
'true_check_jacobian_applications': 1,
})
blocks.append({'role': role, 'variant': variant, 'block': 'equilibrium',
'normalized_rhs_norm': 5.0, 'normalized_defect_norm': defect})
actions.append({'role': role, 'variant': variant,
'context_equivalence_required': int(role == 'baseline'),
'context_action_relative_difference': 0.0})
correction_blocks.append({'role': role, 'variant': variant, 'block': 'state',
'normalized_norm': checker.norm(solution),
'physical_norm': checker.norm(solution)})
history.append({'role': role, 'variant': variant, 'iteration': 4,
'reported_norm': defect, 'final': 1})
for file, data in [('preconditioner_summary.csv', summary), ('preconditioner_blocks.csv', blocks),
('preconditioner_actions.csv', actions), ('preconditioner_history.csv', history),
('preconditioner_correction_blocks.csv', correction_blocks)]:
write_csv(self.folder / file, data)
invariants = []
for role in ('context_reference', 'baseline', 'candidate'):
for metric in ('accepted_physical_bitwise_identical', 'accepted_normalized_bitwise_identical',
'accepted_residual_bitwise_identical', 'correction_bitwise_identical',
'trial_state_bitwise_identical', 'trial_residual_bitwise_identical',
'candidate_physical_bitwise_identical', 'state_factors_bitwise_identical',
'residual_factors_bitwise_identical', 'preparation_generation_unchanged',
'normalized_operator_current', 'jacobian_action_bitwise_identical'):
invariants.append({'role': role, 'metric': metric, 'value': 1})
invariants.append({'role': role, 'metric': 'jacobian_action_difference_norm', 'value': 0})
write_csv(self.folder / 'preconditioner_invariants.csv', invariants)
write_csv(self.folder / 'preconditioner_reference.csv', [{'metric': 'context_inverse_correction_norm', 'value': 5.0}])
for name, values in {
'benchmark_normalized_rhs.txt': self.rhs,
'benchmark_normalized_base.txt': [2.0, 3.0], 'benchmark_physical_base.txt': [2.0, 3.0],
'benchmark_state_factors.txt': [1.0, 1.0], 'benchmark_residual_factors.txt': [1.0, 1.0],
'benchmark_invariant_direction.txt': [1.0, 0.0], 'benchmark_invariant_action.txt': [1.0, 0.0],
'context_reference_normalized_correction.txt': self.rhs,
}.items():
write_vector(self.folder / name, values)
(self.folder / 'metadata.txt').write_text('n=1\nJ=0\ncentral_phase=0\npreconditioner_probe=1\n')
(self.folder / 'preconditioner_scope.txt').write_text('Synthetic J=I accuracy-gate comparison.\n')
self.snapshot = self.campaign / 'snapshot.txt'
write_vector(self.snapshot, [2.0, 3.0])
self.mesh = self.campaign / 'mesh.smesh'
self.mesh.write_text('Synthetic fixture; never opened by a solver.\n')
executable = self.campaign / 'plateau_diagnostic'
executable.write_bytes(b'Synthetic executable provenance; not executable.\n')
write_json(self.campaign / 'provenance.json', {
'commit': 'synthetic-test', 'source': str(self.campaign),
'binary_sha256': checker.digest(executable), 'mesh': str(self.mesh),
'mesh_sha256': checker.digest(self.mesh), 'cases': [{
'id': 'screen', 'arguments': ['--probe-state', str(self.snapshot),
'--preconditioner-variant', 'stellar-ldu']}],
})
write_json(self.campaign / 'screen_run.json', {
'exit_code': 0, 'finished_utc': '2026-09-18T00:00:00Z',
'input_state_hashes': {str(self.snapshot): checker.digest(self.snapshot)},
'output_hashes': {p.name: checker.digest(p) for p in self.folder.iterdir()},
})
def alter_rows(self, filename, change):
path = self.folder / filename
data = checker.rows(path)
change(data)
write_csv(path, data)
def assert_no_claim(self, result):
self.assertFalse(result['passed'])
self.assertFalse(result['comparison']['comparison_valid'])
self.assertIsNone(result['comparison']['equal_gate_speedup'])
def test_matched_verified_gates_allow_measured_speedup(self):
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertEqual(comparison['outcome'], 'both_pass')
self.assertAlmostEqual(comparison['equal_gate_speedup'], 10.1 / 5.1)
def test_failed_gates_are_valid_observations_without_speedup(self):
for baseline, candidate, outcome in ((0.2, 0.3, 'neither_passes'),
(0.2, 0.1, 'candidate_only_pass'),
(0.1, 0.2, 'baseline_only_pass')):
with self.subTest(outcome=outcome):
self.make_case(baseline, candidate)
result = checker.inspect_campaign(self.campaign)
self.assertTrue(result['passed'], result)
comparison = result['cases']['screen']['comparison']
self.assertEqual(comparison['outcome'], outcome)
self.assertIsNone(comparison['equal_gate_speedup'])
def test_exact_zero_defect_does_not_invent_a_defect_ratio(self):
self.make_case(0.0, 0.0)
result = checker.inspect_case(self.folder)
self.assertTrue(result['passed'], result)
self.assertIsNone(result['comparison']['candidate_over_baseline_defect'])
self.assertIsNotNone(result['comparison']['equal_gate_speedup'])
json.dumps(result, allow_nan=False)
def test_missing_invariant_cannot_support_speedup(self):
self.alter_rows('preconditioner_invariants.csv', lambda data: data.pop(0))
self.assert_no_claim(checker.inspect_case(self.folder))
def test_duplicate_invariant_cannot_hide_a_failed_check(self):
def corrupt(data):
duplicate = dict(data[0])
data[0]['value'] = 0
data.append(duplicate)
self.alter_rows('preconditioner_invariants.csv', corrupt)
self.assert_no_claim(checker.inspect_case(self.folder))
def test_false_reported_gate_is_rejected(self):
self.make_case(0.2, 0.1)
self.alter_rows('preconditioner_summary.csv', lambda data: data[0].update(true_gate_pass=1))
self.assert_no_claim(checker.inspect_case(self.folder))
def test_faster_total_cannot_omit_verification_or_setup(self):
self.alter_rows('preconditioner_summary.csv',
lambda data: data[1].update(setup_solve_and_true_check_seconds=0.01))
self.assert_no_claim(checker.inspect_case(self.folder))
def test_saved_defect_and_csv_must_agree(self):
write_vector(self.folder / 'candidate_normalized_linear_defect.txt', [1.0, 0.0])
self.assert_no_claim(checker.inspect_case(self.folder))
def test_same_vector_norm_does_not_hide_wrong_block_distribution(self):
layout = [
{'kind': 'residual', 'block': 'gravity', 'offset': 0, 'size': 1, 'scale': 1},
{'kind': 'residual', 'block': 'material', 'offset': 1, 'size': 1, 'scale': 1},
]
write_csv(self.folder / 'layout.csv', layout)
block_rows = []
for role, defect in [('baseline', 0.1), ('candidate', 0.05)]:
block_rows.extend([{'role': role, 'variant': 'unused', 'block': 'gravity',
'normalized_rhs_norm': 3.0, 'normalized_defect_norm': defect},
{'role': role, 'variant': 'unused', 'block': 'material',
'normalized_rhs_norm': 4.0, 'normalized_defect_norm': 0.0}])
write_csv(self.folder / 'preconditioner_blocks.csv', block_rows)
write_vector(self.folder / 'candidate_normalized_linear_defect.txt', [0.0, 0.05])
self.assert_no_claim(checker.inspect_case(self.folder))
def test_default_equivalence_cannot_be_disabled_in_csv(self):
self.alter_rows('preconditioner_actions.csv',
lambda data: data[0].update(context_equivalence_required=0))
self.assert_no_claim(checker.inspect_case(self.folder))
def test_omitted_input_or_output_hash_cannot_support_speedup(self):
for key, remove in [('input_state_hashes', str(self.snapshot)),
('output_hashes', 'candidate_normalized_linear_defect.txt')]:
with self.subTest(key=key):
self.make_case()
path = self.campaign / 'screen_run.json'
record = json.loads(path.read_text())
record[key].pop(remove)
write_json(path, record)
result = checker.inspect_campaign(self.campaign)
self.assertFalse(result['passed'])
self.assert_no_claim(result['cases']['screen'])
def test_changed_provenance_input_executable_mesh_or_output_blocks_claim(self):
for name in ('input', 'executable', 'mesh', 'output'):
with self.subTest(name=name):
self.make_case()
path = {'input': self.snapshot, 'executable': self.campaign / 'plateau_diagnostic',
'mesh': self.mesh, 'output': self.folder / 'preconditioner_summary.csv'}[name]
with path.open('a') as stream:
stream.write('\n')
result = checker.inspect_campaign(self.campaign)
self.assertFalse(result['passed'])
self.assert_no_claim(result['cases']['screen'])
def test_in_progress_record_is_not_a_completed_comparison(self):
path = self.campaign / 'screen_run.json'
record = json.loads(path.read_text())
record.pop('finished_utc')
write_json(path, record)
result = checker.inspect_campaign(self.campaign)
self.assertFalse(result['passed'])
self.assert_no_claim(result['cases']['screen'])
if __name__ == '__main__':
unittest.main()