180 lines
11 KiB
Python
180 lines
11 KiB
Python
"""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())
|