kangshifu-board/test/wecom-tool-result-install.test.py

"""Installer fixtures are isolated: no gateway, profiles or connection settings."""
import importlib.util
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parent.parent
spec = importlib.util.spec_from_file_location('hotfix_installer', ROOT / 'scripts/install-wecom-tool-result-hotfix.py')
installer = importlib.util.module_from_spec(spec); spec.loader.exec_module(installer)
BASELINE = Path(os.environ.get('KSF_TOOL_HOTFIX_BASELINE') or ROOT / 'work/adapter-v4-review-20261102/adapter-patch-v4')


class InstallTests(unittest.TestCase):
    def setUp(self):
        temp = tempfile.TemporaryDirectory(prefix='ksf-result-hotfix-'); self.addCleanup(temp.cleanup)
        self.base = Path(temp.name).resolve()
        self.dest = self.base / 'platform'; self.dest.mkdir()
        self.source = self.base / 'package'; (self.source / 'integrations/wecom-identity').mkdir(parents=True)
        manifest = ROOT / 'HOTFIX.json'
        if not manifest.is_file(): manifest = ROOT / 'work/tool-result-hotfix-20261102/HOTFIX.json'
        (self.source / 'HOTFIX.json').write_bytes(manifest.read_bytes())
        for name in installer.BEFORE:
            (self.source / 'integrations/wecom-identity' / name).write_bytes((ROOT / 'integrations/wecom-identity' / name).read_bytes())
        for name in installer.BEFORE:
            origin = BASELINE / ('platform' if name == 'workshop_registration.py' else '') / name
            data = origin.read_bytes()
            self.assertEqual(installer.digest(data), installer.BEFORE[name])
            (self.dest / name).write_bytes(data)
        self.backup = self.base / 'backup'

    def test_readonly_install_repeat_and_no_state_change(self):
        before = {p.name: p.read_bytes() for p in self.dest.iterdir()}
        result = installer.install(self.source, self.dest)
        self.assertTrue(result['readOnly']); self.assertFalse(self.backup.exists())
        self.assertEqual(before, {p.name: p.read_bytes() for p in self.dest.iterdir()})
        result = installer.install(self.source, self.dest, True, self.backup)
        self.assertFalse(result['gatewayRestarted']); self.assertFalse(result['identityStateChanged'])
        self.assertEqual(result['businessWrites'], 0)
        for name, data in before.items():
            self.assertEqual((self.backup / name).read_bytes(), data)
        self.assertTrue(all(x['action'] == 'unchanged' for x in installer.install(self.source, self.dest, True)['files']))

    def test_unknown_version_and_backup_collision_stop_before_writes(self):
        first = self.dest / 'identity_bridge.py'; first.write_text('external-change', encoding='utf-8')
        with self.assertRaisesRegex(installer.PatchError, 'UNKNOWN_TARGET_VERSION'):
            installer.install(self.source, self.dest, True, self.backup)
        self.assertFalse(self.backup.exists())

    def test_failure_rolls_back_and_preserves_concurrent_external_change(self):
        atomic = installer.atomic; count = 0
        def fail_second(path, data, mode, uid, gid):
            nonlocal count
            count += 1
            if count == 2:
                raise OSError('fixture')
            atomic(path, data, mode, uid, gid)
        with patch.object(installer, 'atomic', side_effect=fail_second):
            with self.assertRaises(installer.PatchError) as caught:
                installer.install(self.source, self.dest, True, self.backup)
        self.assertTrue(caught.exception.receipt['rollbackComplete'])
        for name in installer.BEFORE:
            self.assertEqual(installer.digest((self.dest / name).read_bytes()), installer.BEFORE[name])
        count = 0
        def fail_with_concurrent_change(path, data, mode, uid, gid):
            nonlocal count
            count += 1
            if count == 2:
                (self.dest / 'workshop_registration.py').write_text('external-owner-edit', encoding='utf-8')
                raise OSError('fixture')
            atomic(path, data, mode, uid, gid)
        with patch.object(installer, 'atomic', side_effect=fail_with_concurrent_change):
            with self.assertRaises(installer.PatchError) as caught:
                installer.install(self.source, self.dest, True, self.base / 'backup2')
        self.assertFalse(caught.exception.receipt['rollbackComplete'])
        self.assertEqual((self.dest / 'workshop_registration.py').read_text(), 'external-owner-edit')


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