import json from pathlib import Path import tempfile import unittest from paper_atlas.audits.coverage import report from paper_atlas.db.store import Store def add_coverage_evidence(store): paper, record = store.record('official_metadata', 'known', 'https://example.org/known', 'Known paper with two source observations', 'Complete canonical abstract', ['Jane Doe', 'John Smith'], {'openreview': 'coverageForum', 'arxiv': '2401.98765'}) edition = store.decision(paper, record, 'COLING+LREC', 2024, 'main', 'known', 'https://example.org/accept', 'accepted', 'Accept', aliases=['COLING', 'LREC'], identities=[('openreview', 'coverageForum')], observed_at='2025-05-01T00:00:00Z') same, mirror = store.record('bibliography', 'mirror', 'https://example.org/mirror', 'Known paper with two source observations', None, ['Jane Doe', 'John Smith'], {'openreview': 'coverageForum'}) store.decision(same, mirror, 'COLING+LREC', 2024, 'main', 'mirror', 'https://example.org/older', 'rejected', 'Reject', identities=[('openreview', 'coverageForum')], observed_at='2025-05-01T00:00:00Z', decision_date='2025-04-30T00:00:00Z') unknown, unknown_record = store.record('unknown_catalogue', 'unknown', 'https://example.org/unknown', 'Indexed work without a public decision', None, []) store.decision(unknown, unknown_record, 'COLING+LREC', 2024, 'main', 'unknown', 'https://example.org/unknown', None, '', identities=[('openreview', 'unknownForum')]) same, later = store.record('official_metadata', 'later', 'https://example.org/later', 'Known paper with two source observations', 'Complete canonical abstract', ['Jane Doe', 'John Smith'], {'openreview': 'coverageNext', 'arxiv': '2401.98765'}) store.decision(same, later, 'COLING+LREC', 2025, 'findings', 'later', 'https://example.org/later', 'accepted', 'Accept', identities=[('openreview', 'coverageNext')], observed_at='2025-06-01T00:00:00Z') source = store.source('official_metadata', 'https://example.org') for url, status in [('https://example.org/list', 'complete'), ('https://example.org/identity-refresh', 'partial')]: store.conn.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count) VALUES(?,?,?,?,?,?)''', (edition, source, url, status, 1, 1)) store.conn.commit() class CoverageTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) self.root=Path(self.tmp.name);self.db=self.root/'papers.sqlite';self.store=Store(self.db);self.addCleanup(self.store.conn.close) self.registry=self.root/'venues.json';self.registry.write_text(json.dumps({'scope_note':'Candidate scope','venues':[{'acronym':'COLING','field':'language'},{'acronym':'AAAI','field':'general_ai'}]})) def test_zero_coverage_targets_are_listed_and_joint_alias_is_counted_once(self): paper,record=self.store.record('acl_anthology','joint-paper','https://example.org','Joint paper','Abstract',['Jane Doe']) edition=self.store.decision(paper,record,'COLING+LREC',2024,'main','joint-paper','https://example.org','accepted','Accepted',aliases=['COLING','LREC']) self.store.decision(paper,record,'COLING+LREC',2024,'main','joint-paper','https://example.org/another-observation','accepted','Accepted',aliases=['COLING']) self.store.conn.commit();result=report(self.db,self.registry) self.assertEqual(result['target_venue_count'],2);self.assertEqual(result['venues_with_imported_records'],1) coling,aaai=result['venues'];self.assertEqual(coling['imported_submissions'],1) self.assertEqual(coling['editions'][0]['canonical_venue'],'COLING+LREC') self.assertEqual(aaai['history_status'],'no_imported_records');self.assertEqual(aaai['editions'],[]) self.assertNotEqual(coling['history_status'],'complete') def test_same_paper_across_years_is_not_two_distinct_papers(self): paper,record=self.store.record('source','one','https://example.org','A paper','Abstract',[]) for year in (2024,2025): self.store.decision(paper,record,'COLING',year,'main',str(year),'https://example.org','accepted','Accepted') self.store.conn.commit();coling=report(self.db,self.registry)['venues'][0] self.assertEqual(coling['imported_submissions'],2);self.assertEqual(coling['distinct_papers'],1) def test_observations_attempts_and_missing_source_abstracts_stay_separate(self): add_coverage_evidence(self.store) result = report(self.db, self.registry) self.assertEqual(result['engine'], 'sqlite') self.assertEqual(result['database'], str(self.db)) coling = result['venues'][0] self.assertEqual((coling['imported_submissions'], coling['distinct_papers']), (3, 2)) edition, findings = coling['editions'] self.assertEqual((edition['submissions'], edition['papers'], edition['papers_without_abstract_text']), (2, 2, 1)) self.assertEqual(edition['latest_observed_status_counts']['accepted'], 1) self.assertEqual(edition['latest_observed_status_counts']['rejected'], 0) self.assertEqual(edition['latest_observed_status_counts']['unknown'], 1) self.assertEqual(edition['decision_observation_counts']['accepted'], 1) self.assertEqual(edition['decision_observation_counts']['rejected'], 1) self.assertEqual(edition['decision_observation_counts']['unknown'], 0) counts = {row['source']: row for row in edition['source_record_counts']} self.assertEqual(counts['bibliography']['source_records_without_abstract_text'], 1) self.assertEqual(counts['official_metadata']['source_records'], 1) self.assertEqual(len(edition['sources']), 2) self.assertEqual(findings['track'], 'findings') self.assertEqual(result['imported_venues_outside_registry'], []) def test_missing_sqlite_audit_does_not_create_a_database(self): path = self.root / 'absent.sqlite' with self.assertRaisesRegex(ValueError, 'Existing database'): report(path, self.registry) self.assertFalse(path.exists()) def test_empty_editions_remain_visible_after_shared_aggregation(self): self.store.conn.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES('AAAI','AAAI','general_ai','conference')") venue = self.store.conn.execute("SELECT id FROM venues WHERE acronym='AAAI'").fetchone()[0] self.store.conn.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2026,'main')", (venue,)) self.store.conn.commit() aaai = report(self.db, self.registry)['venues'][1] self.assertEqual(aaai['history_status'], 'no_imported_records') self.assertEqual(len(aaai['editions']), 1) edition = aaai['editions'][0] self.assertEqual((edition['submissions'], edition['papers'], edition['papers_without_abstract_text']), (0, 0, 0)) self.assertEqual(sum(edition['latest_observed_status_counts'].values()), 0) self.assertEqual(edition['source_record_counts'], []) if __name__=='__main__':unittest.main()