kangshifu-board/test/test_workshop_onboarding.py

"""Fresh-install and preservation tests using the integrity-checked official artifact."""
import argparse
import importlib.util
import io
import json
from pathlib import Path
import tarfile
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
RELEASE = ROOT / 'docs/workshop-onboarding-20261109'
spec = importlib.util.spec_from_file_location('ksf_onboarding', ROOT / 'scripts/workshop-onboarding/install.py')
installer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(installer)


class OnboardingTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='ksf-onboarding-test-')
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.skills = self.root / 'host-skills'
        self.args = argparse.Namespace(skills_dir=str(self.skills), out=str(self.root / 'result'),
            voc_archive=str(RELEASE / 'source-audit/voc-skill-0.3.42.tgz'), apply=False, prepare_runtime=False)

    def run_install(self):
        return installer.run(self.args, RELEASE / 'package')

    def test_dry_checks_both_complete_sources_without_creating_targets(self):
        result = self.run_install()
        self.assertTrue(result['readOnly'])
        self.assertEqual([a['files'] for a in result['actions']], [43, 169])
        self.assertFalse(self.skills.exists())
        self.assertFalse(Path(self.args.out).exists())

    def test_fresh_install_complete_and_repeat_is_verify_only(self):
        self.args.apply = True
        result = self.run_install()
        self.assertTrue(result['filesInstalled'])
        self.assertEqual(result['hostSkillsLoaded'], 'not-verified')
        self.assertEqual(result['vocLiveAccess'], 'not-verified')
        config = json.loads((self.skills / 'voc-intelligence/.mcp.json').read_bytes())
        self.assertEqual(config['mcpServers']['voc']['args'],
                         [str(self.skills / 'voc-intelligence/mcp/src/server.js')])
        marker = self.skills / 'ksf-workshop-runtime/credential-binding.json'
        marker.write_text('fixture-only-do-not-copy')
        before = (self.skills / 'voc-intelligence/package.json').read_bytes()
        self.args.apply = False
        repeated = self.run_install()
        self.assertTrue(all(a['action'] == 'verify-only' for a in repeated['actions']))
        self.assertEqual(marker.read_text(), 'fixture-only-do-not-copy')
        self.assertEqual((self.skills / 'voc-intelligence/package.json').read_bytes(), before)
        self.assertEqual(len(list(self.skills.glob('.ksf-init-*'))), 0)

    def test_existing_different_skill_blocks_before_other_skill_or_receipt_created(self):
        target = self.skills / 'ksf-workshop-runtime'
        target.mkdir(parents=True)
        (target / 'SKILL.md').write_text('unrecognized-existing-skill')
        self.args.apply = True
        with self.assertRaisesRegex(installer.InstallError, 'EXISTING_SKILL_DIFFERS'):
            self.run_install()
        self.assertEqual((target / 'SKILL.md').read_text(), 'unrecognized-existing-skill')
        self.assertFalse((self.skills / 'voc-intelligence').exists())
        self.assertFalse(Path(self.args.out).exists())

    def test_corrupt_archive_stops_before_writes(self):
        archive = self.root / 'bad.tgz'
        archive.write_bytes(b'bad official artifact')
        self.args.voc_archive = str(archive)
        self.args.apply = True
        with self.assertRaisesRegex(installer.InstallError, 'VOC_ARCHIVE_HASH_MISMATCH'):
            self.run_install()
        self.assertFalse(self.skills.exists())

    def test_archive_traversal_and_symlinks_rejected(self):
        for name, kind in [('package/../outside', tarfile.REGTYPE), ('package/link', tarfile.SYMTYPE)]:
            buffer = io.BytesIO()
            with tarfile.open(fileobj=buffer, mode='w:gz') as tar:
                info = tarfile.TarInfo(name)
                info.type = kind
                info.linkname = '../outside' if kind == tarfile.SYMTYPE else ''
                tar.addfile(info, io.BytesIO(b''))
            raw = buffer.getvalue()
            expected = {'bytes': len(raw), 'sha256': installer.digest(raw), 'files': {}}
            with self.assertRaises(installer.InstallError):
                installer.voc_archive_files(raw, expected)

    def test_failure_committing_second_skill_restores_first_and_preserves_unrelated(self):
        self.skills.mkdir()
        unrelated = self.skills / 'another-skill'
        unrelated.mkdir()
        (unrelated / 'SKILL.md').write_text('keep me')
        original = Path.rename

        def rename(source, dest):
            if source.name == 'voc-intelligence':
                raise OSError('synthetic failure')
            return original(source, dest)

        self.args.apply = True
        with patch.object(Path, 'rename', rename), self.assertRaises(OSError):
            self.run_install()
        self.assertFalse((self.skills / 'ksf-workshop-runtime').exists())
        self.assertFalse((self.skills / 'voc-intelligence').exists())
        self.assertEqual((unrelated / 'SKILL.md').read_text(), 'keep me')
        self.assertEqual(len(list(self.skills.glob('.ksf-init-*'))), 0)
        receipt = json.loads((Path(self.args.out) / 'install-receipt.json').read_bytes())
        self.assertFalse(receipt['ok'])

    def test_missing_node_does_not_claim_runtime_ready(self):
        self.args.apply = self.args.prepare_runtime = True
        with patch.object(installer, 'node_and_npm', return_value=None):
            result = self.run_install()
        self.assertTrue(result['filesInstalled'])
        self.assertEqual(result['vocRuntime'], {'status': 'pending', 'reason': 'NODE_OR_NPM_NOT_AVAILABLE'})
        self.assertEqual(result['businessSaveAccess'], 'not-verified')

    def test_result_directory_reuse_and_target_overlap_stop(self):
        self.args.apply = True
        Path(self.args.out).mkdir()
        with self.assertRaisesRegex(installer.InstallError, 'RESULT_DIRECTORY_EXISTS'):
            self.run_install()
        self.assertFalse(self.skills.exists())
        self.args.out = str(self.skills / 'results')
        with self.assertRaisesRegex(installer.InstallError, 'RESULT_TARGET_OVERLAP'):
            self.run_install()


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