105 lines
7.1 KiB
Python
105 lines
7.1 KiB
Python
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()
|