Digital Drosophila Archive
169 lines
9.4 KiB
Python
169 lines
9.4 KiB
Python
"""Temporary metadata fixtures only; never starts a model or reads its arrays."""
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('_progress_subject', Path(__file__).with_name('progress_normal.py'))
|
|
P = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(P)
|
|
|
|
|
|
def write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
raw = (json.dumps(value, sort_keys=True) + '\n').encode()
|
|
path.write_bytes(raw)
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
class Fixture:
|
|
def __init__(self, root, register=True):
|
|
self.exp, self.control = root / 'EXP', root / 'CONTROL'
|
|
self.exp.mkdir(); self.control.mkdir()
|
|
self.inventory = dict(train_calls=1, probe_calls=1, brain_calls=2, original_E_calls=1, training_cpu_validation_calls=1)
|
|
self.count = 0
|
|
if register:
|
|
self.sha = write(self.exp / 'protocol/preregistered.json', dict(schema='normal-conditioning-registration-v1',
|
|
scientific_scope='normal-acquisition-only', subsequent_assays_authorized=False, normal_inventory=self.inventory))
|
|
write(self.exp / '.append-ledger/header.json', dict(schema='normal-append-ledger-v1', root=str(self.exp),
|
|
registration_sha256=self.sha, initial_inventory={'protocol/preregistered.json': {'sha256': self.sha}}))
|
|
|
|
def start(self, entry_id):
|
|
job = self.exp / 'jobs' / (entry_id + '.json')
|
|
jobsha = write(job, dict(kind='sample' if entry_id == 'input-generation' else 'fixture'))
|
|
value = dict(schema='normal-append-ledger-v1-start', entry_id=entry_id, registration_sha256=self.sha,
|
|
next_sequence=self.count + 1, prev_entry_sha256=None, terminal_state='unknown', retry_authorized=False,
|
|
job=dict(path=str(job), sha256=jobsha))
|
|
write(self.exp / '.append-ledger/started' / (entry_id + '.json'), value)
|
|
return jobsha
|
|
|
|
def process(self, entry_id, jobsha, code=0, complete=True, timed_out=False):
|
|
path = self.exp / 'receipts' / entry_id / 'process.json'
|
|
digest = write(path, dict(entry_id=entry_id, registration_sha256=self.sha, job_sha256=jobsha,
|
|
complete=complete, exit_code=code, timed_out=timed_out, exit_observed_by_parent=code is not None or timed_out))
|
|
return dict(path=str(path), relative_path=path.relative_to(self.exp).as_posix(), sha256=digest, bytes=path.stat().st_size)
|
|
|
|
def result(self, task):
|
|
key = 'task-' + hashlib.sha256(task['task_id'].encode()).hexdigest()
|
|
path = self.exp / 'results' / (key + '.json')
|
|
digest = write(path, dict(schema='normal-task-result-v1', registration_sha256=self.sha, complete=True, task=task))
|
|
return dict(path=str(path), relative_path=path.relative_to(self.exp).as_posix(), sha256=digest, bytes=path.stat().st_size)
|
|
|
|
def commit(self, entry_id, task=None):
|
|
jobsha = self.start(entry_id)
|
|
outputs = [self.process(entry_id, jobsha)]
|
|
if task is not None:
|
|
outputs.append(self.result(task))
|
|
self.count += 1
|
|
return write(self.exp / f'.append-ledger/entries/{self.count:016d}.json', dict(
|
|
schema='normal-append-ledger-v1-entry', entry_id=entry_id, sequence=self.count, registration_sha256=self.sha,
|
|
scientific_gate_claimed=False, new_outputs=outputs, metadata=dict(observed_returncode=0,
|
|
job_file=str(self.exp / 'jobs' / (entry_id + '.json')),
|
|
process_receipt_file=str(self.exp / 'receipts' / entry_id / 'process.json'))))
|
|
|
|
def task(self, kind):
|
|
task = dict(task_id='block/normal/paired/' + ('000' if kind == 'train' else 'checkpoint-00/probe-A'),
|
|
kind=kind, block_id='block', arm='paired')
|
|
key = 'task-' + hashlib.sha256(task['task_id'].encode()).hexdigest()
|
|
return task, key
|
|
|
|
def read(self):
|
|
return P.read_progress(self.exp, self.control)
|
|
|
|
|
|
class ProgressTests(unittest.TestCase):
|
|
def test_no_registration_not_learning_or_inferred_preparation_liveness(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp), register=False)
|
|
value = f.read()
|
|
self.assertEqual(value['status'], 'PREPARING-OR-NOT-REGISTERED')
|
|
self.assertFalse(value['registered'] or value['OS_liveness_checked'] or value['learning_pass_claimed'])
|
|
|
|
def test_registered_and_launch_pid_are_not_completion(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp))
|
|
self.assertEqual(f.read()['status'], 'REGISTERED-NOT-STARTED')
|
|
write(f.control / 'launch.json', dict(registration_sha256=f.sha, supervisor_pid=123,
|
|
launch_observed=True, final_exit_observed=False, not_evidence_of_completion=True))
|
|
value = f.read()
|
|
self.assertEqual(value['status'], 'LAUNCH-RECORDED-EXIT-UNKNOWN')
|
|
self.assertEqual(value['expected_process_entries'], 5)
|
|
self.assertEqual(value['committed_process_entries'], 0)
|
|
|
|
def test_sampler_started_vs_committed_and_latest_actual_exit(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); f.start('input-generation')
|
|
self.assertEqual(f.read()['status'], 'SAMPLER-STARTED-EXIT-UNKNOWN')
|
|
f.commit('input-generation')
|
|
value = f.read()
|
|
self.assertTrue(value['sampler_committed'])
|
|
self.assertEqual(value['latest_entry']['entry_id'], 'input-generation')
|
|
self.assertTrue(value['latest_entry']['process']['sha_checked_against_latest_entry'])
|
|
self.assertFalse(value['integrity_verified'])
|
|
|
|
def test_written_result_never_substitutes_for_verify_commit(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); f.commit('input-generation')
|
|
task, key = f.task('train')
|
|
f.commit(key + '--brain'); f.commit(key + '--update')
|
|
jobsha = f.start(key + '--verify'); f.result(task)
|
|
before = f.read()
|
|
self.assertEqual(before['training_epochs_completed'], 0)
|
|
self.assertEqual(before['result_files_present'], 1)
|
|
self.assertEqual(len(before['result_files_without_commit']), 1)
|
|
f.process(key + '--verify', jobsha)
|
|
self.assertEqual(f.read()['status'], 'CHILD-EXITED-ZERO-NOT-COMMITTED')
|
|
f.commit(key + '--verify', task)
|
|
after = f.read()
|
|
self.assertEqual(after['training_epochs_completed'], 1)
|
|
self.assertEqual(after['result_files_without_commit'], [])
|
|
|
|
def test_probe_and_train_commits_separate_and_no_neural_files_read(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); f.commit('input-generation')
|
|
train, key = f.task('train')
|
|
f.commit(key + '--brain'); f.commit(key + '--update'); f.commit(key + '--verify', train)
|
|
probe, key = f.task('probe'); f.commit(key + '--brain', probe)
|
|
state = f.exp / 'runs' / key / 'state.npz'; state.parent.mkdir(parents=True); state.write_bytes(b'not-an-array')
|
|
value = f.read()
|
|
self.assertEqual((value['training_epochs_completed'], value['probes_completed']), (1, 1))
|
|
self.assertEqual(value['task_results_ledger_owned'], 2)
|
|
self.assertEqual(value['latest_entry']['task']['kind'], 'probe')
|
|
self.assertFalse(value['scientific_metrics_evaluated'] or value['model_executed'])
|
|
|
|
def test_failed_child_and_real_supervisor_exit_and_stop_record(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); jobsha = f.start('input-generation')
|
|
f.process('input-generation', jobsha, code=7, complete=False)
|
|
self.assertEqual(f.read()['status'], 'CHILD-FAILED-UNCOMMITTED')
|
|
write(f.control / 'exit.json', dict(registration_sha256=f.sha, final_exit_observed=True, exit_code=1, complete=False))
|
|
self.assertEqual(f.read()['status'], 'PARENT-EXITED-NONZERO')
|
|
write(f.exp / 'audit/stopped.json', dict(registration_sha256=f.sha, complete=False, error='fixture'))
|
|
self.assertEqual(f.read()['status'], 'STOPPED-RECORD-PRESENT')
|
|
|
|
def test_missing_fields_bad_sequence_or_registration_are_unknown(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); f.commit('input-generation')
|
|
value = P.read_progress(f.exp, f.control, '0' * 64)
|
|
self.assertEqual(value['status'], 'UNKNOWN-METADATA-INCONSISTENT')
|
|
path = f.exp / '.append-ledger/entries/0000000000000001.json'
|
|
row, _ = P.read_json(path); del row['metadata']['observed_returncode']; write(path, row)
|
|
self.assertEqual(f.read()['status'], 'UNKNOWN-METADATA-INCONSISTENT')
|
|
path.rename(path.with_name('0000000000000002.json'))
|
|
self.assertEqual(f.read()['status'], 'UNKNOWN-METADATA-INCONSISTENT')
|
|
|
|
def test_latest_receipt_mutation_is_unknown_not_metadata_success(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
f = Fixture(Path(temp)); f.commit('input-generation')
|
|
path = f.exp / 'receipts/input-generation/process.json'
|
|
row, _ = P.read_json(path); row['exit_code'] = 9; row['complete'] = False; write(path, row)
|
|
value = f.read()
|
|
self.assertEqual(value['status'], 'UNKNOWN-METADATA-INCONSISTENT')
|
|
self.assertFalse(value['completion_counts_available'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|