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)