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

"""Observable enterprise boundaries, installation idempotency and credential preservation."""
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'deploy/enterprise-workshop'))
import bootstrap as core
import gogs_setup
import host_bootstrap
import readiness
import registry

spec = importlib.util.spec_from_file_location('enterprise_packager', ROOT / 'scripts/package-enterprise-workshop.py')
packager = importlib.util.module_from_spec(spec)
spec.loader.exec_module(packager)


class EnterpriseTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.payload = packager.content()

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='ksf-enterprise-test-')
        self.addCleanup(self.temp.cleanup)
        root = Path(self.temp.name)
        self.package, self.data = root / 'package', root / 'data'
        for name, raw in self.payload.items():
            p = self.package / name
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_bytes(raw)
        self.current = self.data / '.fmode/config/enterprise/current.json'
        self.cfg = json.loads((self.package / 'enterprise.json').read_bytes())
        self.save_current()

    def save_current(self, **changes):
        value = {k: self.cfg[k] for k in ('companyObjectId', 'appId', 'serverURL', 'functionsURL')}
        value.update(provisionStatus='success', masterKey='PRIVATE-MASTER-DO-NOT-COPY')
        value.update(changes)
        self.current.parent.mkdir(parents=True, exist_ok=True)
        self.current.write_bytes(core.encode(value))
        self.current.chmod(0o600)

    def install(self):
        p = core.plan(self.package, self.data)
        return core.apply(self.package, self.data, p)

    def test_install_and_same_plan_replay_are_idempotent(self):
        unrelated = self.data / 'skills/existing-skill/SKILL.md'
        core.atomic_create(unrelated, b'EXISTING SKILL')
        p = core.plan(self.package, self.data)
        self.assertTrue(p['changes'])
        first = core.apply(self.package, self.data, p)
        second = core.apply(self.package, self.data, p)
        self.assertTrue(first['ok'])
        self.assertFalse(second['applied'])
        self.assertEqual(second['installedFiles'], 0)
        self.assertEqual(unrelated.read_bytes(), b'EXISTING SKILL')
        self.assertFalse((self.data / 'skills/ksf-voc-library/SKILL.md').exists())
        self.assertFalse(first['agentLoaded'])
        self.assertFalse(first['businessAuthorized'])
        self.assertFalse(first['automaticLifecycleHookVerified'])
        self.assertNotIn(b'PRIVATE-MASTER', b''.join(p.read_bytes() for p in
                         (self.data / 'ksf-enterprise-workshop').rglob('*') if p.is_file()))

    def test_existing_private_files_survive_verify(self):
        self.install()
        key = self.data / 'skills/ksf-workshop-runtime/credential-binding.json'
        core.atomic_create(key, b'PRIVATE-HOST-BINDING')
        before = key.read_bytes(), key.stat().st_mtime_ns
        result = core.plan(self.package, self.data)
        self.assertEqual(result['changes'], [])
        core.verify(self.package, self.data)
        self.assertEqual((key.read_bytes(), key.stat().st_mtime_ns), before)

    def test_wrong_tenant_and_underscore_mismatch_stop_before_writes(self):
        for fields in ({'companyObjectId': 'OTHER'}, {'appId': 'FKieRbIoGtmain'}):
            self.save_current(**fields)
            with self.assertRaisesRegex(core.Stopped, 'ENTERPRISE_MISMATCH'):
                core.plan(self.package, self.data)
            self.assertFalse((self.data / 'skills').exists())

    def test_wrong_endpoint_and_unprovisioned_stop(self):
        for fields in ({'functionsURL': 'https://other.fmode.cn/api/functions'}, {'provisionStatus': 'pending'}):
            self.save_current(**fields)
            with self.assertRaises(core.Stopped):
                core.plan(self.package, self.data)
            self.assertFalse((self.data / 'skills').exists())

    def test_payload_tampering_and_unlisted_files_stop(self):
        p = self.package / 'payload/skills/ksf-workshop-runtime/SKILL.md'
        before = p.read_bytes()
        p.write_bytes(b'UNTRUSTED')
        with self.assertRaisesRegex(core.Stopped, 'PACKAGE_HASH_MISMATCH'):
            core.plan(self.package, self.data)
        p.write_bytes(before)
        (self.package / 'unlisted.py').write_bytes(b'UNLISTED')
        with self.assertRaisesRegex(core.Stopped, 'UNDECLARED_PACKAGE_FILE'):
            core.plan(self.package, self.data)

    def test_unknown_existing_component_not_overwritten(self):
        p = self.data / 'skills/ksf-workshop-runtime/SKILL.md'
        core.atomic_create(p, b'LOCAL-UNKNOWN-VERSION')
        with self.assertRaisesRegex(core.Stopped, 'EXISTING_COMPONENT'):
            core.plan(self.package, self.data)
        self.assertEqual(p.read_bytes(), b'LOCAL-UNKNOWN-VERSION')

    def test_package_staged_inside_data_dir_can_install(self):
        staged = self.data / 'ksf-enterprise-packages/package-r1'
        shutil.copytree(self.package, staged)
        p = core.plan(staged, self.data)
        self.assertTrue(core.apply(staged, self.data, p)['ok'])

    def test_changed_plan_is_rejected_without_install(self):
        p = core.plan(self.package, self.data)
        core.atomic_create(self.data / 'skills/ksf-workshop-runtime/other-local-file', b'CONCURRENT')
        with self.assertRaises(core.Stopped):
            core.apply(self.package, self.data, p)
        self.assertFalse((self.data / 'skills/ksf-voc-library').exists())

    def test_research_skill_second_entry_is_rejected(self):
        core.atomic_create(self.data / 'skills/ksf-voc-library/SKILL.md', b'OLD ENTRY')
        with self.assertRaisesRegex(core.Stopped, 'RESEARCH_SKILL_ENTRYPOINT_STILL_ACTIVE'):
            core.plan(self.package, self.data)

    def test_environment_activation_preserves_secret_bytes_and_backup(self):
        env = self.data / '.env'
        original = b'OTHER=keep\r\nPRIVATE_TOKEN="do-not-copy-to-repository"'
        core.atomic_create(env, original)
        self.assertTrue(readiness.activate_research(self.data))
        expected = original + b'\r\nKSF_LIBRARY_CHANNEL=' + str(
            self.data / 'ksf-library-channel/channel.json').encode() + b'\r\n'
        self.assertEqual(env.read_bytes(), expected)
        self.assertFalse(readiness.activate_research(self.data))
        backup = list((self.data / 'ksf-enterprise-workshop').glob('env-before-*'))
        self.assertEqual(len(backup), 1)
        self.assertEqual(backup[0].read_bytes(), original)

    def test_environment_binding_conflict_does_not_change_file(self):
        env = self.data / '.env'
        for raw in (b'KSF_LIBRARY_CHANNEL=/other/private.json\n',
                    b'KSF_LIBRARY_CHANNEL=x\nKSF_LIBRARY_CHANNEL=x\n'):
            env.write_bytes(raw)
            with self.assertRaisesRegex(core.Stopped, 'EXISTING_RESEARCH_BINDING_DIFFERS'):
                readiness.activate_research(self.data)
            self.assertEqual(env.read_bytes(), raw)

    def test_gogs_uses_designated_administrator_before_mutation(self):
        class WrongUser:
            calls = []
            def call(inner, method, route, body=None, missing=False):
                inner.calls.append((method, route))
                return {'username': 'someone-else'}
        api = WrongUser()
        with self.assertRaisesRegex(core.Stopped, 'GOGS_ADMINISTRATOR_MISMATCH'):
            gogs_setup.setup(api, self.package, True)
        self.assertEqual(api.calls, [('GET', '/api/v1/user')])

    def test_existing_organization_owner_mismatch_stops(self):
        class WrongOwner:
            calls = []
            def call(inner, method, route, body=None, missing=False):
                inner.calls.append(method)
                if route.endswith('/user'):
                    return {'username': '18307996893'}
                if route.endswith('/teams'):
                    return [{'id': 1, 'name': 'Owners'}]
                if route.endswith('/members'):
                    return [{'username': 'someone-else'}]
                return {'id': 5}
        api = WrongOwner()
        with self.assertRaisesRegex(core.Stopped, 'EXISTING_ORGANIZATION_OWNER_MISMATCH'):
            gogs_setup.setup(api, self.package, True)
        self.assertEqual(set(api.calls), {'GET'})

    def fake_host(self, current, commit='a' * 40):
        calls = []
        def runner(args, **kwargs):
            calls.append(args)
            if 'rev-parse' in args:
                output = commit.encode()
            elif 'status' in args:
                output = b''
            elif 'id' in args:
                output = b'10000\n'
            elif 'cat' in args:
                output = core.encode(current)
            else:
                raise AssertionError('Unexpected mutating command: ' + repr(args))
            return subprocess.CompletedProcess(args, 0, output, b'')
        return calls, runner

    def test_host_checks_enterprise_before_docker_copy(self):
        current = json.loads(self.current.read_bytes())
        current['companyObjectId'] = 'OTHER'
        calls, runner = self.fake_host(current)
        with self.assertRaisesRegex(core.Stopped, 'ENTERPRISE_MISMATCH'):
            host_bootstrap.initialize(self.package, 'a' * 40, 'agent-node-kangshifu', True, runner)
        self.assertFalse(any('cp' in x for x in calls))

    def test_host_does_not_install_changed_checkout(self):
        calls, runner = self.fake_host(json.loads(self.current.read_bytes()), 'b' * 40)
        with self.assertRaisesRegex(core.Stopped, 'CONFIGURATION_COMMIT_MISMATCH'):
            host_bootstrap.initialize(self.package, 'a' * 40, 'agent-node-kangshifu', True, runner)
        self.assertFalse(any(x[0] == 'docker' for x in calls))

    def test_unmapped_enterprises_are_not_changed(self):
        file = self.data / 'registry.json'
        core.atomic_create(file, core.encode({'schema': 'ksf.enterprise-registry.v1', 'companies': {}}))
        calls, runner = self.fake_host({'companyObjectId': 'OTHER'})
        result = registry.after_twin(file, 'other-agent', True, runner)
        self.assertTrue(result['skipped'])
        self.assertEqual(len(calls), 1)

    def test_registry_binding_preserves_other_enterprises_and_is_idempotent(self):
        file = self.data / 'enterprise-registry.json'
        other = {'repositoryURL': 'https://git.fmode.cn/other/config.git', 'commit': 'b' * 40}
        original = core.encode({'schema': 'ksf.enterprise-registry.v1', 'companies': {'OTHER': other}})
        core.atomic_create(file, original)
        def runner(args, **kwargs):
            if 'rev-parse' in args:
                raw = b'a' * 40
            elif 'remote' in args:
                raw = b'https://git.fmode.cn/kangshifu/workshop-config.git'
            elif 'status' in args:
                raw = b''
            else:
                raise AssertionError(args)
            return subprocess.CompletedProcess(args, 0, raw, b'')
        self.assertTrue(registry.register(file, self.package, 'a' * 40, True, runner)['mappingChanged'])
        value = json.loads(file.read_bytes())
        self.assertEqual(value['companies']['OTHER'], other)
        self.assertEqual(value['companies']['FKieRbIoGt']['commit'], 'a' * 40)
        self.assertFalse(registry.register(file, self.package, 'a' * 40, True, runner)['mappingChanged'])
        backups = list(self.data.glob('enterprise-registry.json.before-*'))
        self.assertEqual(len(backups), 1)
        self.assertEqual(backups[0].read_bytes(), original)

    def test_registry_rejects_other_repository_without_writing(self):
        file = self.data / 'enterprise-registry.json'
        def runner(args, **kwargs):
            raw = b'a' * 40 if 'rev-parse' in args else b'https://git.fmode.cn/unrelated/config.git'
            return subprocess.CompletedProcess(args, 0, raw, b'')
        with self.assertRaisesRegex(core.Stopped, 'CONFIGURATION_REPOSITORY_MISMATCH'):
            registry.register(file, self.package, 'a' * 40, True, runner)
        self.assertFalse(file.exists())

    def test_research_runtime_missing_channel_fails_without_snapshot_fallback(self):
        script = self.package / 'payload/skills/ksf-voc-library/scripts/query_voc.py'
        env = {**os.environ, 'KSF_LIBRARY_CHANNEL': str(self.data / 'missing-channel.json')}
        result = subprocess.run([sys.executable, '-B', str(script), 'contract'], env=env,
                                capture_output=True, timeout=20)
        self.assertEqual(result.returncode, 1)
        value = json.loads(result.stderr)
        self.assertFalse(value['ok'])
        self.assertEqual(value['error']['code'], 'CHANNEL_FILE_NOT_FOUND_OR_REDIRECTED')


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