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

"""Offline registration handler tests, including a real Python to Node pipe."""
import contextvars
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import types
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parent.parent
spec = importlib.util.spec_from_file_location('workshop_registration', ROOT / 'integrations/wecom-identity/workshop_registration.py')
reg = importlib.util.module_from_spec(spec); spec.loader.exec_module(reg)
V3 = Path(os.environ.get('KSF_ADAPTER_REVIEW_ROOT') or ROOT / 'work/adapter-v4-review-20261102/adapter-patch-v4')
sys.path.insert(0, str(V3))
import aibot_identity as ai
bridge_spec = importlib.util.spec_from_file_location('identity_bridge', ROOT / 'integrations/wecom-identity/identity_bridge.py')
br = importlib.util.module_from_spec(bridge_spec); bridge_spec.loader.exec_module(br)
# PYTHONPATH=/opt/hermes can bind the bridge to the qualified production module,
# while the baseline import above creates a different, byte-identical module.
# Patch the module actually referenced by read_context, not its baseline twin.
bound_identity_module = sys.modules.get(br.read_context.__module__)
if bound_identity_module is None or br.read_context is not getattr(bound_identity_module, 'read_context', None):
    raise RuntimeError('TEST_IDENTITY_MODULE_BINDING_UNAVAILABLE')
ai = bound_identity_module


class RegistrationTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='ksf-registration-'); self.addCleanup(self.temp.cleanup)
        self.base = Path(self.temp.name)
        self.node = shutil.which('node')
        self.identity = ai.build_context(bot_id='synthetic-bot', sender_id='synthetic-sender', chat_type='single', chat_id=None, message_id='msg1')
        module = (ROOT / 'skills/ksf-workshop-runtime/scripts/ksf-identity.mjs').as_uri()
        bridge = (ROOT / 'skills/ksf-workshop-runtime/scripts/ksf-identity-bridge.mjs').as_uri()
        opts = {'profileDir': str(self.base / 'agents'), 'participantDir': str(self.base / 'participants')}
        agent = {'name': '王刚001', 'role': 'workshop', 'identityBasis': 'runtime_config', 'runtimePlatform': 'hermes', 'channel': 'wecom', 'account': {'platform': 'wecom', 'accountId': self.identity['botId']}}
        code = f'import {{registerAgentProfile}} from {json.dumps(module)}; console.log(JSON.stringify(registerAgentProfile({json.dumps(agent, ensure_ascii=False)},{json.dumps(opts)}).profile));'
        p = subprocess.run([self.node, '--input-type=module', '-e', code], capture_output=True, text=True, encoding='utf-8', check=True)
        self.agent = json.loads(p.stdout)
        self.config = {'version': 'ksf.current-employee-registration.v1', 'agentProfileKey': self.agent['key'], 'botIdSha256': reg.hashlib.sha256(self.identity['botId'].encode()).hexdigest()}
        self.worker = self.base / 'test-worker.mjs'
        self.worker.write_text(f'import fs from "node:fs"; import {{registerBridgeIdentity}} from {json.dumps(bridge)}; console.log(JSON.stringify(registerBridgeIdentity(JSON.parse(fs.readFileSync(0,"utf8")),{json.dumps(opts)})));', encoding='utf-8')

    def decode_tool(self, value):
        self.assertIsInstance(value, str, 'Hermes only accepts string or multimodal tool results')
        if os.environ.get('KSF_REQUIRE_REAL_TOOL_REGISTRY') == '1':
            import tools.registry as tool_registry
            candidates = [getattr(cls, '_normalize_handler_result', None) or getattr(cls, 'normalize_handler_result', None)
                          for cls in vars(tool_registry).values() if isinstance(cls, type)]
            candidates = list(dict.fromkeys(fn for fn in candidates if callable(fn)))
            self.assertEqual(len(candidates), 1, 'Find the actual Hermes result normalizer; no fixture fallback')
            value = candidates[0](reg.TOOL_NAME, value)
            self.assertIsInstance(value, str)
        return json.loads(value)

    def run_tool(self, checked=None):
        with patch.object(reg, '_current_identity', return_value=checked or {'ok': True, 'identity': self.identity}), patch.object(reg, '_read_maintenance_config', return_value=self.config), patch.object(reg, 'WORKER', self.worker):
            return self.decode_tool(reg.register_current_employee({}))

    def test_real_pipe_reuses_profile_and_keeps_ids_out_of_receipt(self):
        a = self.run_tool(); b = self.run_tool()
        self.assertTrue(a['ok'], a); self.assertTrue(a['changed']); self.assertFalse(b['changed'])
        self.assertEqual(a['profile']['key'], b['profile']['key'])
        self.assertNotIn('synthetic-', json.dumps(a))
        rows = list((self.base / 'participants').glob('*.json')); self.assertEqual(len(rows), 1)
        self.assertEqual(json.loads(rows[0].read_text(encoding='utf-8'))['account']['accountId'], self.identity['senderId'])

    def test_no_arguments_even_equal_identity(self):
        with patch.object(reg, '_current_identity') as read:
            for args in ({'messageId': 'msg1'}, [], 'text'):
                self.assertEqual(self.decode_tool(reg.register_current_employee(args))['error'], 'NO_ARGUMENTS_ALLOWED')
            read.assert_not_called()

    def test_failed_context_never_spawns(self):
        with patch.object(reg.subprocess, 'run') as spawn:
            self.assertFalse(self.run_tool({'ok': False, 'error': 'synthetic-private-error'})['ok']); spawn.assert_not_called()

    def test_bot_mismatch_never_spawns(self):
        self.config['botIdSha256'] = '0' * 64
        with patch.object(reg.subprocess, 'run') as spawn:
            self.assertEqual(self.run_tool()['error'], 'REGISTRATION_INSTANCE_MISMATCH'); spawn.assert_not_called()

    def test_worker_failure_does_not_leak_output(self):
        bad = types.SimpleNamespace(returncode=1, stdout='synthetic-private', stderr='synthetic-private')
        with patch.object(reg.subprocess, 'run', return_value=bad):
            r = self.run_tool(); self.assertEqual(r['error'], 'REGISTRATION_WORKER_FAILED'); self.assertNotIn('synthetic-', json.dumps(r))

    def test_private_maintainer_config_is_required(self):
        location = self.base / 'plugins/wecom-identity'; location.mkdir(parents=True)
        file = location / 'registration.json'; file.write_text(json.dumps(self.config), encoding='utf-8'); file.chmod(0o600)
        with patch.dict(os.environ, {'HERMES_HOME': str(self.base)}):
            self.assertEqual(reg._read_maintenance_config(), self.config)
            file.write_text('{"version":"unknown"}', encoding='utf-8')
            with self.assertRaises(reg.RegistrationError): reg._read_maintenance_config()

    def test_v3_context_to_handler_to_real_node_profile(self):
        sc = types.ModuleType('gateway.session_context'); sc._UNSET = object()
        values = {'PLATFORM': 'wecom', 'MESSAGE_ID': 'msg1', 'USER_ID': self.identity['senderId'], 'CHAT_TYPE': 'dm', 'CHAT_ID': self.identity['senderId']}
        sc._VAR_MAP = {'HERMES_SESSION_' + k: contextvars.ContextVar(k, default=sc._UNSET) for k in values}
        for k, v in values.items(): sc._VAR_MAP['HERMES_SESSION_' + k].set(v)
        packages = {}
        for n in ('gateway', 'plugins', 'plugins.platforms', 'plugins.platforms.wecom'):
            packages[n] = types.ModuleType(n); packages[n].__path__ = []
        packages.update({'gateway.session_context': sc, 'plugins.platforms.wecom.identity_bridge': br})
        with patch.dict(sys.modules, packages), patch.object(br, 'read_instance_registry', return_value={'botId': self.identity['botId']}), patch.object(ai, '_read_raw', return_value=self.identity), patch.object(reg, '_read_maintenance_config', return_value=self.config), patch.object(reg, 'WORKER', self.worker):
            r = self.decode_tool(reg.register_current_employee({})); self.assertTrue(r['ok'], r)
            sc._VAR_MAP['HERMES_SESSION_MESSAGE_ID'].set(sc._UNSET)
            with patch.object(reg.subprocess, 'run') as spawn:
                self.assertFalse(self.decode_tool(reg.register_current_employee({}))['ok']); spawn.assert_not_called()

    def test_registered_handlers_return_text_and_keep_internal_query_structured(self):
        packages = {}
        for name in ('plugins', 'plugins.platforms', 'plugins.platforms.wecom'):
            packages[name] = types.ModuleType(name); packages[name].__path__ = []
        packages.update({'plugins.platforms.wecom.identity_bridge': br,
                         'plugins.platforms.wecom.workshop_registration': reg})
        entry = types.ModuleType('combined_entry')
        with patch.dict(sys.modules, packages):
            exec((ROOT / 'integrations/wecom-identity/__init__.py').read_text(encoding='utf-8'), entry.__dict__)
        handlers = {}
        ctx = types.SimpleNamespace(register_tool=lambda name, _set, _schema, handler, **_kw: handlers.update({name: handler}))
        entry.register(ctx)
        self.assertEqual(set(handlers), {'wecom_identity_current', 'ksf_register_current_employee'})
        checked = {'ok': True, 'identity': self.identity}
        with patch.object(br, 'current_identity', return_value=checked):
            self.assertEqual(self.decode_tool(handlers[br.TOOL_NAME]({})), checked)
            self.assertIsInstance(br.current_identity(), dict)
        with patch.object(reg, '_current_identity', return_value=checked), patch.object(reg, '_read_maintenance_config', return_value=self.config), patch.object(reg, 'WORKER', self.worker):
            first = self.decode_tool(handlers[reg.TOOL_NAME]({}))
            second = self.decode_tool(handlers[reg.TOOL_NAME]({}))
        self.assertTrue(first['changed']); self.assertFalse(second['changed'])
        self.assertEqual(first['profile']['key'], second['profile']['key'])
        self.assertIn('员工待绑定', first['profile']['name'])
        with patch.object(br, 'current_identity', return_value={'ok': False, 'error': 'IDENTITY_CONTEXT_UNAVAILABLE'}):
            self.assertFalse(self.decode_tool(handlers[br.TOOL_NAME]({}))['ok'])
        with patch.object(br, 'current_identity') as read:
            self.assertEqual(self.decode_tool(handlers[br.TOOL_NAME]({'messageId': 'spoof'}))['error'], 'NO_ARGUMENTS_ALLOWED')
            read.assert_not_called()

    def test_every_registration_error_path_is_serialized_and_redacted(self):
        with patch.object(reg, '_current_identity', side_effect=RuntimeError('private-config-path-and-secret')):
            value = reg.register_current_employee({})
            self.assertEqual(self.decode_tool(value)['error'], 'IDENTITY_REGISTRATION_FAILED')
            self.assertNotIn('private-config', value)
        with patch.object(reg, '_current_identity', return_value={'ok': True, 'identity': self.identity}), patch.object(reg, '_read_maintenance_config', side_effect=reg.RegistrationError('REGISTRATION_CONFIG_UNSAFE')):
            self.assertEqual(self.decode_tool(reg.register_current_employee({}))['error'], 'REGISTRATION_CONFIG_UNSAFE')
        malformed = types.SimpleNamespace(returncode=0, stdout='not-json-private', stderr='private')
        with patch.object(reg.subprocess, 'run', return_value=malformed):
            self.assertEqual(self.run_tool()['error'], 'IDENTITY_REGISTRATION_FAILED')
        invalid = types.SimpleNamespace(returncode=0, stdout='{}', stderr='private')
        with patch.object(reg.subprocess, 'run', return_value=invalid):
            self.assertEqual(self.run_tool()['error'], 'REGISTRATION_RECEIPT_INVALID')


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