kangshifu-board/test/test_workshop_quality_runner.py

"""Runner control-flow checks; exact filesystem mutations are covered by Node."""
import contextlib
import importlib.util
import io
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
COUNTS = {'r2': (17, 14, 3), 'r3': (16, 14, 2), 'r4': (9, 8, 1)}


class RunnerTests(unittest.TestCase):
    def run_case(self, mode='success', version='r4'):
        with tempfile.TemporaryDirectory(prefix='ksf-quality-runner-') as temp:
            temp = Path(temp)
            package = temp / 'package'
            shutil.copytree(ROOT / 'docs/workshop-skill-release-20261109-r5/package', package)
            spec = importlib.util.spec_from_file_location('quality_runner_fixture', package / 'run-update.py')
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            calls = []
            total, updates, creates = COUNTS[version]

            def child(cmd, **kwargs):
                self.assertIsInstance(cmd, list)
                self.assertTrue(kwargs['capture_output'])
                self.assertNotIn('shell', kwargs)
                step = cmd[2]
                calls.append(step)
                if mode == step.removeprefix('--') + '-failure':
                    return subprocess.CompletedProcess(cmd, 1, b'', b'{"ok":false,"error":{"code":"FIXTURE_FAILURE"}}')
                payload = {'ok': True, 'readOnly': True, 'alreadyApplied': mode == 'already',
                           'baselineId': 'target' if mode == 'already' else version,
                           'changedFiles': 0 if mode == 'already' else total,
                           'changes': [] if mode == 'already' else [{'action':'update'}] * updates + [{'action':'create'}] * creates,
                           'skillSha256':'fixture-skill', 'manifestSha256':'fixture-manifest'}
                if mode == 'bad-plan':
                    payload['changedFiles'] = 999
                if step == '--apply':
                    payload['backup'] = 'fixture-backup'
                if step == '--verify':
                    self.assertEqual(cmd[-2:], ['--backup', 'fixture-backup'])
                    payload.update(installedFilesVerified=41, changedFilesVerified=total,
                                   runtimeAndBindingPreserved=True, researchFilesUnchanged=True, legacyEntrypointRetired=True)
                    if mode == 'bad-verify':
                        payload['installedFilesVerified'] = 40
                return subprocess.CompletedProcess(cmd, 0, json.dumps(payload).encode(), b'')

            output = temp / 'results'
            with patch.object(sys,'argv',['run-update.py','--out',str(output)]), \
                 patch.object(module.sys,'platform','linux'), \
                 patch.object(module.os,'getuid',return_value=10000,create=True), \
                 patch.object(module.subprocess,'run',side_effect=child), \
                 contextlib.redirect_stdout(io.StringIO()):
                code = module.main()
            receipt = json.loads((output / 'install-receipt.json').read_text(encoding='utf-8'))
            for step in calls:
                for suffix in ('.stdout.json','.stderr.json','.exit'):
                    self.assertTrue((output / (step.removeprefix('--') + suffix)).is_file())
            self.assertFalse(receipt['agentLoaded'])
            self.assertFalse(receipt['currentReadVerified'])
            return code, calls, receipt

    def test_three_baselines_apply_once(self):
        for version in COUNTS:
            with self.subTest(version=version):
                code,calls,receipt = self.run_case(version=version)
                self.assertEqual((code,calls),(0,['--dry','--apply','--verify']))
                self.assertEqual(receipt['changedFiles'],COUNTS[version][0])

    def test_failed_dry_stops(self):
        code,calls,_=self.run_case('dry-failure')
        self.assertEqual((code,calls),(1,['--dry']))

    def test_bad_plan_stops(self):
        code,calls,_=self.run_case('bad-plan')
        self.assertEqual((code,calls),(1,['--dry']))

    def test_failed_apply_not_repeated(self):
        code,calls,_=self.run_case('apply-failure')
        self.assertEqual((code,calls),(1,['--dry','--apply']))

    def test_failed_verify_not_reinstalled(self):
        code,calls,receipt=self.run_case('verify-failure')
        self.assertEqual((code,calls),(1,['--dry','--apply','--verify']))
        self.assertFalse(receipt['filesInstalled'])

    def test_bad_verify_not_success(self):
        code,_,receipt=self.run_case('bad-verify')
        self.assertEqual(code,1)
        self.assertFalse(receipt['ok'])

    def test_already_installed_does_not_apply(self):
        code,calls,receipt=self.run_case('already')
        self.assertEqual((code,calls),(0,['--dry']))
        self.assertTrue(receipt['alreadyApplied'])


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