kangshifu-board/test/test_workshop_research_runner.py

"""Test runner sequencing with synthetic child outputs; real mutation is tested 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]


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

            def child(cmd, **kwargs):
                self.assertIsInstance(cmd, list)
                self.assertEqual(kwargs['capture_output'], True)
                self.assertNotIn('shell', kwargs)
                step = cmd[2]
                calls.append(step)
                if mode == 'dry-failure' and step == '--dry':
                    return subprocess.CompletedProcess(cmd, 1, b'', b'{"ok":false,"error":{"code":"UNKNOWN_TARGET_SKILL"}}')
                if mode == 'apply-failure' and step == '--apply':
                    return subprocess.CompletedProcess(cmd, 1, b'', b'{"ok":false,"backup":"fixture-backup","recoveryStatus":"FAILED_RESTORED"}')
                payload = {'ok': True, 'readOnly': True, 'alreadyApplied': mode == 'already',
                           'baselineId': 'r3' if mode == 'r3' else 'target' if mode == 'already' else 'r2',
                           'changedFiles': 0 if mode == 'already' else 12 if mode == 'r3' else 14,
                           'changes': [] if mode == 'already' else [{'action': 'update'}] * (11 if mode == 'r3' else 12) + [{'action': 'create'}] * (1 if mode == 'r3' else 2),
                           'skillSha256': 'fixture-skill', 'manifestSha256': 'fixture-manifest'}
                if mode == 'bad-plan':
                    payload['changedFiles'] = 10
                if step == '--apply':
                    payload['backup'] = 'fixture-backup'
                if step == '--verify':
                    self.assertEqual(cmd[-2:], ['--backup', 'fixture-backup'])
                    payload.update(installedFilesVerified=40, changedFilesVerified=12 if mode == 'r3' else 14,
                                   runtimeAndBindingPreserved=True, researchFilesUnchanged=True,
                                   legacyEntrypointRetired=True)
                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()):
                exit_code = module.main()
            receipt = json.loads((output / 'install-receipt.json').read_text(encoding='utf-8'))
            for step in calls:
                label = step.removeprefix('--')
                self.assertTrue((output / (label + '.exit')).exists())
                self.assertTrue((output / (label + '.stderr.json')).exists())
            self.assertFalse(receipt['agentLoaded'])
            self.assertFalse(receipt['currentReadVerified'])
            return exit_code, calls, receipt

    def test_success_runs_dry_apply_once_verify_in_order(self):
        code, calls, receipt = self.run_case('success')
        self.assertEqual((code, calls), (0, ['--dry', '--apply', '--verify']))
        self.assertTrue(receipt['filesInstalled'])

    def test_r3_success_runs_once(self):
        code, calls, receipt = self.run_case('r3')
        self.assertEqual((code, calls), (0, ['--dry', '--apply', '--verify']))
        self.assertEqual(receipt['changedFiles'], 12)

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

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

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

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


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