"""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)