kangshifu-board/server/voc-library/test_query_skill_install.py

import hashlib
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

MODULE = Path(__file__).resolve().parents[2] / 'scripts/install-query-skills.py'
spec = importlib.util.spec_from_file_location('query_install', MODULE)
installer = importlib.util.module_from_spec(spec)
spec.loader.exec_module(installer)


class QuerySkillInstallTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name) / 'data'
        self.package = self.root / 'delivery'
        self.targets = self.root / 'skills'
        baselines = {}
        for name in installer.NAMES:
            source = self.package / 'skills' / name
            source.mkdir(parents=True)
            (source / 'SKILL.md').write_text('new rules', encoding='utf-8')
            (source / 'FILES.sha256').write_text(hashlib.sha256(b'new rules').hexdigest() + '  SKILL.md\n', encoding='utf-8')
            target = self.targets / name
            target.mkdir(parents=True)
            (target / 'SKILL.md').write_text('old rules', encoding='utf-8')
            (target / 'FILES.sha256').write_text(hashlib.sha256(b'old rules').hexdigest() + '  SKILL.md\n', encoding='utf-8')
            baselines[name] = [installer.snapshot(target)]
        (self.package / 'baselines.json').write_text(json.dumps(baselines), encoding='utf-8')
        (self.targets / 'ksf-workshop-runtime/credential-binding.json').write_text('private sentinel', encoding='utf-8')
        self.state = self.root / 'ksf-workshop-state/sessions'
        self.state.mkdir(parents=True)
        (self.state / 'existing.json').write_text('existing session', encoding='utf-8')

    def test_read_only_plan_apply_and_noop_preserve_binding_and_state(self):
        before = installer.snapshot(self.root)
        plan = installer.install(self.package, self.root)
        self.assertTrue(plan['readOnly']); self.assertFalse(plan['applied'])
        self.assertEqual(installer.snapshot(self.root), before)
        binding = (self.targets / 'ksf-workshop-runtime/credential-binding.json').read_bytes()
        state = installer.snapshot(self.root / 'ksf-workshop-state')
        result = installer.install(self.package, self.root, True)
        self.assertTrue(result['filesystemVerified'])
        self.assertEqual((self.targets / 'ksf-workshop-runtime/credential-binding.json').read_bytes(), binding)
        self.assertEqual(installer.snapshot(self.root / 'ksf-workshop-state'), state)
        self.assertEqual((Path(result['backup']) / 'ksf-workshop-runtime/SKILL.md').read_text(), 'old rules')
        self.assertFalse(installer.install(self.package, self.root, True)['applied'])

    def test_unknown_baseline_and_tampered_package_fail_before_writing(self):
        (self.targets / 'ksf-voc-library/SKILL.md').write_text('concurrent change', encoding='utf-8')
        before = installer.snapshot(self.root)
        with self.assertRaisesRegex(ValueError, 'UNKNOWN_SKILL_BASELINE'):
            installer.install(self.package, self.root, True)
        self.assertEqual(installer.snapshot(self.root), before)
        (self.package / 'skills/ksf-workshop-runtime/SKILL.md').write_text('tampered', encoding='utf-8')
        with self.assertRaisesRegex(ValueError, 'PACKAGE_HASH_MISMATCH'):
            installer.install(self.package, self.root, True)

    def test_partial_install_failure_restores_only_our_changes(self):
        before = installer.snapshot(self.targets)
        replace = installer.os.replace
        count = 0
        def fail_second(source, target):
            nonlocal count
            count += 1
            if count == 2:
                raise OSError('simulated failure')
            return replace(source, target)
        with patch.object(installer.os, 'replace', fail_second):
            with self.assertRaisesRegex(OSError, 'simulated failure'):
                installer.install(self.package, self.root, True)
        self.assertEqual(installer.snapshot(self.targets), before)