228 lines
13 KiB
Python
228 lines
13 KiB
Python
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()
|