kangshifu-board/test/test_workshop_report_runner.py

"""Exercise packaged r7 runner sequencing with synthetic child process outputs."""
import contextlib
import importlib.util
import io
import json
import os
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-report-runner-') as temp:
            temp = Path(temp)
            package = temp / 'package'
            shutil.copytree(Path(os.environ.get('KSF_REPORT_PACKAGE_DIR', ROOT / 'docs/workshop-skill-release-20261109-r7/package')), package)
            spec = importlib.util.spec_from_file_location('report_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.assertTrue(kwargs['capture_output'])
                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}')
                if mode == 'apply-failure' and step == '--apply':
                    return subprocess.CompletedProcess(cmd, 1, b'', b'{"ok":false}')
                payload = {'ok': True, 'readOnly': True, 'alreadyApplied': mode == 'already',
                           'baselineId': 'target' if mode == 'already' else 'r6',
                           'changedFiles': 0 if mode == 'already' else 23,
                           'changes': [] if mode == 'already' else [{'action': 'update'}] * 22 + [{'action': 'create'}],
                           'skillSha256': 'fixture-skill', 'manifestSha256': 'fixture-manifest'}
                if mode == 'bad-plan':
                    payload['changedFiles'] = 22
                if step == '--apply':
                    payload['backup'] = 'fixture-backup'
                if step == '--verify':
                    self.assertEqual(cmd[-2:], ['--backup', 'fixture-backup'])
                    payload.update(installedFilesVerified=42 if mode == 'bad-verify' else 43,
                                   changedFilesVerified=23, 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()):
                code = module.main()
            receipt = json.loads((output / 'install-receipt.json').read_text(encoding='utf-8'))
            self.assertFalse(receipt['agentLoaded'])
            for step in calls:
                label = step.removeprefix('--')
                self.assertTrue((output / (label + '.exit')).exists())
                self.assertTrue((output / (label + '.stdout.json')).exists())
                self.assertTrue((output / (label + '.stderr.json')).exists())
            return code, calls, receipt

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

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

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

    def test_wrong_plan_never_applies(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_wrong_verify_count_not_marked_installed(self):
        code, calls, receipt = self.run_case('bad-verify')
        self.assertEqual((code, calls), (1, ['--dry', '--apply', '--verify']))
        self.assertFalse(receipt['filesInstalled'])


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