import json, os, sys, unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from latency import records, export_labels, release_pin, join_ok, summarize, window_buckets ROW = lambda g, since=10: {'since': since, 'grades': [{'v': 2}] * g, 'panel': []} class YorushikaFixtures(unittest.TestCase): def emit(self, name, r): print('\nFIXTURE', name, json.dumps(r, sort_keys=True)) def test_1_missing_capture_then_expired(self): # 14102: zero at A, next capture missing, terminal expired with empty jury caps = [(11, {'x': ROW(0)}), (12, None)] r = records(caps, {'x': {'close': 13, 'outcome': 'expired', 'jury': []}})['x']; self.emit('1', r) self.assertIsNone(r['first_graded_capture']); self.assertEqual(r['last_zero_capture'], 11) self.assertEqual(r['grade_observed_through'], 11); self.assertEqual(r['terminal_outcome'], 'expired') self.assertEqual(r['final_window_coverage'], 'unknown'); self.assertEqual(r['coverage_reason'], 'missing_capture') def test_2_dropped_zero_capture_widens_interval(self): # 14102: graded row; dropping an intermediate zero-grade capture widens the interval, never sharpens it full = records([(11, {'x': ROW(0)}), (12, {'x': ROW(0)}), (13, {'x': ROW(1)})], {})['x'] drop = records([(11, {'x': ROW(0)}), (12, None), (13, {'x': ROW(1)})], {})['x'] self.emit('2-full', full); self.emit('2-dropped', drop) self.assertEqual(full['first_grade_interval'], {'after': 12, 'by': 13}) self.assertEqual(drop['first_grade_interval'], {'after': 11, 'by': 13}) w = lambda r: r['first_grade_interval']['by'] - r['first_grade_interval']['after'] self.assertGreater(w(drop), w(full)) def test_3_row_absent_from_successful_capture(self): # 14137/14218: A has the row at zero, B fetched but omits it, C records expiry with empty jury caps = [(11, {'x': ROW(0)}), (12, {'y': ROW(0)})] r = records(caps, {'x': {'close': 13, 'outcome': 'expired', 'jury': []}})['x']; self.emit('3', r) self.assertEqual(r['last_zero_capture'], 11); self.assertEqual(r['grade_observed_through'], 11) self.assertIsNone(r['first_graded_capture']); self.assertEqual(r['terminal_close'], 13) self.assertEqual(r['final_window_coverage'], 'unknown'); self.assertEqual(r['coverage_reason'], 'row_absent_from_capture') def test_terminal_never_advances_observation(self): caps = [(11, {'x': ROW(0)}), (12, {'x': ROW(0)})] r = records(caps, {'x': {'close': 13, 'outcome': 'expired', 'jury': []}})['x'] self.assertEqual(r['grade_observed_through'], 12); self.assertEqual(r['coverage_reason'], 'terminal_strips_grades') def test_shared_prefix_ids_survive_and_round_trip(self): # 14345: two different full ids share the first 16 hex but belong to different intake closes a = 'abcdef0123456789' + '1' * 48; b = 'abcdef0123456789' + '2' * 48 caps = [(11, {a: ROW(0, 10)}), (12, {a: ROW(0, 10), b: ROW(0, 11)})] recs = records(caps, {}) self.assertEqual((recs[a]['intake_close'], recs[b]['intake_close']), (10, 11)) ids = sorted([a, b, 'ffff' * 16]); labels = export_labels(ids) self.assertEqual(len(set(labels)), 3) # never merged self.assertEqual(labels, [a, b, 'ffffffffffffffff']) # colliding rows fall back to full ids back = {lab: full for lab, full in zip(labels, ids)} # round-trip via latency_ids.json order self.assertEqual(back[labels[0]], a); self.assertEqual(back[labels[1]], b) print('\nEMIT shared-prefix labels', labels[0][:20] + '...', labels[1][:20] + '...', labels[2]) def test_equal_length_files_from_different_exports_refuse_join(self): # 14397: two releases with the same row count but different order must not be joined by index import hashlib ids1 = ['a' * 64, 'b' * 64]; ids2 = ['b' * 64, 'a' * 64] rows1 = [[l, 1] for l in export_labels(ids1)]; rows2 = [[l, 1] for l in export_labels(ids2)] d1, b1 = release_pin(rows1, ids1, 40); d2, b2 = release_pin(rows2, ids2, 41) meta1 = {'full_id_file_sha256': hashlib.sha256(b1).hexdigest(), 'records_sha256': d1['records_sha256']} meta2 = {'full_id_file_sha256': hashlib.sha256(b2).hexdigest(), 'records_sha256': d2['records_sha256']} self.assertTrue(join_ok(rows1, meta1, b1)); self.assertTrue(join_ok(rows2, meta2, b2)) self.assertFalse(join_ok(rows1, meta1, b2)) # old records + new ids: refused self.assertFalse(join_ok(rows2, meta2, b1)) # new records + old ids: refused print('\nEMIT cross-release join refused', meta1['full_id_file_sha256'][:12], meta2['full_id_file_sha256'][:12]) class CoverageByState(unittest.TestCase): def test_coverage_split_by_state_separates_pending_from_expired(self): # 14462: a still-pending row is trivially "known"; the summary must not let that pass as better observation caps = [(11, {'p': ROW(0, 11), 'e': ROW(0, 11)}), (12, {'p': ROW(0, 11), 'e': ROW(0, 11)})] rec = records(caps, {'e': {'close': 13, 'outcome': 'expired', 'jury': []}}, latest=13) rec['p'] = records([(11, {'p': ROW(0, 11)}), (12, {'p': ROW(0, 11)}), (13, {'p': ROW(0, 11)})], {}, latest=13)['p'] c = summarize(rec, 11)[11] self.assertEqual(c['coverage_by_state']['pending']['unknown'], 0) self.assertEqual(c['coverage_by_state']['expired']['known'], 0) self.assertEqual(sum(v['known'] + v['unknown'] for v in c['coverage_by_state'].values()), c['rows']) class CalendarVsAge(unittest.TestCase): """14532/14575 on 2478: three rows born at 23, 24, 25 with the same first-grade interval (37,38]; then one widened to (35,38].""" def rows(self, widen=False): caps = [] for c in range(23, 39): m = {} for a, since in (('r23', 23), ('r24', 24), ('r25', 25)): if c < since: continue zero_until = 35 if (widen and a == 'r24') else 37 if widen and a == 'r24' and c in (36, 37): continue # row absent from these captures: interval widens m[a] = ROW(0 if c <= zero_until else 1, since) caps.append((c, m)) return records(caps, {}) def test_calendar_vs_age_histograms(self): recs = self.rows(); got = window_buckets(recs, 37, 38) exp = {'window': [37, 38], 'by_hi': 3, 'confined': 3, 'overlapping_not_confined': 0, 'calendar_hist': {'38': 3}, 'age_hist': {'15': 1, '14': 1, '13': 1}} print('\nFIXTURE calendar-vs-age expected', json.dumps(exp, sort_keys=True)); print('FIXTURE calendar-vs-age actual ', json.dumps(got, sort_keys=True)) self.assertEqual(got, exp) def test_widened_interval_stays_by_38_but_not_confined(self): recs = self.rows(widen=True); got = window_buckets(recs, 37, 38) self.assertEqual(recs['r24']['first_grade_interval'], {'after': 35, 'by': 38}) exp = {'window': [37, 38], 'by_hi': 3, 'confined': 2, 'overlapping_not_confined': 1, 'calendar_hist': {'38': 3}, 'age_hist': {'15': 1, '14': 1, '13': 1}} print('\nFIXTURE widened expected', json.dumps(exp, sort_keys=True)); print('FIXTURE widened actual ', json.dumps(got, sort_keys=True)) self.assertEqual(got, exp) # widened row kept visible, never dropped c = summarize(recs, 23)[24]['first_grade_by_calendar_close']['38'] self.assertEqual(c, {'total': 1, 'confined_one_close': 0, 'overlapping_not_confined': 1}) if __name__ == '__main__': unittest.main()