144 lines
7.7 KiB
Python
144 lines
7.7 KiB
Python
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.collectors import aamas, aamas2026
|
|
|
|
|
|
FIXTURE = json.loads(gzip.decompress((Path(__file__).parent/'fixtures/aamas2026-fixture.json.gz').read_bytes()))
|
|
EVIDENCE = {'url': aamas2026.URL, 'observed_at': FIXTURE['receipts']['receipts'][0]['finished_at']}
|
|
|
|
|
|
class Aamas2026Tests(unittest.TestCase):
|
|
def parse(self, html=None):
|
|
return aamas2026.parse_contents(FIXTURE['contents'] if html is None else html, EVIDENCE)
|
|
|
|
def snapshots(self, folder):
|
|
folder.mkdir()
|
|
(folder/'contents.html').write_text(FIXTURE['contents'])
|
|
(folder/'receipts.json').write_text(json.dumps(FIXTURE['receipts']))
|
|
return folder
|
|
|
|
def test_every_publisher_title_author_affiliation_and_page_preserved(self):
|
|
rows, excluded, held = self.parse()
|
|
self.assertEqual((len(rows), len(excluded), len(held)), (639, 3, 1))
|
|
expected = {r['publisher_page']: r for r in FIXTURE['expected_entries']}
|
|
self.assertEqual(len(expected), 643)
|
|
all_rows = rows+excluded+held
|
|
self.assertEqual({r['metadata']['publisher_page'] for r in all_rows}, set(expected))
|
|
for r in all_rows:
|
|
e = expected[r['metadata']['publisher_page']]
|
|
self.assertEqual(r['title'], e['title'])
|
|
self.assertEqual(r['authors'], e['authors'])
|
|
self.assertEqual(r['metadata']['author_affiliations'], e['affiliations'])
|
|
self.assertEqual(r['metadata']['section_heading'], e['section'])
|
|
for r in rows:
|
|
self.assertIsNone(r['abstract'])
|
|
self.assertEqual(r['identifiers'], {})
|
|
if r['external_id'] != '2026/p3178':
|
|
self.assertEqual(r['status'], 'accepted')
|
|
self.assertNotIn('decision_date', r)
|
|
|
|
def test_explicit_withdrawal_preserves_status_date_and_no_pdf(self):
|
|
rows, _, _ = self.parse()
|
|
row, = [r for r in rows if r['external_id'] == '2026/p3178']
|
|
self.assertEqual(row['status'], 'withdrawn')
|
|
self.assertEqual(row['decision_date'], '2026-06-08')
|
|
self.assertEqual(row['raw_decision'], 'Withdrawn June 8, 2026')
|
|
self.assertNotIn('paper_url', row['metadata'])
|
|
self.assertEqual(row['metadata']['publication_state'], 'withdrawn')
|
|
self.assertEqual(sum(r['status'] == 'accepted' for r in rows), 638)
|
|
|
|
def test_retraction_is_not_accepted_or_withdrawn_and_retains_date(self):
|
|
rows, _, held = self.parse()
|
|
self.assertNotIn('2026/p238', {r['external_id'] for r in rows})
|
|
h, = held
|
|
self.assertEqual(h['publication_state'], 'retracted')
|
|
self.assertEqual(h['retraction_date'], '2026-06-08')
|
|
self.assertEqual(h['raw_publication_state'], 'Retracted June 8, 2026')
|
|
self.assertNotIn('status', h)
|
|
self.assertNotIn('paper_url', h['metadata'])
|
|
self.assertEqual(h['metadata']['source_evidence'], EVIDENCE)
|
|
|
|
def test_tracks_formats_and_prior_decisions_are_not_conflated(self):
|
|
rows, _, _ = self.parse()
|
|
counts = {t: sum(r['track'] == t for r in rows) for t in {r['track'] for r in rows}}
|
|
self.assertEqual(counts, {'proceedings_unspecified': 522, 'aaai_transfer': 19,
|
|
'blue_sky': 17, 'doctoral': 34, 'demo': 44, 'jaamas': 3})
|
|
for row in rows:
|
|
self.assertEqual(row['tags'][0]['namespace'], 'aamas_proceedings:publication_format')
|
|
if row['track'] == 'aaai_transfer':
|
|
self.assertIn('none inferred', row['metadata']['prior_submission_note'])
|
|
self.assertNotIn('decision_events', row)
|
|
if row['track'] == 'jaamas':
|
|
self.assertEqual(row['metadata']['publication_format'], 'journal_presentation')
|
|
self.assertIn('not another independent', row['metadata']['track_scope_note'])
|
|
|
|
def test_refuses_changed_year_section_page_link_author_and_retraction(self):
|
|
html = FIXTURE['contents']
|
|
replacements = [('AAMAS 2026:', 'AAMAS 2027:'),
|
|
('Research Paper Track</font>', 'Workshop Track</font>'),
|
|
('(Page 4)', '(Page 1)'),
|
|
('../pdfs/HQQZ1937.pdf', 'https://other.example/paper.pdf'),
|
|
('Michael Wooldridge <i>', '<i>'),
|
|
('Retracted June 8, 2026', 'Withdrawn June 8, 2026')]
|
|
for old, new in replacements:
|
|
self.assertIn(old, html)
|
|
with self.subTest(change=old), self.assertRaises(ValueError):
|
|
self.parse(html.replace(old, new, 1))
|
|
|
|
def test_frozen_collection_is_partial_with_full_held_evidence(self):
|
|
with tempfile.TemporaryDirectory() as tmp, patch.object(aamas2026, 'fetch', side_effect=AssertionError('No HTTP')):
|
|
root = Path(tmp)
|
|
report = aamas.collect(2026, root/'out', snapshots=self.snapshots(root/'source'))
|
|
self.assertEqual(report['records'], 639)
|
|
self.assertEqual(report['status_counts'], {'accepted': 638, 'withdrawn': 1})
|
|
self.assertEqual(len(report['files']), 6)
|
|
self.assertEqual(len(report['held_publications']), 1)
|
|
self.assertFalse(report['database_written'])
|
|
self.assertFalse(report['pdfs_downloaded'])
|
|
for m in report['files']:
|
|
self.assertEqual(m['status'], 'partial')
|
|
self.assertEqual(m['missing_abstract_count'], m['count'])
|
|
self.assertEqual(hashlib.sha256(Path(m['file']).read_bytes()).hexdigest(), m['file_sha256'])
|
|
self.assertTrue((root/'out/aamas-proceedings-2026.json').is_file())
|
|
|
|
def test_frozen_response_rejects_tamper_unsafe_path_status_and_symlink(self):
|
|
for mode in ('tamper', 'path', 'status', 'symlink'):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); source = self.snapshots(root/'source')
|
|
r = json.loads((source/'receipts.json').read_text())
|
|
if mode == 'tamper':
|
|
(source/'contents.html').write_text('bad')
|
|
elif mode == 'path':
|
|
r['receipts'][0]['file'] = '../outside.html'
|
|
elif mode == 'status':
|
|
r['receipts'][0]['status'] = 403
|
|
else:
|
|
(root/'outside.html').write_text(FIXTURE['contents'])
|
|
(source/'contents.html').unlink()
|
|
(source/'contents.html').symlink_to(root/'outside.html')
|
|
(source/'receipts.json').write_text(json.dumps(r))
|
|
with self.assertRaises(ValueError):
|
|
aamas.collect(2026, root/'out', snapshots=source)
|
|
self.assertFalse((root/'out/aamas-proceedings-2026.json').exists())
|
|
|
|
def test_live_mode_fetches_only_html_and_preserves_receipt(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); cache = root/'cache'; cache.mkdir()
|
|
receipt = {'retrieved_at': EVIDENCE['observed_at'], 'url': aamas2026.URL}
|
|
(cache/hashlib.sha256(aamas2026.URL.encode()).hexdigest()).with_suffix('.json').write_text(json.dumps(receipt))
|
|
with patch.object(aamas2026, 'fetch', return_value=FIXTURE['contents'].encode()) as fetch:
|
|
report = aamas.collect(2026, root/'out', cache=cache)
|
|
fetch.assert_called_once_with(aamas2026.URL, cache, 8*1024**2, delay=1)
|
|
self.assertEqual(report['source_receipts'][0]['response_receipt'], receipt)
|
|
|
|
def test_frozen_fixture_is_original_hashed_response(self):
|
|
r, = FIXTURE['receipts']['receipts']
|
|
self.assertEqual(hashlib.sha256(FIXTURE['contents'].encode()).hexdigest(), r['sha256'])
|
|
self.assertEqual(r['bytes'], len(FIXTURE['contents'].encode()))
|