kangshifu-board/test/test_workshop_platform_baseline_upgrade.py

"""Actual r6 documentation addition, private-file preservation and rollback in an isolated tree."""
import importlib.util
import json
from pathlib import Path
import shutil
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
PACKAGE = ROOT / 'docs/ksf-workshop-r10-baseline-upgrade-20261110-r1/package'
BASELINE = ROOT / 'docs/workshop-skill-release-20261109-r7/rehearsal-baseline-r6'
spec = importlib.util.spec_from_file_location('platform_upgrade', PACKAGE / 'upgrade.py')
up = importlib.util.module_from_spec(spec)
spec.loader.exec_module(up)


def tree(directory):
    return {p.relative_to(directory).as_posix(): (p.read_bytes(), p.stat().st_mode & 0o777)
            for p in directory.rglob('*') if p.is_file()}


class PlatformBaselineTests(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix='ksf-r6-r10-test-')
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name)
        self.target = self.root / 'skills/ksf-workshop-runtime'
        shutil.copytree(BASELINE, self.target)
        self.output = self.root / 'results'
        self.binding = self.target / 'credential-binding.json'
        self.binding.write_text('fixture-only-private-binding')
        self.binding.chmod(0o600)
        (self.target / 'local-settings.json').write_text('fixture-local-settings')
        self.before = tree(self.target)

    def run_upgrade(self, command):
        return up.run(command, str(self.target), str(self.output), PACKAGE)

    def test_plan_recognizes_exact_r6_without_skill_writes(self):
        result = self.run_upgrade('plan')
        self.assertEqual(result['baselineId'], 'r6')
        self.assertEqual((result['changes'], result['creates']), (24, 1))
        self.assertEqual(tree(self.target), self.before)

    def test_apply_adds_missing_document_and_preserves_full_backup_and_private_files(self):
        self.run_upgrade('plan')
        self.run_upgrade('apply')
        result = self.run_upgrade('verify')
        self.assertTrue(result['ok'])
        self.assertEqual(result['publicFilesVerified'], 43)
        self.assertEqual(tree(self.output / 'before-full'), self.before)
        self.assertTrue((self.target / 'references/workshop-report-design.md').is_file())
        for name, value in self.before.items():
            if name.startswith('scripts/') or name in ['runtime-lock.json', 'credential-binding.json', 'local-settings.json']:
                self.assertEqual(tree(self.target)[name], value)
        after = tree(self.target)
        self.assertTrue(self.run_upgrade('apply')['alreadyApplied'])
        self.assertEqual(tree(self.target), after)

    def test_modified_document_rejected_without_skill_or_result_writes(self):
        p = self.target / 'references/workshop-tasks.md'
        p.write_bytes(p.read_bytes() + b'\nlocal modification\n')
        before = tree(self.target)
        with self.assertRaisesRegex(Exception, 'UPGRADE_BASELINE_CHANGED'):
            self.run_upgrade('plan')
        self.assertEqual(tree(self.target), before)
        self.assertFalse(self.output.exists())

    def test_modified_protected_code_rejected(self):
        p = self.target / 'scripts/ksf-secure.mjs'
        p.write_bytes(p.read_bytes() + b'\nchanged code\n')
        before = tree(self.target)
        with self.assertRaisesRegex(Exception, 'UPGRADE_BASELINE_CHANGED'):
            self.run_upgrade('plan')
        self.assertEqual(tree(self.target), before)
        self.assertFalse(self.output.exists())

    def test_new_document_path_collision_is_not_overwritten(self):
        p = self.target / 'references/workshop-report-design.md'
        p.write_text('independent local work')
        before = tree(self.target)
        with self.assertRaisesRegex(Exception, 'UPGRADE_NEW_FILE_ALREADY_EXISTS'):
            self.run_upgrade('plan')
        self.assertEqual(tree(self.target), before)
        self.assertFalse(self.output.exists())

    def test_change_after_plan_stops_before_backup_or_replacement(self):
        self.run_upgrade('plan')
        self.binding.write_text('fixture-binding-edited-after-plan')
        before = tree(self.target)
        with self.assertRaisesRegex(Exception, 'TARGET_CHANGED_SINCE_PLAN'):
            self.run_upgrade('apply')
        self.assertEqual(tree(self.target), before)
        self.assertFalse((self.output / 'before-full').exists())

    def test_failure_after_new_document_restores_old_files_and_removes_addition(self):
        self.run_upgrade('plan')
        core = up.core_at(PACKAGE)
        original_replace = core.os.replace

        def fail_entry(source, destination):
            if Path(destination) == self.target / 'SKILL.md':
                self.assertTrue((self.target / 'references/workshop-report-design.md').exists())
                raise OSError('fixture replacement failure')
            return original_replace(source, destination)

        with patch.object(up, 'core_at', return_value=core), patch.object(core.os, 'replace', fail_entry):
            with self.assertRaises(OSError):
                self.run_upgrade('apply')
        self.assertEqual(tree(self.target), self.before)
        self.assertEqual(tree(self.output / 'before-full'), self.before)
        self.assertTrue(json.loads((self.output / 'apply.json').read_bytes())['originalRestored'])
        self.assertFalse(list(self.target.rglob('*.ksf-upgrade-new')))

    def test_inventory_contains_only_declared_public_hashes(self):
        result = self.run_upgrade('inventory')
        self.assertEqual(len(result['files']), 43)
        self.assertIsNone(result['files']['references/workshop-report-design.md'])
        self.assertNotIn('credential-binding.json', result['files'])
        self.assertNotIn('fixture-only-private-binding', json.dumps(result))
        self.assertEqual(tree(self.target), self.before)
        self.assertFalse(self.output.exists())


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