kangshifu-board/test/enterprise-voc.test.py

"""VOC bootstrap behavior: verified reuse, missing install, failure isolation and r1 upgrade."""
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import unittest

sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('enterprise_fixtures', ROOT / 'test/enterprise-workshop.test.py')
base = importlib.util.module_from_spec(spec)
spec.loader.exec_module(base)
import bootstrap as core
import gogs_setup
import voc


class VocTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.payload = base.packager.content()
        cls.archive = ROOT / 'work/enterprise-workshop-20261110/voc-0344/voc-skill-0.3.44.tgz'

    setUp = base.EnterpriseTests.setUp
    save_current = base.EnterpriseTests.save_current

    def official_fixture(self, name='voc-intelligence'):
        _, meta = voc.release(self.package)
        target = self.data / 'skills' / name
        for relative, raw in voc.unpack(self.archive.read_bytes(), meta).items():
            core.atomic_create(target / relative, raw, 0o644)
        (target / '.mcp.json').write_bytes(core.encode(voc.fragment(target)))
        return target

    def runner(self, npm_ok=True):
        calls = []
        def run(args, **kwargs):
            calls.append(args)
            if 'ci' in args:
                return subprocess.CompletedProcess(args, 0 if npm_ok else 1, b'', b'')
            self.assertTrue(any(str(x).endswith('check-voc.cjs') for x in args))
            value = {'ok': True, 'version': '0.3.44', 'credentialAvailable': False,
                     'collectionRequests': 0, 'tools': ['voc_api_search','voc_api_doc','voc_api_call']}
            return subprocess.CompletedProcess(args, 0, core.encode(value), b'')
        return calls, run

    def test_missing_voc_is_required_even_with_generic_social_skill(self):
        core.atomic_create(self.data / 'skills/voc-social-suite/SKILL.md', b'GENERIC VOC GUIDE')
        result = voc.plan(self.package, self.data)
        self.assertEqual(result['action'], 'install-missing')
        self.assertTrue(result['target'].endswith('voc-intelligence'))

    def test_existing_official_package_is_reused_without_npm_or_config_changes(self):
        target = self.official_fixture('voc-social-suite')
        config = voc.fragment(target)
        config['mcpServers']['voc']['env']['PRIVATE_EXISTING_VALUE'] = 'local-only'
        (target / '.mcp.json').write_bytes(core.encode(config))
        before = (target / '.mcp.json').read_bytes(), (target / '.mcp.json').stat().st_mtime_ns
        output = self.data / 'results'
        output.mkdir()
        calls, runner = self.runner()
        result = voc.apply(self.package, self.data, voc.plan(self.package, self.data), output, runner=runner)
        self.assertFalse(result['vocModified'])
        self.assertEqual(result['action'], 'verified-existing')
        self.assertEqual(len(calls), 1)
        self.assertEqual(before, ((target / '.mcp.json').read_bytes(), (target / '.mcp.json').stat().st_mtime_ns))
        self.assertFalse(result['runtime']['credentialAvailable'])

    def test_other_version_is_not_silently_downgraded_or_upgraded(self):
        for version in ('0.3.42', '0.3.45'):
            path = self.data / 'skills/voc-intelligence/package.json'
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(core.encode({'name': '@vocmarket/voc-skill', 'version': version}))
            with self.assertRaisesRegex(core.Stopped, 'EXISTING_VOC_VERSION_DIFFERS'):
                voc.plan(self.package, self.data)
            self.assertEqual(json.loads(path.read_bytes())['version'], version)

    def test_modified_public_script_stops_before_install(self):
        target = self.official_fixture()
        script = target / 'mcp/src/server.js'
        script.write_bytes(b'UNKNOWN LOCAL CODE')
        with self.assertRaisesRegex(core.Stopped, 'EXISTING_VOC_FILES_CHANGED'):
            voc.plan(self.package, self.data)
        self.assertEqual(script.read_bytes(), b'UNKNOWN LOCAL CODE')

    def test_tampered_tarball_is_rejected(self):
        _, meta = voc.release(self.package)
        with self.assertRaisesRegex(core.Stopped, 'VOC_ARCHIVE_HASH_MISMATCH'):
            voc.unpack(self.archive.read_bytes() + b'TAMPER', meta)

    def test_missing_install_stages_then_commits_and_replay_does_not_reinstall(self):
        output = self.data / 'results'
        output.mkdir()
        original_plan = voc.plan(self.package, self.data)
        calls, runner = self.runner()
        result = voc.apply(self.package, self.data, original_plan, output, self.archive, runner)
        self.assertTrue(result['vocModified'])
        self.assertEqual(result['publicFilesVerified'], 169)
        again = voc.apply(self.package, self.data, original_plan, output, self.archive, runner)
        self.assertFalse(again['vocModified'])
        self.assertEqual(sum('ci' in call for call in calls), 1)
        self.assertFalse(result['liveAccessVerified'])
        self.assertFalse(result['hostRegistrationVerified'])

    def test_dependency_failure_does_not_leave_an_installed_skill(self):
        output = self.data / 'results'
        output.mkdir()
        calls, runner = self.runner(False)
        with self.assertRaisesRegex(core.Stopped, 'VOC_DEPENDENCY_INSTALL_FAILED'):
            voc.apply(self.package, self.data, voc.plan(self.package, self.data), output, self.archive, runner)
        self.assertFalse((self.data / 'skills/voc-intelligence').exists())

    def test_r1_binding_upgrade_preserves_skill_and_private_identity(self):
        old = ROOT / 'docs/ksf-enterprise-workshop-20261110-r1/package'
        for source in (old / 'payload').rglob('*'):
            if source.is_file():
                core.atomic_create(self.data / source.relative_to(old / 'payload'), source.read_bytes(), 0o644)
        before = core.encode({'schema': 'ksf.enterprise-workshop-binding.v1', 'companyObjectId': 'FKieRbIoGt',
                              'appId': 'FKieRbIoGt_main', 'configurationSha256': core.sha((old / 'enterprise.json').read_bytes()),
                              'packageManifestSha256': core.sha((old / 'FILES.sha256').read_bytes()),
                              'skill': 'ksf-workshop-runtime', 'version': 'r13'})
        binding = self.data / 'ksf-enterprise-workshop/binding.json'
        core.atomic_create(binding, before)
        identity = self.data / 'skills/ksf-workshop-runtime/credential-binding.json'
        core.atomic_create(identity, b'ORIGINAL PRIVATE IDENTITY')
        p = core.plan(self.package, self.data)
        self.assertTrue(p['bindingUpdateRequired'])
        self.assertEqual(p['changes'], [])
        result = core.apply(self.package, self.data, p)
        self.assertEqual(result['installedFiles'], 0)
        self.assertEqual(identity.read_bytes(), b'ORIGINAL PRIVATE IDENTITY')
        backups = list(binding.parent.glob('binding-before-*'))
        self.assertEqual(len(backups), 1)
        self.assertEqual(backups[0].read_bytes(), before)

    def test_only_exact_r1_configuration_repository_can_upgrade(self):
        old = ROOT / 'docs/ksf-enterprise-workshop-20261110-r1/package'
        files = {p.relative_to(old).as_posix(): p.read_bytes() for p in old.rglob('*') if p.is_file()}
        _, candidate = core.sealed(self.package)
        self.assertEqual(gogs_setup.repository_state(files, self.package, candidate), 'known-r1')
        files['README.md'] += b'UNKNOWN LOCAL CHANGE'
        with self.assertRaisesRegex(core.Stopped, 'EXISTING_REPOSITORY_CONTENT_CHANGED'):
            gogs_setup.repository_state(files, self.package, candidate)


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