from contextlib import closing, redirect_stdout import io import json import os from pathlib import Path import tempfile import unittest import uuid from paper_atlas import querying from paper_atlas.curation.annotations import (assert_annotation, available, initialize, main, observations, retract_annotation) from paper_atlas.db.backend import open_store from paper_atlas.db.store import Store from paper_atlas.postgres import schema from paper_atlas.postgres.client import connect from paper_atlas.postgres.migration import migrate class AnnotationTests(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) self.c = self.store.conn self.paper, self.record = self.store.record('fixture', 'one', 'https://example.org/one', 'First research', None, []) self.other, self.other_record = self.store.record('fixture', 'two', 'https://example.org/two', 'Other research', None, []) self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'original'}]) self.c.commit() def add(self, **changes): args = dict(paper=self.paper, kind='tag', namespace='custom:topic', value='reading', request_id='request-one', evidence_key='manual:reading', reason='Reviewed metadata', actor='reviewer') args.update(changes) return assert_annotation(self.c, **args) def query(self, *args): self.c.commit() with redirect_stdout(io.StringIO()) as output: querying.main(['--db', str(self.db), *map(str, args)]) return json.loads(output.getvalue()) def test_read_does_not_install_extension_and_write_requires_explicit_init(self): self.assertFalse(available(self.c)) self.assertEqual(observations(self.c, self.paper), []) with self.assertRaisesRegex(RuntimeError, 'init explicitly'): self.add() self.assertEqual(self.query('--research-type', 'custom:method=empirical'), []) self.assertFalse(available(self.c)) self.assertEqual(len(schema.inspect(self.c)), 21) def test_schema_upgrade_is_transactional_and_marker_mismatch_fails_closed(self): self.assertTrue(initialize(self.c)); self.c.rollback() self.assertFalse(available(self.c)) self.c.execute("INSERT INTO schema_migrations(name) VALUES('annotation_events_v1')") with self.assertRaisesRegex(RuntimeError, 'mismatch'): initialize(self.c) self.c.rollback() self.assertTrue(initialize(self.c)); self.assertFalse(initialize(self.c)) self.assertEqual(len(schema.inspect(self.c)), 22) def test_unmarked_table_cannot_receive_assertions(self): initialize(self.c) self.c.execute("DELETE FROM schema_migrations WHERE name='annotation_events_v1'") with self.assertRaisesRegex(RuntimeError, 'init explicitly'): self.add() self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0) def test_types_tags_and_platform_labels_are_independent_and_queryable(self): initialize(self.c) self.add() self.add(kind='research_type', namespace='custom:topic', request_id='type', value='reading') self.assertEqual(len(self.query('--tag', 'custom:topic=reading')), 1) self.assertEqual(len(self.query('--research-type', 'custom:topic=reading')), 1) self.assertEqual(len(self.query('--tag', 'fixture:topic=original')), 1) detail = self.query('--paper', self.paper)[0] self.assertEqual(len(detail['annotation_tags']), 1) self.assertEqual(len(detail['research_types']), 1) self.assertNotIn('annotation_history', detail) self.assertEqual(detail['tags'][0]['namespace'], 'fixture:topic') def test_repeated_request_is_idempotent_and_changed_content_fails(self): initialize(self.c); event = self.add() self.assertEqual(self.add(), event) for change in ({'value':'different'}, {'kind':'research_type'}, {'reason':'changed'}, {'paper':self.other}): with self.assertRaisesRegex(ValueError, 'different annotation'): self.add(**change) self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 1) def test_retraction_keeps_original_and_other_assertions_of_same_label(self): initialize(self.c); first = self.add(); second = self.add(request_id='second', evidence_key='independent') kwargs = dict(request_id='retract', reason='No longer relevant', actor='reviewer') removed = retract_annotation(self.c, first, **kwargs) self.assertEqual(retract_annotation(self.c, first, **kwargs), removed) self.assertEqual([r['id'] for r in observations(self.c, self.paper)], [second]) self.assertEqual(len(self.query('--paper', self.paper, '--annotation-history')[0]['annotation_history']), 3) with self.assertRaisesRegex(ValueError, 'already retracted'): retract_annotation(self.c, first, **{**kwargs,'request_id':'duplicate'}) with self.assertRaisesRegex(ValueError, 'existing assertion'): retract_annotation(self.c, removed, **{**kwargs,'request_id':'nested'}) def test_source_assertions_require_matching_paper_and_platform(self): initialize(self.c) for change in ({'namespace':'fixture:type'}, {'namespace':'other:type','source_record_id':self.record}, {'namespace':'fixture:type','source_record_id':self.other_record}): with self.assertRaises(ValueError): self.add(**change) event = self.add(kind='research_type', namespace='fixture:declared_type', source_record_id=self.record) with self.assertRaisesRegex(ValueError, 'Manual CLI'): retract_annotation(self.c, event, request_id='cannot', reason='x', actor='y', custom_only=True) self.assertEqual(observations(self.c, self.paper)[0]['source_record_id'], self.record) def test_platform_tag_refresh_preserves_both_custom_kinds(self): initialize(self.c); self.add(); self.add(kind='research_type', request_id='type') self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'refreshed'}]) self.assertEqual(len(observations(self.c, self.paper)), 2) self.assertEqual(len(self.query('--tag','fixture:topic=refreshed')), 1) self.assertEqual(self.query('--tag','fixture:topic=original'), []) def test_merge_preserves_retractions_evidence_and_original_paper_ids(self): initialize(self.c); event = self.add(source_record_id=self.record) removed = retract_annotation(self.c, event, request_id='withdraw', reason='corrected', actor='reviewer') active = self.add(request_id='active', kind='research_type', value='survey') self.store.merge(self.other, self.paper, 'Independently reviewed identity') rows = observations(self.c, self.other, history=True) self.assertEqual([r['id'] for r in rows], [event, removed, active]) self.assertEqual({r['original_paper_id'] for r in rows}, {self.paper}) self.assertEqual({r['paper_id'] for r in rows}, {self.other}) self.assertEqual([r['id'] for r in observations(self.c, self.other)], [active]) self.assertEqual(self.add(source_record_id=self.record), event) # exact retry survives canonical merge self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(), []) def test_invalid_input_does_not_create_events(self): initialize(self.c) for change in ({'paper':True}, {'paper':999999}, {'kind':'main'}, {'namespace':'arxiv'}, {'namespace':'custom:topic;DROP'}, {'value':' '}, {'reason':''}, {'actor':''}, {'request_id':''}, {'evidence_key':''}, {'source_record_id':True}, {'evidence_url':'file:///etc/passwd'}, {'evidence_url':'https://user:secret@example.org/'}): with self.subTest(change=change), self.assertRaises(ValueError): self.add(**change) self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0) def test_cli_manual_scope_and_complete_assert_retract_lifecycle(self): with redirect_stdout(io.StringIO()): main(['init','--db',str(self.db)]) main(['assert','--db',str(self.db),'--paper',str(self.paper),'--kind','research_type', '--namespace','custom:method','--value','empirical','--evidence-key','manual:review', '--request-id','cli-one','--reason','Read abstract','--actor','reviewer']) self.assertEqual(len(self.query('--research-type','custom:method=empirical')), 1) event = observations(self.c, self.paper)[0]['id'] with redirect_stdout(io.StringIO()): main(['retract','--db',str(self.db),'--event',str(event),'--request-id','cli-two', '--reason','Corrected classification','--actor','reviewer']) self.assertEqual(self.query('--research-type','custom:method=empirical'), []) with redirect_stdout(io.StringIO()) as out: main(['list','--db',str(self.db),'--paper',str(self.paper),'--history']) self.assertEqual(len(json.loads(out.getvalue())), 2) @unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires approved remote PostgreSQL service') class PostgreSQLAnnotationTests(unittest.TestCase): def setUp(self): self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE'] self.target = 'pa_anno_' + uuid.uuid4().hex[:12] self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup) self.db = Path(self.tmp.name) / 'source.sqlite' self.s = Store(self.db) self.paper, self.record = self.s.record('fixture','one','https://example.org','Paper',None,[]) self.s.conn.commit() self.addCleanup(self.s.conn.close); self.addCleanup(self.cleanup) def cleanup(self): with connect(self.service) as c: for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.target,)): c.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE') def migrate(self): self.s.conn.commit() self.s.conn.close() # Freeze/checkpoint the fixture, just like a real source migration. with redirect_stdout(io.StringIO()): return migrate(self.db, self.service, Path(self.tmp.name)/'report.json', schema=self.target, min_free_gib=0) def args(self, **changes): a = dict(paper=self.paper,kind='research_type',namespace='custom:method',value='survey\x00and\x01', request_id='native',evidence_key='manual',reason='Reviewed\x00evidence',actor='reviewer') a.update(changes); return a def test_native_schema_upgrade_roundtrip_retraction_and_queries(self): self.migrate() with closing(open_store(pg_service=self.service, pg_schema=self.target).conn) as c: self.assertTrue(initialize(c)); self.assertFalse(initialize(c)) event = assert_annotation(c, **self.args()); c.commit() self.assertEqual(assert_annotation(c, **self.args()), event) self.assertEqual(observations(c,self.paper)[0]['value'], 'survey\x00and\x01') c.commit() with redirect_stdout(io.StringIO()) as out: querying.main(['--pg-service',self.service,'--pg-schema',self.target,'--paper',str(self.paper)]) self.assertEqual(json.loads(out.getvalue())[0]['research_types'][0]['id'], event) with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c: retract_annotation(c,event,request_id='removed',reason='Changed',actor='reviewer');c.commit() self.assertEqual(observations(c,self.paper), []) self.assertEqual(len(observations(c,self.paper,history=True)), 2) def test_initialized_sqlite_extension_migrates_all_rows_hashes_and_identity(self): initialize(self.s.conn) event = assert_annotation(self.s.conn, **self.args()) retract_annotation(self.s.conn,event,request_id='retracted',reason='Corrected',actor='reviewer') report = self.migrate() self.assertEqual(len(report['tables']),22) self.assertEqual(report['tables']['annotation_events']['source_rows'],2) self.assertEqual(report['tables']['annotation_events']['source_sha256'],report['tables']['annotation_events']['target_sha256']) with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c: new = assert_annotation(c, **self.args(request_id='after-migration')) self.assertGreater(new,2); c.commit() self.assertEqual(len(observations(c,self.paper,history=True)),3) def test_native_upgrade_rollback_keeps_legacy_schema_usable(self): self.migrate() with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c: initialize(c); c.rollback() self.assertFalse(available(c)) self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1) self.assertTrue(initialize(c)) with self.assertRaises(ValueError): assert_annotation(c, **self.args(namespace='wrong:type',source_record_id=self.record)) self.assertEqual(c.execute('SELECT count(*) FROM annotation_events').fetchone()[0],0) c.rollback()