from pathlib import Path import tempfile import unittest from paper_atlas.audits.histories import report from paper_atlas.db.store import Store def add_history_fixture(store): def event(work, forum, venue, year, track, status, observed): paper, record = store.record('history_fixture', work + '/' + forum, 'https://example.org/' + forum, work + ' title\x00 preserved', 'Abstract', ['Jane Doe'], {'arxiv': work}) store.decision(paper, record, venue, year, track, forum, 'https://example.org/' + forum + '/' + str(status), status, status or '', identities=[('openreview', forum)], observed_at=observed) return paper # The negative observation was collected later: observation order is not chronology. event('2501.00001', 'HistoryA01', 'ICLR', 2024, 'main', 'rejected', '2026-02-01T00:00:00Z') event('2501.00001', 'HistoryA02', 'ICML', 2025, 'main', 'accepted', '2026-01-01T00:00:00Z') event('2501.00002', 'HistoryB01', 'ICLR', 2024, 'main', 'rejected', '2026-01-01T00:00:00Z') event('2501.00002', 'HistoryB01', 'ICLR', 2024, 'main', 'accepted', '2026-02-01T00:00:00Z') event('2501.00003', 'HistoryC01', 'ICLR', 2024, 'main', 'desk_rejected', '2026-01-01T00:00:00Z') event('2501.00003', 'HistoryC02', 'ICLR', 2024, 'workshop_example', 'accepted', '2026-02-01T00:00:00Z') event('2501.00004', 'HistoryD01', 'ICML', 2025, 'main', 'accepted', '2026-01-01T00:00:00Z') event('2501.00004', 'HistoryD01', 'ICML', 2025, 'main', 'withdrawn', '2026-02-01T00:00:00Z') event('2501.00004', 'HistoryD02', 'ICLR', 2024, 'main', 'rejected', '2026-01-01T00:00:00Z') event('2501.00005', 'HistoryE01', 'COLM', 2026, 'main', None, '2026-01-01T00:00:00Z') store.conn.commit() class HistoryTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name) / 'sample.sqlite' self.store = Store(self.db) self.addCleanup(self.store.conn.close) def test_distinct_attempts_venues_and_latest_acceptances_are_not_observation_counts(self): add_history_fixture(self.store) result = report(self.db) self.assertEqual(result['counts'], { 'papers_with_submissions': 5, 'papers_with_multiple_submissions': 3, 'papers_with_historical_negative': 4, 'papers_with_current_known_acceptance': 3, 'papers_with_negative_and_acceptance_in_distinct_attempts': 2, 'papers_with_negative_and_acceptance_in_distinct_venues': 1, 'submissions': 8, 'decision_observations': 9, 'attempts_with_historical_negative': 5, 'attempts_with_historical_negative_and_acceptance': 2, }) self.assertEqual(result['latest_attempt_status_counts'], {'accepted': 3, 'desk_rejected': 1, 'rejected': 2, 'unknown': 1, 'withdrawn': 1}) self.assertEqual(len(result['examples']), 2) first = result['examples'][0] self.assertIn('\x00', first['title']) self.assertEqual({r['venue'] for r in first['attempts']}, {'ICLR', 'ICML'}) self.assertTrue(all(r['decisions'][0]['sources'] for r in first['attempts'])) self.assertTrue(all(r['decisions'][0]['decision_date'] is None for r in first['attempts'])) self.assertIn('Not inferred', first['chronology']) self.assertEqual(result['status'], 'observed_history_audit_not_final_dedup') self.assertTrue(all(v == 0 for v in result['evidence_relationship_checks'].values())) def test_missing_and_mislinked_source_evidence_is_reported_without_repair(self): add_history_fixture(self.store) self.store.conn.execute('DELETE FROM decision_sources WHERE decision_id=1') self.store.conn.execute('DELETE FROM submission_sources WHERE submission_id=1') # Record3 belongs to paper2; move it to paper1 to create a real mismatch. self.store.conn.execute('UPDATE source_records SET paper_id=1 WHERE id=3') self.store.conn.commit() result = report(self.db, examples=0) checks = result['evidence_relationship_checks'] self.assertEqual(checks['decisions_without_source_links'], 1) self.assertEqual(checks['submissions_without_source_links'], 1) self.assertEqual(checks['submission_source_paper_mismatches'], 1) self.assertEqual(checks['decision_source_paper_mismatches'], 2) self.assertEqual(self.store.conn.execute('SELECT paper_id FROM source_records WHERE id=3').fetchone()[0], 1) def test_example_limit_does_not_change_global_counts(self): add_history_fixture(self.store) none, one = report(self.db, examples=0), report(self.db, examples=1) self.assertEqual(none['counts'], one['counts']) self.assertEqual(none['examples'], []) self.assertEqual(len(one['examples']), 1) def test_empty_snapshot_and_invalid_limit_do_not_create_missing_database(self): empty = report(self.db) self.assertTrue(all(v == 0 for v in empty['counts'].values())) self.assertEqual(empty['latest_attempt_status_counts'], {}) absent = self.db.parent / 'absent.sqlite' with self.assertRaisesRegex(ValueError, 'Examples'): report(absent, examples=-1) with self.assertRaisesRegex(ValueError, 'Existing database'): report(absent) self.assertFalse(absent.exists()) def test_truncated_example_keeps_the_cross_venue_acceptance_visible(self): paper, record = self.store.record('fixture', 'many', 'https://example.org/many', 'Many attempts', 'Abstract', []) for index in range(26): accepted = index == 25 forum = f'BoundedForum{index}' self.store.decision(paper, record, 'ICML' if accepted else 'ICLR', 2025, 'main', forum, 'https://example.org/' + forum, 'accepted' if accepted else 'rejected', 'Accept' if accepted else 'Reject', identities=[('openreview', forum)]) self.store.conn.commit() result = report(self.db) example, = result['examples'] self.assertEqual((example['attempt_count'], len(example['attempts'])), (26, 20)) self.assertTrue(example['attempts_truncated']) self.assertEqual({row['latest_observed_status'] for row in example['attempts']}, {'accepted', 'rejected'}) self.assertEqual(result['counts']['submissions'], 26)