"""Pure inventory/dependency tests; no I/O, GPU, scientific input or runtime.""" import copy import json import unittest import engineering_plan as p class FixedPlan(unittest.TestCase): def test_complete_current_inventory_not_future_supplements(self): plan = p.build_plan() self.assertEqual(plan['total_calls'], len(plan['rows'])) self.assertEqual(plan['total_calls'], 32) self.assertEqual(plan['total_ticks'], sum(row['steps'] for row in plan['rows'])) self.assertEqual(plan['total_ticks'], 11074) self.assertEqual(plan['overlays'], ['zero', 'static25']) self.assertEqual(plan['chain_lengths'], [5, 13, 1, 481, 500]) self.assertFalse(plan['training']) self.assertFalse(plan['candidate_generated']) self.assertEqual(p.validate_plan(json.loads(json.dumps(plan))), plan) for overlay in plan['overlays']: rows = [r for r in plan['rows'] if r['overlay'] == overlay] self.assertEqual(len(rows), 16) self.assertEqual(sum(r['steps'] for r in rows), 5537) self.assertEqual([(r['mode'], r['steps']) for r in rows if r['kind'] == 'fresh'], [('reference', 1000), ('light', 1000), ('light', 1000)]) self.assertEqual([r['stop'] for r in rows if r['kind'] == 'prefix'], [18, 19, 500]) self.assertFalse(any(r['stop'] > 1000 for r in rows)) def test_all_prerequisites_precede_rows_and_fresh_prefixes_precede_chains(self): rows = p.build_plan()['rows'] self.assertTrue(all(r['kind'] in ('fresh', 'prefix') for r in rows[:12])) self.assertTrue(all(r['kind'] == 'chain' for r in rows[12:])) positions = {r['row_id']: i for i, r in enumerate(rows)} self.assertEqual(len(positions), 32) for index, row in enumerate(rows): if row['parent_row_id'] is not None: self.assertLess(positions[row['parent_row_id']], index) for check in row['expected_comparisons']: reference = check['reference_row_id'] if reference is not None: self.assertLess(positions[reference], index) for candidate in check['candidate_row_ids']: self.assertLessEqual(positions[candidate], index) self.assertLess(check['start'], check['stop']) def test_each_chain_is_exact_partition_with_local_phase_distinct_from_global(self): for overlay in p.OVERLAYS: for mode in p.MODES: chain = [r for r in p.build_plan()['rows'] if (r['overlay'], r['mode'], r['kind']) == (overlay, mode, 'chain')] self.assertEqual([(r['start'], r['stop']) for r in chain], [(0, 5), (5, 18), (18, 19), (19, 500), (500, 1000)]) self.assertEqual([r['steps'] % 19 for r in chain], [5, 13, 1, 6, 6]) self.assertEqual([r['stop'] % 19 for r in chain], [5, 18, 0, 6, 12]) self.assertIsNone(chain[0]['parent_row_id']) for earlier, later in zip(chain, chain[1:]): self.assertEqual(later['parent_row_id'], earlier['row_id']) self.assertEqual(later['start'], earlier['stop']) self.assertEqual(chain[2]['start'], 18) # source tick 0 arrives at this child's local 0. self.assertEqual(chain[2]['steps'], 1) last = next(c for c in chain[-1]['expected_comparisons'] if c['check'] == 'chain_exact') self.assertEqual(last['candidate_row_ids'], [r['row_id'] for r in chain]) self.assertEqual((last['start'], last['stop']), (0, 1000)) def test_prefix5_is_reference_first_segment_without_extra_run(self): rows = p.build_plan()['rows'] self.assertFalse(any(r['kind'] == 'prefix' and r['stop'] == 5 for r in rows)) for overlay in p.OVERLAYS: first = {r['mode']: r for r in rows if r['kind'] == 'chain' and r['overlay'] == overlay and r['start'] == 0} refs = [c for c in first['light']['expected_comparisons'] if c['check'] == 'checkpoint_exact'] self.assertEqual(len(refs), 1) self.assertEqual(refs[0]['reference_row_id'], first['reference']['row_id']) self.assertFalse(any(c['reference_row_id'] == first['reference']['row_id'] for c in first['reference']['expected_comparisons'])) def test_fresh_am_and_light_repeat_comparisons_are_explicit(self): rows = p.build_plan()['rows'] for overlay in p.OVERLAYS: fresh = [r for r in rows if r['overlay'] == overlay and r['kind'] == 'fresh'] self.assertEqual(len(fresh), 3) for row in fresh: am = [c for c in row['expected_comparisons'] if c['check'] == 'am_exact'] self.assertEqual(len(am), 1) self.assertEqual(am[0]['external_reference'], 'sealed_AM_selected_case_same_overlay_r0') self.assertIsNone(am[0]['reference_row_id']) self.assertEqual(fresh[1]['expected_comparisons'][1]['reference_row_id'], fresh[0]['row_id']) self.assertEqual(fresh[2]['expected_comparisons'][1]['reference_row_id'], fresh[1]['row_id']) def test_every_result_is_independent_not_shared_mutable_inventory(self): original = p.build_plan() checked = p.validate_plan(original) checked['rows'][0]['expected_comparisons'][0]['candidate_row_ids'].append('mutation') checked['chain_lengths'][0] = 99 self.assertEqual(original, p.build_plan()) original['rows'][0]['expected_comparisons'][0]['scope'] = 'mutation' self.assertNotEqual(original, p.build_plan()) def test_exact_types_reject_bool_int_float_and_sequence_conflation(self): mutations = ( lambda x: x.update(total_calls=32.0), lambda x: x.update(training=0), lambda x: x['rows'][0].update(start=False), lambda x: x['rows'][14].update(steps=True), lambda x: x['rows'][14].update(steps=1.0), lambda x: x['rows'][0]['expected_comparisons'][0].update(start=False), lambda x: x.update(rows=tuple(x['rows'])), ) for mutate in mutations: plan = p.build_plan(); mutate(plan) with self.subTest(mutation=repr(mutate)), self.assertRaisesRegex(ValueError, 'type'): p.validate_plan(plan) def test_unknown_fields_missing_overlays_and_reordering_are_rejected(self): mutations = ( lambda x: x.update(extra=True), lambda x: x['rows'][0].update(extra='not allowed'), lambda x: x['rows'][0]['expected_comparisons'][0].update(tolerance=0), lambda x: x.update(overlays=['zero']), lambda x: x['rows'].pop(0), lambda x: x['rows'].append(copy.deepcopy(x['rows'][-1])), lambda x: x['rows'].__setitem__(slice(0, 2), x['rows'][1::-1]), lambda x: x['rows'][0].update(overlay='static25'), lambda x: x['rows'][0].update(mode='light'), lambda x: x['rows'][0].update(row_id='arbitrary'), ) for mutate in mutations: plan = p.build_plan(); mutate(plan) with self.subTest(mutation=repr(mutate)), self.assertRaises(ValueError): p.validate_plan(plan) def test_gaps_wrong_parents_and_dropped_comparisons_never_pass(self): mutations = ( lambda x: x['rows'][13].update(start=6, steps=12), lambda x: x['rows'][13].update(stop=17, steps=12), lambda x: x['rows'][13].update(parent_row_id=None), lambda x: x['rows'][12].update(parent_row_id=x['rows'][13]['row_id']), lambda x: x['rows'][16].update(stop=999, steps=499), lambda x: x['rows'][0].update(expected_comparisons=[]), lambda x: x['rows'][2]['expected_comparisons'].pop(), lambda x: x['rows'][16]['expected_comparisons'][-1]['candidate_row_ids'].pop(), lambda x: x['rows'][16]['expected_comparisons'][-1].update(scope='final_only'), ) for mutate in mutations: plan = p.build_plan(); mutate(plan) with self.subTest(mutation=repr(mutate)), self.assertRaises(ValueError): p.validate_plan(plan) if __name__ == '__main__': unittest.main()