"""Cross-version migration: old research remains intact, framework remains clean.""" from __future__ import annotations import importlib.util import json import os from pathlib import Path import shutil import sys import tempfile import unittest from zipfile import ZipFile, ZIP_DEFLATED ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('c4h_migrate', ROOT / 'tools' / 'migrate.py') mig = importlib.util.module_from_spec(spec) sys.modules[spec.name] = mig spec.loader.exec_module(mig) class MigrationTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.base = Path(self.tmp.name) self.old = self.base / 'old' self.new = self.base / 'new' shutil.copytree(ROOT, self.old, ignore=shutil.ignore_patterns('.git', '__pycache__', '.pytest_cache')) shutil.copytree(ROOT, self.new, ignore=shutil.ignore_patterns('.git', '__pycache__', '.pytest_cache')) self.bundle = self.base / 'research.c4h.zip' def make_research(self): (self.old / 'topic.md').write_text('历史主题：托儿所与社会责任\n', encoding='utf-8') (self.old / 'USERINFO.md').write_text('只使用中文。\n', encoding='utf-8') (self.old / 'STATE.md').write_text('已精读三份史料；正在准备章节一。\n', encoding='utf-8') (self.old / '01_sources/originals/primary/ARCHIVE.pdf').write_bytes(b'%PDF-1.7\nraw original bytes\x00\xff') (self.old / '02_evidence/EV-0001-sample.md').write_text('原文已校核，第32页。', encoding='utf-8') (self.old / '06_arguments/syntheses/SYN-0001-compare.md').write_text('多份材料形成对照。', encoding='utf-8') (self.old / '07_manuscript/sections/01-draft.md').write_text('第一节：制度变化。', encoding='utf-8') (self.old / '09_outputs/稿件终稿.docx').write_bytes(b'office binary\x00') (self.old / '00_governance/research-log.md').write_text('2026-10-17 阅读完成', encoding='utf-8') (self.old / '.env').write_text('SECRET_SHOULD_NOT_LEAK=true') (self.old / 'AGENTS.md').write_text('OLD USER RULE: do not discard this customization', encoding='utf-8') (self.old / '.agents/skills/custom-skill/SKILL.md').parent.mkdir(parents=True, exist_ok=True) (self.old / '.agents/skills/custom-skill/SKILL.md').write_text('custom code/skill', encoding='utf-8') def test_export_and_restore_protects_original_and_rules(self): self.make_research() initial_original = (self.old / '01_sources/originals/primary/ARCHIVE.pdf').read_bytes() original_rules = (self.new / 'AGENTS.md').read_bytes() package = mig.export(self.old, self.bundle, quiet=True) names = {x['path']: x['kind'] for x in package['entries']} self.assertEqual(names['AGENTS.md'], 'framework_customization') self.assertEqual(names['.agents/skills/custom-skill/SKILL.md'], 'framework_customization') self.assertIn('01_sources/originals/primary/ARCHIVE.pdf', names) self.assertNotIn('.env', names) self.assertNotIn('tools/migrate.py', names) self.assertNotIn('tools/migration_baselines.json', names) summary = mig.restore(self.bundle, self.new, quiet=True) self.assertGreater(summary['fill-stock-placeholder'], 1) self.assertEqual((self.new / 'USERINFO.md').read_text(encoding='utf-8'), '只使用中文。\n') self.assertEqual((self.new / 'STATE.md').read_text(encoding='utf-8'), '已精读三份史料；正在准备章节一。\n') self.assertEqual((self.new / 'topic.md').read_text(encoding='utf-8'), '历史主题：托儿所与社会责任\n') self.assertEqual((self.new / '01_sources/originals/primary/ARCHIVE.pdf').read_bytes(), initial_original) self.assertEqual((self.old / '01_sources/originals/primary/ARCHIVE.pdf').read_bytes(), initial_original) self.assertEqual((self.new / 'AGENTS.md').read_bytes(), original_rules) self.assertFalse((self.new / '.agents/skills/custom-skill').exists()) self.assertTrue(list((self.new / '.code4history-migration/conflicts').glob('*/framework_customizations/AGENTS.md'))) self.assertTrue(list((self.new / '.code4history-migration/conflicts').glob('*/framework_customizations/.agents/skills/custom-skill/SKILL.md'))) self.assertTrue((self.new / '09_outputs/稿件终稿.docx').exists()) def test_exchange_area_and_pending_upload_survive_upgrade(self): # The workbench is project data, even when the pending files are gitignored. self.make_research() for name in ('upload','todo','requests','handoff'): (self.old/name).mkdir(exist_ok=True) (self.old/'upload/unread.pdf').write_bytes(b'%PDF-1.4\nnever throw away') (self.old/'upload/unread.pdf.source.md').write_text('Archive reference unknown; ask researcher.') (self.old/'todo/TODO-0007-confirm.md').write_text('---\nstatus: open\n---\nNeed original page.') (self.old/'requests/recheck.md').write_text('Please compare these two translations.') (self.old/'handoff/latest.md').write_text('研究进展：等待补充原件。') mig.export(self.old,self.bundle,quiet=True) mig.restore(self.bundle,self.new,quiet=True) self.assertEqual((self.new/'upload/unread.pdf').read_bytes(),b'%PDF-1.4\nnever throw away') self.assertEqual((self.new/'upload/unread.pdf.source.md').read_text(),'Archive reference unknown; ask researcher.') self.assertIn('Need original page.',(self.new/'todo/TODO-0007-confirm.md').read_text()) self.assertIn('translations.',(self.new/'requests/recheck.md').read_text()) self.assertEqual((self.new/'handoff/latest.md').read_text(),'研究进展：等待补充原件。') self.assertTrue((self.old/'upload/unread.pdf').exists()) def test_existing_user_content_not_overwritten(self): self.make_research() (self.new / 'topic.md').write_text('新项目自己已经有主题，不能覆盖。', encoding='utf-8') mig.export(self.old, self.bundle, quiet=True) summary = mig.restore(self.bundle, self.new, quiet=True) self.assertGreater(summary['conflict'], 0) self.assertEqual((self.new / 'topic.md').read_text(encoding='utf-8'), '新项目自己已经有主题，不能覆盖。') incoming = list((self.new / '.code4history-migration/conflicts').glob('*/incoming/topic.md')) self.assertEqual(len(incoming), 1) self.assertIn('历史主题', incoming[0].read_text(encoding='utf-8')) def test_tampered_bundle_rejected_before_writes(self): self.make_research() mig.export(self.old, self.bundle, quiet=True) damaged = self.base / 'tampered.zip' with ZipFile(self.bundle, 'r') as src, ZipFile(damaged, 'w', ZIP_DEFLATED) as dst: for member in src.namelist(): data = src.read(member) if member == 'research/topic.md': data = b'TAMPERED' dst.writestr(member, data) original_topic = (self.new / 'topic.md').read_bytes() with self.assertRaises(ValueError): mig.restore(damaged, self.new, quiet=True) self.assertEqual((self.new / 'topic.md').read_bytes(), original_topic) self.assertFalse((self.new / '02_evidence/EV-0001-sample.md').exists()) def test_export_standalone_and_dry_run(self): self.make_research() manifest = mig.export(self.old, self.bundle, dry_run=True, quiet=True) self.assertTrue(any(x['path'] == 'topic.md' for x in manifest['entries'])) self.assertFalse(self.bundle.exists()) mig.export(self.old, self.bundle, quiet=True) before = (self.new / 'topic.md').read_bytes() summary = mig.restore(self.bundle, self.new, dry_run=True, quiet=True) self.assertGreater(summary['fill-stock-placeholder'], 0) self.assertEqual((self.new / 'topic.md').read_bytes(), before) def test_reject_unsafe_bundle_path(self): b = self.base / 'evil.zip' payload = {'format': mig.BUNDLE_TYPE, 'version': 1, 'entries': [ {'path': '../other', 'kind': 'research', 'size': 1, 'sha256': '0'*64} ]} with ZipFile(b, 'w') as z: z.writestr('manifest.json', json.dumps(payload)) z.writestr('research/../other', 'x') with self.assertRaises(ValueError): mig.restore(b, self.new, quiet=True) def test_restore_rejects_target_symlink(self): self.make_research() mig.export(self.old, self.bundle, quiet=True) original = self.base / 'outer' original.mkdir() shutil.rmtree(self.new / '02_evidence') try: (self.new / '02_evidence').symlink_to(original, target_is_directory=True) except (OSError, NotImplementedError): self.skipTest('OS disallows symlink') with self.assertRaises(ValueError): mig.restore(self.bundle, self.new, quiet=True) self.assertEqual(list(original.iterdir()), []) if __name__ == '__main__': unittest.main() class LegacyUpgradeTests(unittest.TestCase): def test_v01_and_v03_upgrade_using_real_old_archives(self): import subprocess for version in ('0.1', '0.3', '0.6'): with self.subTest(version=version), tempfile.TemporaryDirectory() as td: home = Path(td) archive = Path(os.environ.get('C4H_LEGACY_ARCHIVE_DIR', str(ROOT.parent))) / f'code4history-v{version}.zip' if not archive.exists(): self.skipTest('可设置 C4H_LEGACY_ARCHIVE_DIR 指向旧版本 ZIP 文件所在目录进行回归测试') with ZipFile(archive) as z: z.extractall(home / 'previous') old = home / 'previous/code4history' (old / 'topic.md').write_text(f'选题版本{version}，实际研究问题。', encoding='utf-8') (old / 'STATE.md').write_text('完成了若干篇史料精读。', encoding='utf-8') (old / '01_sources/originals/primary/copy.pdf').write_bytes(b'%PDF-1.4\nold research data') template = home / 'code4history-v0.8.zip' with ZipFile(template, 'w', ZIP_DEFLATED) as z: for f in ROOT.rglob('*'): if f.is_file() and '__pycache__' not in f.parts: z.write(f, 'code4history/' + f.relative_to(ROOT).as_posix()) target = home / 'new-code4history' mig.upgrade(old, template, target) self.assertEqual((target / 'topic.md').read_text(encoding='utf-8'), f'选题版本{version}，实际研究问题。') self.assertEqual((target / 'STATE.md').read_text(encoding='utf-8'), '完成了若干篇史料精读。') self.assertEqual((target / '01_sources/originals/primary/copy.pdf').read_bytes(), b'%PDF-1.4\nold research data') self.assertIn("Codex instructions", (target / "AGENTS.md").read_text(encoding="utf-8")) self.assertTrue(old.is_dir())