From 4e0e3f0ce82faf9c7ebf26475b37626e7d17e49d Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Fri, 6 Nov 2026 22:15:56 +0800 Subject: [PATCH] Migrate existing PMLR submission tracks with atomic provenance checks --- docs/operations.md | 15 ++ paper_atlas/cli.py | 1 + paper_atlas/ingestion/pmlr_tracks.py | 200 +++++++++++++++++++++++++++ tests/test_pmlr_track_migration.py | 156 +++++++++++++++++++++ 4 files changed, 372 insertions(+) create mode 100644 paper_atlas/ingestion/pmlr_tracks.py create mode 100644 tests/test_pmlr_track_migration.py diff --git a/docs/operations.md b/docs/operations.md index 030827a..a455d06 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -47,3 +47,18 @@ paper-atlas coverage --db /data/paper-index/papers.sqlite \ manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要；`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留，未出现在某份清单不等于被拒稿；第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。 当前代码与合并状态以 [Pull Requests](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls) 为准；CI 结果以对应提交的 Actions 检查为准。工作分支交付后等待用户手动合并。合并代码不自动部署、不自动启动采集，也不代表数据批次已完成。 +## 修正已有 PMLR 投稿的轨道 + +先用采集器 v3 把 ICML 多轨道卷采集到新目录，保留旧文件。以下命令默认只读生成迁移计划；`--inventory` 可重复指定，以完整卷为单位核对，不能只提供 Position 子集： + +```sh +paper-atlas maintain pmlr-tracks --db /data/paper-index/papers.sqlite \ + --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v235.inventory.json \ + --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v267.inventory.json \ + --inventory /data/paper-index/collected/pmlr-tracks-v3/pmlr-v306.inventory.json \ + --out /data/paper-index/collected/pmlr-track-plan.json +``` + +检查报告后，在同一命令添加 `--apply` 并使用新的 `--out` 路径执行迁移。计划使用共享维护锁，实际迁移使用独占锁；已有 arXiv 类别修复或其他写入运行时会拒绝启动，不会停止或重启它们。一次事务内核对整个来源集合、原始题目/摘要/作者、forum 身份、已有事件及目标冲突，更新原投稿的 edition 和身份归属，回填官方轨道出处及逐轨道覆盖数，最后复查轨道与外键后提交。论文、来源、投稿、决定的 ID 及决定观察均保留，不重新创建事件、不做实体合并、不改论文正文元数据或标签。 + +报告记录每个迁移事件的旧 edition、原身份、完整决定观察及新轨道证据。`planned` 是未执行的计划；`complete` 才表示事务已提交；`error` 表示执行失败。若进程在提交前后中断而报告停在 `prepared`，先对照主库和报告核查，不根据日志缺失自行重启任务。新路径重跑会重新核对当前数据库，已正确的轨道不会再次迁移。共享多个来源、重复目标事件或冲突决定会停止迁移，需另行核对，不能直接删掉其中一条。 diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 25722ed..b2d6a7f 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -27,6 +27,7 @@ COMMANDS = { ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'), ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'), ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'), + ('maintain','pmlr-tracks'): ('paper_atlas.ingestion.pmlr_tracks', {}, '核对并迁移既有 PMLR 投稿轨道'), ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'), ('audit','iclr-legacy'): ('paper_atlas.audits.iclr_legacy', {}, '独立核验 ICLR 2020 旧目录投稿身份'), ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'), diff --git a/paper_atlas/ingestion/pmlr_tracks.py b/paper_atlas/ingestion/pmlr_tracks.py new file mode 100644 index 0000000..a528dc0 --- /dev/null +++ b/paper_atlas/ingestion/pmlr_tracks.py @@ -0,0 +1,200 @@ +"""Move existing PMLR submission events to verified tracks without reimporting.""" +import argparse +from contextlib import closing +import fcntl +import hashlib +import json +from pathlib import Path +import sqlite3 + +from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record +from paper_atlas.db.submissions import identities_for_row, timestamp + + +def save(path, result): + part = path.with_suffix('.part') + part.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n') + part.replace(path) + + +def load_inventories(paths): + scopes, seen, forums = [], set(), set() + for filename in paths: + path = Path(filename).resolve() + inventory = json.loads(path.read_text()) + require_track_manifest(inventory) + if inventory.get('source') != 'pmlr' or inventory.get('venue') != 'ICML' or inventory.get('year', 0) < 2024: + raise ValueError('Only verified modern ICML PMLR volumes may be reclassified') + if not inventory.get('count') or not inventory.get('files'): + raise ValueError('A nonempty complete publication inventory is required') + rows, manifests = [], [] + tracks = set() + for manifest in inventory['files']: + require_track_manifest(manifest) + if any(manifest.get(key) != inventory.get(key) for key in ('source', 'venue', 'year', 'volume', 'track_program')): + raise ValueError('Track manifest differs from volume inventory') + if manifest['track'] not in ('main', 'position') or manifest['track'] in tracks: + raise ValueError('Invalid or repeated track in volume inventory') + tracks.add(manifest['track']) + data = Path(manifest['file']).read_bytes() + if hashlib.sha256(data).hexdigest() != manifest.get('file_sha256'): + raise ValueError('Track file differs from its verified inventory') + chunk = [json.loads(line) for line in data.splitlines()] + if not chunk or len(chunk) != manifest['count']: + raise ValueError('Track inventory count mismatch') + for row in chunk: + if any(row.get(key) != manifest.get(key) for key in ('source', 'venue', 'year', 'track')): + raise ValueError('Record differs from its track manifest') + if not row['external_id'].startswith('v' + str(inventory['volume']) + '/') or row['status'] != 'accepted': + raise ValueError('Record is outside the accepted publication scope') + if row['external_id'] in seen: + raise ValueError('Repeated publication in reclassification inventory') + seen.add(row['external_id']) + require_track_record(row, manifest) + own_forum = row.get('identifiers', {}).get('openreview') + if not own_forum or (row['year'], own_forum) in forums: + raise ValueError('Missing or repeated forum identity in one ICML edition') + forums.add((row['year'], own_forum)) + rows.extend(chunk); manifests.append(manifest) + if len(rows) != inventory['count']: + raise ValueError('Volume count differs from its track files') + scopes.append({'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest(), + 'inventory': inventory, 'manifests': manifests, 'rows': rows}) + if not scopes: + raise ValueError('At least one complete volume inventory is required') + return scopes + + +def plan(c, scopes): + updates, changes = [], [] + for scope in scopes: + volume = scope['inventory']['volume'] + source_rows = c.execute('''SELECT r.id,r.external_id FROM source_records r JOIN sources src ON src.id=r.source_id + WHERE src.name='pmlr' AND r.external_id LIKE ?''', ('v' + str(volume) + '/%',)).fetchall() + if {r[1] for r in source_rows} != {r['external_id'] for r in scope['rows']}: + raise ValueError('Existing PMLR volume differs from the complete audited publication inventory') + for row in scope['rows']: + record = c.execute('''SELECT r.id,r.paper_id,r.title,r.abstract,r.authors_json,r.metadata_json FROM source_records r + JOIN sources src ON src.id=r.source_id WHERE src.name='pmlr' AND r.external_id=?''', (row['external_id'],)).fetchone() + if (record[2] != row['title'] or record[3] != row.get('abstract') + or json.loads(record[4]) != row.get('authors', [])): + raise ValueError('Reclassification must preserve existing publisher text and authors: ' + row['external_id']) + metadata = json.loads(record[5]) + forum = str(metadata.get('openreview') or (metadata.get('record_identifiers') or {}).get('openreview') or '') + forum = forum.split('id=')[-1].split('&')[0] + if not forum or forum != row['identifiers'].get('openreview'): + raise ValueError('Publisher forum identity differs from retained source metadata') + events = c.execute('''SELECT s.id,s.paper_id,s.edition_id,e.venue_id,e.year,e.track,v.acronym,s.source_record_id,s.source_submission_id + FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id JOIN editions e ON e.id=s.edition_id + JOIN venues v ON v.id=e.venue_id WHERE ss.source_record_id=? AND ss.source_submission_id=?''', + (record[0], row['external_id'])).fetchall() + if len(events) != 1: + raise ValueError('Publication must have exactly one existing submission event') + event = events[0] + sid, paper, old_edition, venue, year, old_track, acronym, owner, external = event + if paper != record[1] or year != row['year'] or acronym.upper() != 'ICML' or old_track not in ('main', 'position'): + raise ValueError('Existing submission is outside this publication scope') + identities = c.execute('SELECT scheme,value FROM submission_identifiers WHERE submission_id=? ORDER BY scheme,value', (sid,)).fetchall() + own = identities_for_row(row) + if {value for scheme, value in identities if scheme == 'openreview'} - {forum}: + raise ValueError('Existing event has a conflicting forum identity') + target = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue, year, row['track'])).fetchone() + if target: + for scheme, value in sorted(set(identities) | set(own)): + collision = c.execute('SELECT submission_id FROM submission_identifiers WHERE edition_id=? AND scheme=? AND value=?', (target[0], scheme, value)).fetchone() + if collision and collision[0] != sid: + raise ValueError('Target track already has another event for this identity; reconcile sources before moving') + if c.execute('SELECT 1 FROM submissions WHERE source_record_id=? AND source_submission_id=? AND edition_id=? AND id<>?', (owner, external, target[0], sid)).fetchone(): + raise ValueError('Target track already has this publication event') + if old_track != row['track']: + links = c.execute('SELECT source_record_id,source_submission_id FROM submission_sources WHERE submission_id=?', (sid,)).fetchall() + if links != [(record[0], row['external_id'])] or owner != record[0] or external != row['external_id']: + raise ValueError('Moving an event shared by multiple sources requires a separate reconciliation') + decisions = c.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=? ORDER BY id', (sid,)).fetchall() + if not decisions or any(d[1] != 'accepted' for d in decisions): + raise ValueError('Moving a publication with conflicting decision history requires reconciliation') + changes.append({'submission_id': sid, 'source_record_id': record[0], 'external_id': row['external_id'], + 'paper_id': paper, 'from_edition_id': old_edition, 'from_track': old_track, + 'to_track': row['track'], 'year': year, 'before_identifiers': identities, + 'before_decisions': decisions, 'track_evidence': row['metadata']['track_evidence']}) + updates.append((row, record, event, own)) + return updates, changes + + +def reclassify(db, inventories, out, apply=False): + db, out = Path(db).resolve(), Path(out).resolve() + if not db.is_file() or out == db: + raise ValueError('Existing database and distinct report path required') + scopes = load_inventories(inventories) + inputs = {Path(scope['path']) for scope in scopes} + inputs.update(Path(m['file']).resolve() for scope in scopes for m in scope['manifests']) + inputs.update(Path(m['file']).with_suffix('.manifest.json').resolve() for scope in scopes for m in scope['manifests']) + if out in inputs: + raise ValueError('Report must preserve evidence inputs') + out.parent.mkdir(parents=True, exist_ok=True) + with Path(str(db) + '.maintenance.lock').open('a') as lock, Path(str(out) + '.lock').open('a') as output_lock: + fcntl.flock(lock, (fcntl.LOCK_EX if apply else fcntl.LOCK_SH) | fcntl.LOCK_NB) + fcntl.flock(output_lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + if out.exists(): + raise ValueError('Reclassification report exists; use a new path') + mode = 'rw' if apply else 'ro' + with closing(sqlite3.connect(db.as_uri() + '?mode=' + mode, uri=True)) as c: + c.execute('PRAGMA foreign_keys=ON'); c.execute('PRAGMA cache_size=-32768') + c.execute('BEGIN IMMEDIATE' if apply else 'BEGIN') + updates, changes = plan(c, scopes) + result = {'status': 'planned', 'db': str(db), 'started_at': timestamp(), + 'records_verified': len(updates), 'events_to_move': len(changes), 'changes': changes, + 'inputs': [{key: scope[key] for key in ('path', 'sha256')} for scope in scopes], + 'note': 'Preserves paper/source/submission/decision IDs and all decision observations; no entity merges or text reimport.'} + save(out, result) + if not apply: + return result + try: + for row, record, event, own in updates: + sid, _, old_edition, venue, year, old_track, _, _, _ = event + c.execute('INSERT OR IGNORE INTO editions(venue_id,year,track) VALUES(?,?,?)', (venue, year, row['track'])) + target = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?', (venue, year, row['track'])).fetchone()[0] + if target != old_edition: + c.execute('UPDATE submission_identifiers SET edition_id=? WHERE submission_id=?', (target, sid)) + c.execute('UPDATE submissions SET edition_id=? WHERE id=?', (target, sid)) + for scheme, value in own: + c.execute('INSERT OR IGNORE INTO submission_identifiers VALUES(?,?,?,?)', (target, scheme, value, sid)) + metadata = json.loads(record[5]); metadata.update(row['metadata']) + c.execute('UPDATE source_records SET metadata_json=? WHERE id=?', (json.dumps(metadata, ensure_ascii=False), record[0])) + for scope in scopes: + for m in scope['manifests']: + edition = c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? COLLATE NOCASE AND e.year=? AND e.track=?', ('ICML', m['year'], m['track'])).fetchone()[0] + source = c.execute("SELECT id FROM sources WHERE name='pmlr'").fetchone()[0] + c.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) + VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status, + expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL''', + (edition, source, m['list_url'], 'partial' if m.get('missing_abstract_count') or m.get('status') == 'partial' else 'complete', m['count'], m['count'], timestamp())) + _, remaining = plan(c, scopes) + if remaining: + raise ValueError('Post-migration track comparison still has mismatches') + print(json.dumps({'stage': 'checking_foreign_keys', 'events_moved': len(changes)}), flush=True) + if c.execute('PRAGMA foreign_key_check').fetchone(): + raise ValueError('Foreign key violations after track reclassification') + result.update(status='prepared', events_moved=len(changes)) + save(out, result) + c.commit() + except BaseException as error: + c.rollback(); result.update(status='error', error=str(error)); save(out, result); raise + result.update(status='complete', finished_at=timestamp()) + save(out, result) + return result + + +def main(argv=None, *, prog=None): + parser = argparse.ArgumentParser(prog=prog) + parser.add_argument('--db', required=True) + parser.add_argument('--inventory', action='append', required=True) + parser.add_argument('--out', required=True) + parser.add_argument('--apply', action='store_true', help='Apply the verified migration under an exclusive maintenance lock') + args = parser.parse_args(argv) + result = reclassify(args.db, args.inventory, args.out, args.apply) + print(json.dumps({key: value for key, value in result.items() if key != 'changes'}), flush=True) + + +if __name__ == '__main__': + main() diff --git a/tests/test_pmlr_track_migration.py b/tests/test_pmlr_track_migration.py new file mode 100644 index 0000000..92a1367 --- /dev/null +++ b/tests/test_pmlr_track_migration.py @@ -0,0 +1,156 @@ +import copy +import fcntl +import hashlib +import json +from pathlib import Path +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch + +from paper_atlas.collectors.pmlr_tracks import program_tracks, record_track +from paper_atlas.db.store import Store +from paper_atlas.db.submissions import identities_for_row +from paper_atlas.ingestion.pmlr_tracks import reclassify, plan +from paper_atlas.ingestion.records import ingest_file +from tests.test_pmlr_tracks import event, program, record + + +class PmlrMigrationTests(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.inventory = self.root / 'inventory.json' + self.rows = [record(title='A main research paper'), + record('position1', 'v267/paper2', title='A position paper')] + self.proof = {'url': 'https://icml.cc/static/virtual/data/icml-2025-orals-posters.json', 'sha256': 'a' * 64} + mapping, _ = program_tracks(program(event(), event('position1', 'Position_Paper_Track', 2)), 2025) + store = Store(self.db); c = store.conn + for row in self.rows: + row.update(url='https://proceedings.mlr.press/' + row['external_id'] + '.html', + list_url='https://proceedings.mlr.press/v267/', status='accepted') + paper, source = store.record('pmlr', row['external_id'], row['url'], row['title'], row['abstract'], + row['authors'], row['identifiers'], {'openreview': row['identifiers']['openreview']}) + store.decision(paper, source, 'ICML', 2025, 'main', row['external_id'], row['url'], + 'accepted', 'listed in official proceedings', identities=identities_for_row(row)) + record_track(row, mapping, self.proof) + c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','keep me')") + c.execute("INSERT INTO paper_tags VALUES(1,1,NULL,'custom:manual')") + c.commit(); c.close() + self.write_inventory() + + def write_inventory(self): + files = [] + for row in self.rows: + path = self.root / (row['track'] + '.jsonl') + data = (json.dumps(row) + '\n').encode(); path.write_bytes(data) + m = {'source': 'pmlr', 'venue': 'ICML', 'year': 2025, 'volume': 267, + 'parser_version': 3, 'track': row['track'], 'track_program': self.proof, + 'count': 1, 'file': str(path), 'file_sha256': hashlib.sha256(data).hexdigest(), + 'missing_abstract_count': 0, 'list_url': row['list_url'], 'status': 'collected'} + path.with_suffix('.manifest.json').write_text(json.dumps(m)); files.append(m) + self.inventory.write_text(json.dumps({'source': 'pmlr', 'venue': 'ICML', 'year': 2025, + 'volume': 267, 'parser_version': 3, 'track_program': self.proof, + 'count': 2, 'files': files})) + + def tracks(self): + with sqlite3.connect(self.db) as c: + return c.execute('SELECT s.id,e.track FROM submissions s JOIN editions e ON e.id=s.edition_id ORDER BY s.id').fetchall() + + def test_plan_is_read_only_and_reports_exact_original_event(self): + result = reclassify(self.db, [self.inventory], self.root / 'plan.json') + self.assertEqual((result['status'], result['records_verified'], result['events_to_move']), ('planned', 2, 1)) + self.assertEqual(result['changes'][0]['submission_id'], 2) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) + + def test_apply_preserves_ids_decisions_text_tags_and_is_idempotent(self): + with sqlite3.connect(self.db) as c: + before = {table: c.execute('SELECT * FROM ' + table).fetchall() for table in ('papers', 'decisions', 'decision_sources', 'submission_sources', 'paper_tags')} + result = reclassify(self.db, [self.inventory], self.root / 'apply.json', True) + self.assertEqual((result['status'], result['events_moved']), ('complete', 1)) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')]) + with sqlite3.connect(self.db) as c: + for table, rows in before.items(): + self.assertEqual(c.execute('SELECT * FROM ' + table).fetchall(), rows, table) + self.assertEqual(c.execute('SELECT count(*) FROM submission_identifiers i JOIN submissions s ON s.id=i.submission_id WHERE i.edition_id<>s.edition_id').fetchone()[0], 0) + self.assertEqual(c.execute('SELECT e.track,es.imported_count FROM edition_sources es JOIN editions e ON e.id=es.edition_id ORDER BY e.track').fetchall(), [('main', 1), ('position', 1)]) + self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), []) + again = reclassify(self.db, [self.inventory], self.root / 'again.json', True) + self.assertEqual(again['events_moved'], 0) + for m in json.loads(self.inventory.read_text())['files']: + ingest_file(self.db, m['file']) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'position')]) + with sqlite3.connect(self.db) as c: + self.assertEqual(c.execute('SELECT * FROM decisions').fetchall(), before['decisions']) + + def test_conflicting_target_event_stops_without_merging_history(self): + store = Store(self.db) + _, source = store.record('other', 'position1', 'https://example.org', 'Other evidence', None, [], {'openreview': 'position1'}) + store.decision(2, source, 'ICML', 2025, 'position', 'position1', 'https://example.org', 'rejected', 'Reject', identities=[('openreview', 'position1')]) + store.conn.commit(); store.conn.close() + before = self.tracks() + with self.assertRaisesRegex(ValueError, 'another event'): + reclassify(self.db, [self.inventory], self.root / 'blocked.json', True) + self.assertEqual(self.tracks(), before) + + def test_shared_event_requires_reconciliation_and_keeps_all_sources(self): + store = Store(self.db) + _, source = store.record('mirror', 'position1', 'https://example.org', 'Mirror title', None, [], {'openreview': 'position1'}) + store.conn.execute('INSERT INTO submission_sources VALUES(?,?,?)', (2, source, 'position1')) + store.conn.commit(); store.conn.close() + with self.assertRaisesRegex(ValueError, 'multiple sources'): + reclassify(self.db, [self.inventory], self.root / 'shared.json', True) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) + + def test_text_mismatch_in_last_record_prevents_any_changes(self): + with sqlite3.connect(self.db) as c: + c.execute("UPDATE source_records SET abstract='Changed elsewhere' WHERE id=2") + with self.assertRaisesRegex(ValueError, 'preserve existing publisher text'): + reclassify(self.db, [self.inventory], self.root / 'text.json', True) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) + + def test_mutation_failure_rolls_back_editions_metadata_and_identifiers(self): + calls = 0 + def fail_after_updates(c, scopes): + nonlocal calls + calls += 1 + if calls == 2: + raise ValueError('Injected post-migration failure') + return plan(c, scopes) + with sqlite3.connect(self.db) as c: + before = c.execute('SELECT * FROM source_records').fetchall() + with patch('paper_atlas.ingestion.pmlr_tracks.plan', side_effect=fail_after_updates): + with self.assertRaisesRegex(ValueError, 'Injected'): + reclassify(self.db, [self.inventory], self.root / 'rollback.json', True) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) + with sqlite3.connect(self.db) as c: + self.assertEqual(c.execute('SELECT * FROM source_records').fetchall(), before) + self.assertEqual(c.execute("SELECT count(*) FROM editions WHERE track='position'").fetchone()[0], 0) + self.assertEqual(json.loads((self.root / 'rollback.json').read_text())['status'], 'error') + + def test_changed_evidence_and_active_writer_are_refused(self): + out = self.root / 'blocked.json' + with Path(str(self.db) + '.maintenance.lock').open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + with self.assertRaises(BlockingIOError): + reclassify(self.db, [self.inventory], out, True) + self.assertFalse(out.exists()) + path = self.root / 'position.jsonl'; path.write_text(path.read_text() + ' ') + with self.assertRaisesRegex(ValueError, 'differs from its verified inventory'): + reclassify(self.db, [self.inventory], out, True) + self.assertEqual(self.tracks(), [(1, 'main'), (2, 'main')]) + + def test_input_report_and_duplicate_forums_are_refused(self): + original = self.inventory.read_bytes() + with self.assertRaisesRegex(ValueError, 'preserve evidence'): + reclassify(self.db, [self.inventory], self.inventory, True) + self.assertEqual(self.inventory.read_bytes(), original) + self.rows[1]['identifiers']['openreview'] = self.rows[0]['identifiers']['openreview'] + self.rows[1]['metadata']['track_evidence']['identity_value'] = self.rows[0]['identifiers']['openreview'] + self.write_inventory() + with self.assertRaisesRegex(ValueError, 'repeated forum'): + reclassify(self.db, [self.inventory], self.root / 'duplicate.json', True) + + +if __name__ == '__main__': + unittest.main()