Digital Drosophila Archive
364 lines
22 KiB
Python
364 lines
22 KiB
Python
"""Pure temporary-file synthetic tests; never a brain run or old-file write.
|
|
|
|
Public integration tests mock only the full-case loader to use a tiny graph.
|
|
The sealed output parser, timing/queue reconstructor, overlay bridge, source AST
|
|
and receipt checks remain real. Production rejects these tiny cases unchanged.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import comparison as c
|
|
|
|
|
|
def write_json(path, data):
|
|
Path(path).write_text(json.dumps(data, sort_keys=True, allow_nan=False), encoding='utf-8')
|
|
|
|
|
|
class ComparisonTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='phase4am-comparison-synthetic-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name).resolve()
|
|
self.case_dir = self.root / 'case'
|
|
self.graph_dir = self.root / 'graph'
|
|
self.bundle_dir = self.root / 'bundle'
|
|
for p in (self.case_dir, self.graph_dir, self.bundle_dir):
|
|
p.mkdir()
|
|
self.helper = c._helper(set())
|
|
self.steps = 30
|
|
self.graph = dict(nodes=np.arange(4, dtype=np.int32), edge_id=np.arange(6, dtype=np.int64),
|
|
ptr=np.array([0, 2, 3, 5, 6], np.int64), src=np.array([0, 0, 1, 2, 2, 3], np.int32),
|
|
dst=np.array([3, 2, 2, 0, 3, 1], np.int32), signed_raw=np.array([1, 0, 2, -2, 3, 1], np.int64))
|
|
self.graph['weights'] = self.graph['signed_raw'] * (.275 * .001)
|
|
for key, value in self.graph.items():
|
|
np.save(self.graph_dir / (key + '.npy'), value)
|
|
self.bundle = {'kc_globals.npy': np.r_[np.array([0, 1], np.int32), np.arange(4, 4066, dtype=np.int32)],
|
|
'apl_globals.npy': np.array([2, 3], np.int32), 'input_matrix.npy': np.zeros((2, c.N), np.float32),
|
|
'output_matrix.npy': np.zeros((2, c.K), np.float32)}
|
|
for key, value in self.bundle.items():
|
|
np.save(self.bundle_dir / key, value)
|
|
self.requests = [(0, 0), (1, 1), (2, 0), (22, 0), (23, 3)]
|
|
np.savez(self.case_dir / 'requests.npz', tick=np.array([x[0] for x in self.requests], np.int32),
|
|
neuron_idx=np.array([x[1] for x in self.requests], np.int32))
|
|
write_json(self.case_dir / 'spec.json', {'name': 'synthetic', 'steps': self.steps,
|
|
'dt_ms': .1, 'duration_ms': 3., 'record_history': False, 'apl_delay_steps': 18})
|
|
write_json(self.root / 'coeff.json', {'synthetic': True})
|
|
self.case = dict(spec=json.loads((self.case_dir / 'spec.json').read_text()), graph=self.graph,
|
|
bundle=self.bundle, token=('synthetic', str(self.root)), graph_dir=self.graph_dir, bundle_dir=self.bundle_dir,
|
|
requested=self.requests, input_hashes={'spec.json': c.file_hash(self.case_dir / 'spec.json'),
|
|
'requests.npz': c.file_hash(self.case_dir / 'requests.npz'), 'base_weights.npy': c.file_hash(self.graph_dir / 'weights.npy'),
|
|
'coefficients.json': c.file_hash(self.root / 'coeff.json')})
|
|
self.ids = np.array([0, 2], np.int64)
|
|
self.base = np.array([.275, .55], np.float32)
|
|
np.save(self.root / 'mask.npy', self.ids)
|
|
np.save(self.root / 'base.npy', self.base)
|
|
self.delta = np.zeros(2, np.float32)
|
|
self.overlay_name = 'zero'
|
|
self._overlay_inputs()
|
|
|
|
def _overlay_inputs(self):
|
|
np.savez(self.root / 'overlay.npz', edge_id=self.ids, plastic_delta_mv=self.delta)
|
|
write_json(self.root / 'overlay-manifest.json', dict(schema='phase4am-overlay-v1', overlay_name=self.overlay_name,
|
|
overlay_sha256=c.file_hash(self.root / 'overlay.npz'), mask_path=str(self.root / 'mask.npy'),
|
|
mask_sha256=c.file_hash(self.root / 'mask.npy'), base_candidate_path=str(self.root / 'base.npy'),
|
|
base_candidate_sha256=c.file_hash(self.root / 'base.npy'),
|
|
aj_model_dir=str(Path(c.__file__).resolve().parent.parent / 'apl-timing-lab-v1')))
|
|
|
|
def run_files(self, name, wrapped):
|
|
directory = self.root / name
|
|
directory.mkdir()
|
|
ticks = np.array([0, 1, 18, 22, 23], np.int32)
|
|
nodes = np.array([0, 1, 2, 0, 3], np.int32)
|
|
state = dict(step=ticks, neuron_idx=nodes, counts=np.bincount(nodes, minlength=4).astype(np.int32),
|
|
v=np.full(4, -.052, np.float64), g=np.zeros(4, np.float64))
|
|
hashes, queues, requests = [], [], []
|
|
last = np.full(4, -1000000000, np.int64)
|
|
for tick in range(self.steps):
|
|
fired = nodes[ticks == tick]
|
|
for request_tick, target in self.requests:
|
|
if request_tick == tick:
|
|
ok = tick - last[target] >= 22
|
|
requests.append(dict(tick=tick, neuron_idx=target, attempted=1, accepted=int(ok), rejected=int(not ok)))
|
|
last[fired] = tick
|
|
ref = np.maximum(22 - (tick - last), 0).astype(np.int32)
|
|
hashes.append(dict(tick=tick, v_sha256=c.raw_hash(state['v']), g_sha256=c.raw_hash(state['g']), ref_sha256=c.raw_hash(ref)))
|
|
sources = nodes[ticks == tick - 18] if tick >= 18 else []
|
|
rows = np.array([eid for source in sources for eid in range(self.graph['ptr'][source], self.graph['ptr'][source + 1])], np.int64)
|
|
accepted = (ref[self.graph['dst'][rows]] == 0).astype(np.uint8)
|
|
queues.append(self.helper._queue_digest(tick, rows, accepted))
|
|
state['ref'] = ref
|
|
np.savez(directory / 'state.npz', **state)
|
|
for filename, rows in (('tick-state-hashes.jsonl', hashes), ('queue.jsonl', queues)):
|
|
(directory / filename).write_text(''.join(json.dumps(row) + '\n' for row in rows), encoding='utf-8')
|
|
(directory / 'requests.csv').write_text('tick,neuron_idx,attempted,accepted,rejected\n' +
|
|
''.join(','.join(str(row[key]) for key in c._helper(set()).REQUEST_FIELDS) + '\n' for row in requests), encoding='utf-8')
|
|
for field, width in (('drive', 2), ('state', 2), ('inhibition', c.K), ('applied', c.K)):
|
|
np.save(directory / ('apl-' + field + '.float32.npy'), np.zeros((self.steps, width), np.float32))
|
|
pending = np.zeros((18, c.K), np.float32)
|
|
np.save(directory / 'apl-pending-tail.float32.npy', pending)
|
|
timing = dict(apl_delay_steps=18, steps=self.steps, pending_generated_count=18, generated_rows=self.steps,
|
|
applied_generated_rows=self.steps - 18, pending_shape=[18, c.K], pending_initial_zero_prefix=0,
|
|
pending_application_tick_start=self.steps, pending_application_tick_stop_exclusive=self.steps + 18,
|
|
pending_generated_tick_start=self.steps - 18, pending_generated_tick_stop_exclusive=self.steps,
|
|
pending_raw_sha256=c.raw_hash(pending), final_apl_state_float32=[0., 0.], final_apl_state_u32_bits=[0, 0],
|
|
final_apl_state_nonzero=False, brain_tail_advanced=False, future_apl_decay_or_feedback_generated=False,
|
|
finite_generated_sequence_only=True, closed_loop_branch_integrals_claimed_equal=False,
|
|
biological_delay_measurement_claimed=False)
|
|
write_json(directory / 'apl-timing-audit.json', timing)
|
|
component = dict(ok=True, protocol='phase4ag-apl-v1', n_neurons=c.N, n_apl=2, n_kc=c.K, dt_ms=.1,
|
|
files_sha256={key: c.file_hash(self.bundle_dir / key) for key in self.bundle},
|
|
array_sha256={key: c.raw_hash(value) for key, value in self.bundle.items()},
|
|
decay_f32_bits=f'{int(np.float32(np.exp(-.1 / 20.)).view(np.uint32)):08x}',
|
|
gain_f32_bits=f'{int(np.float32(.01).view(np.uint32)):08x}',
|
|
environment=dict(device_name='NVIDIA GeForce RTX 4090', cuda_matmul_allow_tf32=False,
|
|
float32_matmul_precision='highest', precision_configuration_changed_by_component=False))
|
|
write_json(directory / 'apl-component.json', component)
|
|
(directory / 'sealed-gpu-source.diff').write_text('synthetic diff, not executed\n', encoding='utf-8')
|
|
provenance = dict(sealed_source_sha256=c.SEALED, candidate_sha256=c.AJ_SHA,
|
|
source_diff_sha256=c.file_hash(directory / 'sealed-gpu-source.diff'), module_globals_patched=False)
|
|
write_json(directory / 'source-audit.json', provenance)
|
|
_, effective_sha, apl_count = c._effective_hash(self.case, self.ids if wrapped else None, self.delta if wrapped else None)
|
|
summary = dict(case='synthetic', complete=True, pass_all=True, steps=self.steps, completed_ticks=self.steps,
|
|
nodes=4, edges=6, precision='float64', apl_precision='float32', state_units='SI volts', total_spikes=len(ticks),
|
|
requested=len(requests), accepted_requests=sum(row['accepted'] for row in requests),
|
|
rejected_requests=sum(row['rejected'] for row in requests), arriving_edges=sum(row['count'] for row in queues),
|
|
accepted_arriving_edges=sum(row['accepted_count'] for row in queues), counts_match_sparse_events=True,
|
|
all_finite=True, record_history=False, apl_fast_edges_zeroed=apl_count, kc_targets_present=c.K,
|
|
effective_weights_raw_sha256=effective_sha, apl_component=component,
|
|
apl_schedule='end, after threshold/reset and guarded arrivals, unconditional all KC',
|
|
apl_si_conversion='applied h32 -> float64 -> multiply float64 0.001 -> separate float64 subtraction',
|
|
apl_delay_steps=18, apl_timing=timing, source_provenance=provenance, pending_generated_count=18,
|
|
pending_initial_zero_prefix=0, pending_tail_shape=[18, c.K], future_generated_decay_tail_included=False,
|
|
build=dict(sources={'author_ext.cpp': c.CPP_SHA, 'author_ext.cu': c.CUDA_SHA}, arch_list='8.9',
|
|
cuda_flags=['-O2', '--fmad=false', '--ftz=false', '--prec-div=true', '--prec-sqrt=true']),
|
|
sealed_ae_backend_sha256=c.SEALED['ae_backend'], learning=False, input_sha256=self.case['input_hashes'],
|
|
**{key: 1.0 for key in c.PERFORMANCE})
|
|
write_json(directory / 'summary.json', summary)
|
|
if wrapped:
|
|
self._write_overlay(directory, summary)
|
|
return directory
|
|
|
|
def _write_overlay(self, directory, summary):
|
|
for source, target in (('overlay.npz', 'overlay-input.npz'), ('overlay-manifest.json', 'overlay-manifest.json')):
|
|
shutil.copyfile(self.root / source, directory / target)
|
|
input_rows = {}
|
|
for label, filename in (('overlay', 'overlay.npz'), ('manifest', 'overlay-manifest.json'), ('mask', 'mask.npy'), ('base_candidate', 'base.npy')):
|
|
input_rows[label + '_file'] = str(self.root / filename)
|
|
input_rows[label + '_sha256'] = c.file_hash(self.root / filename)
|
|
audit = dict(schema='phase4am-overlay-audit-v1', complete=True, overlay_name=self.overlay_name, input=input_rows,
|
|
arrays={key: dict(dtype=str(value.dtype), shape=list(value.shape), raw_sha256=c.raw_hash(value)) for key, value in
|
|
dict(edge_id=self.ids, plastic_delta_mv=self.delta, base_candidate_mv=self.base).items()},
|
|
candidate_count=2, nonzero_delta_count=int(np.count_nonzero(self.delta)), mapped_candidate_count=2,
|
|
missing_candidate_count=0, installed_nonzero_count=int(np.count_nonzero(self.delta)), delta_dtype='float32',
|
|
delta_units='mV', state_dtype='float64', bridge=c.BRIDGE, installation_schedule=c.SCHEDULE,
|
|
weights_updated_during_run=False, learning=False, zero_delta_no_write=True, frozen_mask_exact=True,
|
|
bounds_pass=True, apl_overlap_count=0, apl_fast_edges_zeroed=summary['apl_fast_edges_zeroed'],
|
|
base_graph_files_modified=False, untouched_weights_bitwise_preserved=True,
|
|
base_weights_file_sha256=c.file_hash(self.graph_dir / 'weights.npy'), base_weights_raw_sha256=c._base_raw_hash(self.case),
|
|
effective_weights_raw_sha256=summary['effective_weights_raw_sha256'],
|
|
expected_effective_weights_raw_sha256=summary['effective_weights_raw_sha256'],
|
|
copied_overlay_file='overlay-input.npz', copied_manifest_file='overlay-manifest.json',
|
|
inherited_aj_source_provenance=summary['source_provenance'],
|
|
inherited_source_audit_scope='AJ source provenance only; AM overlay semantics are recorded separately',
|
|
static25_is_exact_fraction_of_float64_base_claimed=False, arrival_weight_sampling='arrival time; delta fixed for entire trial')
|
|
write_json(directory / 'overlay-audit.json', audit)
|
|
model = Path(c.__file__).resolve().parent
|
|
receipt = dict(schema='phase4am-adapter-provenance-v1', complete=True,
|
|
scope='constant sparse overlay installation only; no within-trial plasticity or learning',
|
|
adapter_file=str(model / 'adapter.py'), adapter_sha256=c.file_hash(model / 'adapter.py'),
|
|
overlay_module_file=str(model / 'overlay.py'), overlay_module_sha256=c.file_hash(model / 'overlay.py'),
|
|
sealed_sources={relative: dict(path=str(model.parent / relative), sha256=sha) for relative, sha in c.ADAPTER_PINS.items()},
|
|
ast=c._expected_ast(model.parent / 'apl-timing-lab-v1/gpu_timing.py', model / 'adapter.py'),
|
|
installation_calls=1, weights_updated_during_run=False, learning=False, apl_delay_steps=18,
|
|
original_module_globals_patched=False, delta_constant_for_entire_trial=True,
|
|
inherited_source_audit_describes_AM_semantics=False, overlay_audit_sha256=c.file_hash(directory / 'overlay-audit.json'),
|
|
inherited_summary_sha256=c.file_hash(directory / 'summary.json'), effective_weights_raw_sha256=summary['effective_weights_raw_sha256'])
|
|
write_json(directory / 'adapter-provenance.json', receipt)
|
|
|
|
def compare(self, left, right, mode='zero', output=None):
|
|
with patch.object(c, '_load_case', return_value=self.case), patch.object(c, 'CANDIDATES', 2):
|
|
return c.compare_runs(self.case_dir, left, right, output, mode)
|
|
|
|
def assert_pass(self, report):
|
|
self.assertTrue(report['pass_all'], json.dumps({key: report.get(key) for key in ('error', 'failed_checks', 'differences')}))
|
|
self.assertFalse(report['learning_achieved'])
|
|
self.assertFalse(report['candidate_generated'])
|
|
self.assertTrue(report['read_manifest'])
|
|
|
|
def test_zero_complete_real_parsers(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
self.assert_pass(self.compare(left, right))
|
|
|
|
def test_static25_repeat_and_single_validation(self):
|
|
self.overlay_name = 'static25'
|
|
self.delta = np.multiply(self.base, np.float32(-.25))
|
|
self._overlay_inputs()
|
|
left, right = self.run_files('left', True), self.run_files('right', True)
|
|
self.assert_pass(self.compare(left, right, 'repeat'))
|
|
self.assert_pass(self.compare(left, left, 'self'))
|
|
|
|
def test_production_rejects_toy_dimensions(self):
|
|
with self.assertRaises(ValueError):
|
|
c._load_case(self.case_dir, set())
|
|
|
|
def test_missing_adapter_receipt_fails_despite_complete_summary(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
(right / 'adapter-provenance.json').unlink()
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_corrupt_adapter_receipt_ast_and_summary_binding_fail(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
path = right / 'adapter-provenance.json'
|
|
original = json.loads(path.read_text())
|
|
for change in ('ast', 'binding', 'complete'):
|
|
data = copy.deepcopy(original)
|
|
if change == 'ast':
|
|
data['ast']['entire_tick_loop_ast_identical'] = False
|
|
elif change == 'binding':
|
|
data['inherited_summary_sha256'] = '0' * 64
|
|
else:
|
|
data['complete'] = False
|
|
write_json(path, data)
|
|
self.assertFalse(self.compare(left, right)['pass_all'], change)
|
|
|
|
def test_per_neuron_count_corruption_not_hidden_by_equal_total(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
for folder in (left, right):
|
|
with np.load(folder / 'state.npz') as f:
|
|
state = {key: f[key].copy() for key in f.files}
|
|
state['counts'][0] -= 1
|
|
state['counts'][1] += 1
|
|
np.savez(folder / 'state.npz', **state)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_both_equal_wrong_queues_fail_independent_reconstruction(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
for folder in (left, right):
|
|
path = folder / 'queue.jsonl'
|
|
rows = [json.loads(row) for row in path.read_text().splitlines()]
|
|
rows[18]['ordered_edge_sha256'] = '0' * 64
|
|
path.write_text(''.join(json.dumps(row) + '\n' for row in rows), encoding='utf-8')
|
|
report = self.compare(left, right)
|
|
self.assertFalse(report['pass_all'])
|
|
self.assertIn('left_gpu.full_event_queue_request_accounting', report['failed_checks'])
|
|
|
|
def test_both_equal_wrong_requests_fail(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
for folder in (left, right):
|
|
path = folder / 'requests.csv'
|
|
path.write_text(path.read_text().replace('2,0,1,0,1', '2,0,1,1,0'), encoding='utf-8')
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_both_equal_wrong_ref_hash_fails(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
for folder in (left, right):
|
|
path = folder / 'tick-state-hashes.jsonl'
|
|
rows = [json.loads(row) for row in path.read_text().splitlines()]
|
|
rows[1]['ref_sha256'] = '0' * 64
|
|
path.write_text(''.join(json.dumps(row) + '\n' for row in rows), encoding='utf-8')
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_state_signed_zero_detected(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
with np.load(right / 'state.npz') as f:
|
|
state = {key: f[key].copy() for key in f.files}
|
|
state['g'][0] = -0.
|
|
np.savez(right / 'state.npz', **state)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_same_nonfinite_states_fail(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
for folder in (left, right):
|
|
with np.load(folder / 'state.npz') as f:
|
|
state = {key: f[key].copy() for key in f.files}
|
|
state['v'][0] = np.inf
|
|
np.savez(folder / 'state.npz', **state)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_apl_pending_signed_zero_fails(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
path = right / 'apl-pending-tail.float32.npy'
|
|
value = np.load(path)
|
|
value[0, 0] = np.float32(-0.)
|
|
np.save(path, value)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_missing_tick_or_extra_request_fails(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
path = right / 'queue.jsonl'
|
|
path.write_text('\n'.join(path.read_text().splitlines()[:-1]) + '\n', encoding='utf-8')
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_summary_only_whitelisted_performance_may_differ(self):
|
|
a = dict(dynamics='same', build={'source': 'same'}, wall_seconds=1)
|
|
b = dict(a, wall_seconds=5)
|
|
checks, differences = {}, {}
|
|
c._summary_pair(a, b, checks, differences)
|
|
self.assertTrue(all(checks.values()))
|
|
b['build'] = {'source': 'different'}
|
|
c._summary_pair(a, b, checks, differences)
|
|
self.assertFalse(all(checks.values()))
|
|
|
|
def test_overlay_bad_units_and_fake_complete_receipt_fail(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
path = right / 'overlay-audit.json'
|
|
data = json.loads(path.read_text()); data['delta_units'] = 'V'; write_json(path, data)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_source_or_build_mutation_fails(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
path = right / 'summary.json'
|
|
data = json.loads(path.read_text()); data['build']['cuda_flags'].append('--use_fast_math'); write_json(path, data)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_no_existing_output_overwrite_and_no_input_directory_write(self):
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
output = self.root / 'report.json'
|
|
self.assert_pass(self.compare(left, right, output=output))
|
|
with self.assertRaises(ValueError):
|
|
self.compare(left, right, output=output)
|
|
with self.assertRaises(ValueError):
|
|
self.compare(left, right, output=right / 'report.json')
|
|
|
|
def test_fake_repeat_self_disallowed(self):
|
|
left = self.run_files('left', True)
|
|
self.assertFalse(self.compare(left, left, 'repeat')['pass_all'])
|
|
|
|
def test_static25_cannot_be_zero_reference(self):
|
|
self.overlay_name = 'static25'; self.delta = np.multiply(self.base, np.float32(-.25)); self._overlay_inputs()
|
|
left, right = self.run_files('left', False), self.run_files('right', True)
|
|
self.assertFalse(self.compare(left, right)['pass_all'])
|
|
|
|
def test_helper_source_pin_and_adapter_ast_contract(self):
|
|
model = Path(c.__file__).resolve().parent
|
|
self.assertEqual(c.file_hash(model.parent / 'apl-timing-lab-v1/compare_timing.py'), c.HELPER_SHA)
|
|
import adapter
|
|
_, owner = adapter.derive_ast((model.parent / 'apl-timing-lab-v1/gpu_timing.py').read_bytes())
|
|
independent = c._expected_ast(model.parent / 'apl-timing-lab-v1/gpu_timing.py', model / 'adapter.py')
|
|
for key, value in owner.items():
|
|
self.assertEqual(independent[key], value)
|
|
|
|
def test_fresh_sha_detects_same_size_write_even_with_identical_stat(self):
|
|
path = self.root / 'same-stat.bin'
|
|
path.write_bytes(b'first!')
|
|
identity = c._fingerprint(path)
|
|
with patch.object(c, '_fingerprint', return_value=identity):
|
|
before = c.file_hash(path)
|
|
path.write_bytes(b'other!')
|
|
after = c.file_hash(path)
|
|
self.assertEqual(before, hashlib.sha256(b'first!').hexdigest())
|
|
self.assertEqual(after, hashlib.sha256(b'other!').hexdigest())
|
|
self.assertNotEqual(before, after)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|