87 lines
4.8 KiB
Python
87 lines
4.8 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.collectors import icml_public_archive as archive
|
|
from paper_atlas.audits.public_openreview import audit
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
|
|
class ICMLPublicArchiveTests(unittest.TestCase):
|
|
def note(self, **changes):
|
|
return {'id': 'PublicRoot1', 'forum': 'PublicRoot1', 'readers': ['everyone'],
|
|
'domain': archive.DOMAIN, 'invitations': [archive.DOMAIN + '/-/Submission'],
|
|
'license': 'CC BY 4.0', 'cdate': 1769251630136,
|
|
'content': {'title': {'value': 'A public rejected submission'},
|
|
'abstract': {'value': 'Public abstract'},
|
|
'authors': {'value': ['Anonymous', 'Jane Doe']},
|
|
'authorids': {'value': ['private@example.org']},
|
|
'venue': {'value': 'Submitted to ICML 2026'},
|
|
'venueid': {'value': archive.VENUEID},
|
|
'primary_area': {'value': 'learning'},
|
|
'pdf': {'value': '/private.pdf'}},
|
|
'details': {'reviews': ['not retained']}, **changes}
|
|
|
|
def test_explicit_platform_classification_requires_public_root_and_correct_scope(self):
|
|
original = self.note()
|
|
normalized = archive.normalize(original)
|
|
self.assertEqual(normalized['status'], 'rejected')
|
|
self.assertEqual(normalized['raw_decision'], archive.VENUEID)
|
|
self.assertEqual(normalized['metadata']['license'], 'CC BY 4.0')
|
|
self.assertEqual(normalized['metadata']['evidence_kind'], 'public_submission_venueid')
|
|
self.assertFalse(normalized['metadata']['original_decision_signature_verified'])
|
|
self.assertNotIn('private@example.org', json.dumps(normalized))
|
|
self.assertNotIn('not retained', json.dumps(normalized))
|
|
for changes in ({'readers': ['authors']}, {'forum': 'AnotherRoot'}, {'ddate': 1000},
|
|
{'domain': 'ICML.cc/2026/Position_Papers'}, {'invitations': []}):
|
|
with self.subTest(changes=changes), self.assertRaises(ValueError):
|
|
archive.normalize(self.note(**changes))
|
|
for venueid in (None, 'ICML.cc/2026/Conference/Submission', 'ICML.cc/2026/Conference/Accepted_Submission'):
|
|
candidate = self.note()
|
|
candidate['content']['venueid'] = {'value': venueid}
|
|
with self.assertRaisesRegex(ValueError, 'classification'):
|
|
archive.normalize(candidate)
|
|
|
|
def collect_fixture(self, out, notes, count=None, corrupt=False):
|
|
data = json.dumps({'count': len(notes) if count is None else count, 'notes': notes}).encode()
|
|
blob = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest()
|
|
with patch.multiple(archive, BYTES=len(data), BLOB_SHA1=blob, EXPECTED=len(notes)), \
|
|
patch.object(archive.shutil, 'disk_usage', return_value=type('Space', (), {'free': 20 * 1024**3})()):
|
|
return archive.collect(out, fetcher=lambda url, limit: data + b'x' if corrupt else data)
|
|
|
|
def test_verified_archive_roundtrips_through_preflight_and_idempotent_import(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
result = self.collect_fixture(root / 'source', [self.note()])
|
|
db = root / 'db.sqlite'
|
|
Store(db).conn.close()
|
|
report = audit(db, root / 'source/public-openreview-inventory.json', root / 'audit.json')
|
|
self.assertEqual(report['status'], 'ready_for_serial_import')
|
|
self.assertEqual(report['source'], archive.SOURCE)
|
|
self.assertEqual(result['files'][0]['status'], 'partial')
|
|
for _ in range(2):
|
|
ingest_file(db, report['files'][0])
|
|
store = Store(db)
|
|
try:
|
|
self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
|
|
self.assertEqual(store.conn.execute('SELECT status FROM decisions').fetchall(), [('rejected',)])
|
|
finally:
|
|
store.conn.close()
|
|
|
|
def test_bad_hash_count_or_duplicate_identity_publishes_nothing(self):
|
|
for kwargs in ({'corrupt': True}, {'count': 99}, {}):
|
|
notes = [self.note(), self.note()] if not kwargs else [self.note()]
|
|
with self.subTest(kwargs=kwargs), tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises(ValueError):
|
|
self.collect_fixture(tmp, notes, **kwargs)
|
|
self.assertEqual(list(Path(tmp).glob('*.jsonl')), [])
|
|
self.assertFalse((Path(tmp) / 'public-openreview-inventory.json').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|