kangshifu-board/test/test_workshop_onboarding_upgrade_r10.py

"""Observable r7/r8/r9-to-r10 upgrade, preservation and rollback using the official artifact."""
import argparse
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
OLD = ROOT / 'docs/workshop-onboarding-20261109-r2/package'
NEW = ROOT / 'docs/workshop-onboarding-20261109-r5/package'
ARCHIVE = ROOT / 'docs/workshop-onboarding-20261109/source-audit/voc-skill-0.3.42.tgz'
spec = importlib.util.spec_from_file_location('upgrade_installer', ROOT / 'scripts/workshop-onboarding/install.py')
installer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(installer)


class UpgradeTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='ksf-upgrade-test-')
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.skills = self.root / 'skills'
        self.args = argparse.Namespace(skills_dir=str(self.skills), out=str(self.root / 'old-result'),
            voc_archive=str(ARCHIVE), apply=True, upgrade=False, prepare_runtime=False)

    def old_install(self):
        installer.run(self.args, OLD)
        self.args.out = str(self.root / 'new-result')
        self.binding = self.skills / 'ksf-workshop-runtime/credential-binding.json'
        self.binding.write_text('local-fixture-binding')

    def test_fresh_r10_and_repeat(self):
        result = installer.run(self.args, NEW)
        self.assertTrue(result['filesInstalled'])
        self.args.apply = False
        repeat = installer.run(self.args, NEW)
        self.assertEqual([r['action'] for r in repeat['actions']], ['verify-only', 'verify-only'])
        self.assertEqual(len(list(self.skills.glob('*/SKILL.md'))), 2)  # workshop plus official external VOC
        self.assertFalse((self.skills / 'ksf-voc-library').exists())

    def test_r7_plan_no_write_upgrade_backup_and_binding_preserved(self):
        self.old_install()
        before = (self.skills / 'ksf-workshop-runtime/SKILL.md').read_bytes()
        self.args.apply = False
        plan = installer.run(self.args, NEW)
        self.assertEqual(plan['actions'][0]['action'], 'upgrade')
        self.assertEqual((self.skills / 'ksf-workshop-runtime/SKILL.md').read_bytes(), before)
        self.assertFalse(Path(self.args.out).exists())
        self.args.apply = self.args.upgrade = True
        applied = installer.run(self.args, NEW)
        self.assertTrue(applied['filesInstalled'])
        self.assertEqual(self.binding.read_text(), 'local-fixture-binding')
        self.assertEqual((Path(self.args.out) / 'before/ksf-workshop-runtime/SKILL.md').read_bytes(), before)
        installer.verify_component(self.skills / 'ksf-workshop-runtime',
            installer.source_files(NEW, installer.verify_package(NEW), ARCHIVE)['ksf-workshop-runtime'])
        self.args.apply = False
        self.assertEqual(installer.run(self.args, NEW)['actions'][0]['action'], 'verify-only')

    def test_r8_upgrade_preserves_binding_and_verifies(self):
        installer.run(self.args, ROOT / 'docs/workshop-onboarding-20261109-r3/package')
        self.args.out = str(self.root / 'r9-result')
        binding = self.skills / 'ksf-workshop-runtime/credential-binding.json'
        binding.write_text('local-fixture-binding')
        self.args.upgrade = True
        applied = installer.run(self.args, NEW)
        self.assertTrue(applied['filesInstalled'])
        self.assertEqual(binding.read_text(), 'local-fixture-binding')
        self.args.apply = False
        self.assertEqual(installer.run(self.args, NEW)['actions'][0]['action'], 'verify-only')

    def test_r9_upgrade_preserves_identity_config_and_runtime(self):
        installer.run(self.args, ROOT / 'docs/workshop-onboarding-20261109-r4/package')
        self.args.out = str(self.root / 'r10-result')
        skill = self.skills / 'ksf-workshop-runtime'
        protected = {str(p.relative_to(skill)): p.read_bytes() for p in (skill / 'scripts').iterdir() if p.is_file()}
        protected['runtime-lock.json'] = (skill / 'runtime-lock.json').read_bytes()
        for name in ['credential-binding.json', 'private-local.json']:
            (skill / name).write_text('fixture-not-a-real-secret')
        self.args.upgrade = True
        applied = installer.run(self.args, NEW)
        self.assertTrue(applied['filesInstalled'])
        self.assertEqual(applied['actions'][0]['fromVersion'], 'r9')
        for name, before in protected.items():
            self.assertEqual((skill / name).read_bytes(), before)
        for name in ['credential-binding.json', 'private-local.json']:
            self.assertEqual((skill / name).read_text(), 'fixture-not-a-real-secret')
        self.args.apply = False
        self.assertEqual(installer.run(self.args, NEW)['actions'][0]['action'], 'verify-only')

    def test_apply_upgrade_requires_flag_before_writes(self):
        self.old_install()
        before = (self.skills / 'ksf-workshop-runtime/SKILL.md').read_bytes()
        with self.assertRaisesRegex(installer.InstallError, 'UPGRADE_FLAG_REQUIRED'):
            installer.run(self.args, NEW)
        self.assertEqual((self.skills / 'ksf-workshop-runtime/SKILL.md').read_bytes(), before)
        self.assertFalse(Path(self.args.out).exists())

    def test_local_edit_or_code_edit_stops_without_overwrite(self):
        self.old_install()
        path = self.skills / 'ksf-workshop-runtime/references/workshop-tasks.md'
        original = path.read_bytes()
        for target in [path, self.skills / 'ksf-workshop-runtime/scripts/ksf-secure.mjs']:
            old = target.read_bytes()
            target.write_bytes(old + b'\nlocal edit\n')
            self.args.upgrade = True
            with self.assertRaisesRegex(installer.InstallError, 'UPGRADE_BASELINE_CHANGED'):
                installer.run(self.args, NEW)
            self.assertEqual(target.read_bytes(), old + b'\nlocal edit\n')
            self.assertFalse(Path(self.args.out).exists())
            target.write_bytes(old)
        self.assertEqual(path.read_bytes(), original)

    def test_replacement_failure_restores_original_and_keeps_binding(self):
        self.old_install()
        skill = self.skills / 'ksf-workshop-runtime'
        originals = {str(p.relative_to(skill)): p.read_bytes() for p in skill.rglob('*') if p.is_file()}
        real_replace = installer.os.replace

        def fail_entry(source, dest):
            if Path(dest).name == 'SKILL.md':
                raise OSError('synthetic replacement failure')
            return real_replace(source, dest)

        self.args.upgrade = True
        with patch.object(installer.os, 'replace', fail_entry), self.assertRaises(OSError):
            installer.run(self.args, NEW)
        self.assertEqual({str(p.relative_to(skill)): p.read_bytes() for p in skill.rglob('*') if p.is_file()}, originals)
        self.assertFalse(json.loads((Path(self.args.out) / 'install-receipt.json').read_bytes())['ok'])
        self.assertEqual(len(list(self.skills.glob('.ksf-init-*'))), 0)


if __name__ == '__main__':
    unittest.main()