#!/usr/bin/env python3 import sys import tempfile import unittest from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] SDK_SRC = PROJECT_ROOT / "packages" / "jwebot-sdk" / "src" KERNEL_SRC = PROJECT_ROOT / "packages" / "jwebot-kernel" / "src" HOST_SRC = PROJECT_ROOT / "packages" / "jwebot-extension-host" / "src" sys.path[:0] = [str(SDK_SRC), str(KERNEL_SRC), str(HOST_SRC)] from jwebot_extension_host import ( # noqa: E402 ExtensionExecutionError, ExtensionRuntime, HostError, process_request, ) from jwebot_kernel import build_release, install_release, parse_manifest # noqa: E402 from jwebot_sdk import Event, PlanKind # noqa: E402 def manifest(commands=None, service_profiles=None): return parse_manifest( { "id": "hello-command", "author": "tester", "description": "Say hello", "help": ["/hello [name]"], "version": "1.0.0", "api_version": "1", "entrypoint": "hello_command:plugin", "type": "command", "capabilities": { "commands": commands or ["/hello"], "storage": False, "network_profiles": [], "service_profiles": service_profiles or [], }, } ) def normal_source() -> str: return "\n".join( ( "from jwebot_sdk import ExtensionResult, Fact, ReplyPlan", "", "def handle(event, arguments):", " return ExtensionResult(", " reply=ReplyPlan.text(f'hello {event.sender} {arguments}'.strip()),", " facts=(Fact(kind='topic', text='used hello command'),),", " )", "", "def plugin(context):", " context.commands.register('/hello', handle)", ) ) class ExtensionHostTests(unittest.TestCase): def runtime( self, source: str, *, plugin_manifest=None, allow_unsigned=True, service_broker=None, ): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) base = Path(temporary.name) extension = base / "extension" extension.mkdir() (extension / "src").mkdir() (extension / "plugin.toml").write_text( 'id = "hello-command"\n', encoding="utf-8", ) (extension / "src" / "hello_command.py").write_text( source, encoding="utf-8", ) plugin_manifest = plugin_manifest or manifest() artifact, record_path, _ = build_release( extension, base / "dist", manifest=plugin_manifest, ) release_dir = install_release( artifact, record_path, base / "state", allow_unsigned=True, ) return ExtensionRuntime.load( release_dir, manifest=plugin_manifest, allow_unsigned=allow_unsigned, service_broker=service_broker, ) def event(self): return Event( event_id="event-1", observed_at="2026-10-14T10:00:00+08:00", sender="Alice", text="@JWeBot /hello", group_fingerprint="group-a", mention_bot=True, ) def test_verified_extension_returns_structured_result(self): runtime = self.runtime(normal_source()) result = runtime.invoke_command("/hello", self.event(), "world") self.assertEqual(result.reply.kind, PlanKind.TEXT) self.assertEqual(result.reply.message, "hello Alice world") self.assertEqual(result.facts[0].kind, "topic") self.assertEqual(list(runtime.release_dir.rglob("*.pyc")), []) def test_host_rejects_unsigned_release_by_default(self): with self.assertRaisesRegex(HostError, "signed extension release"): self.runtime(normal_source(), allow_unsigned=False) def test_request_protocol_serializes_sdk_result(self): runtime = self.runtime(normal_source()) response = process_request( runtime, { "op": "invoke_command", "command": "/hello", "arguments": "", "event": self.event().to_mapping(), }, ) self.assertEqual(response["status"], "ok") self.assertEqual(response["plugin_id"], "hello-command") self.assertEqual( response["result"]["reply"]["text"], "hello Alice", ) def test_health_reports_frozen_manifest_inventory(self): runtime = self.runtime( normal_source(), plugin_manifest=manifest(service_profiles=["hello-service"]), ) response = process_request(runtime, {"op": "health"}) self.assertEqual(response["status"], "ok") self.assertEqual(response["plugin_id"], "hello-command") self.assertEqual(response["description"], "Say hello") self.assertEqual(response["help"], ["/hello [name]"]) self.assertEqual(response["commands"], ["/hello"]) self.assertEqual(response["service_profiles"], ["hello-service"]) def test_extension_cannot_register_undeclared_command(self): source = normal_source().replace( "context.commands.register('/hello', handle)", "context.commands.register('/other', handle)", ) with self.assertRaisesRegex(HostError, "undeclared command"): self.runtime(source) def test_extension_must_register_every_declared_command(self): with self.assertRaisesRegex(HostError, "did not register"): self.runtime(normal_source(), plugin_manifest=manifest(["/hello", "/other"])) def test_invalid_handler_result_fails_closed(self): source = normal_source().replace( "return ExtensionResult(", "return 'invalid' #", 1, ) # Replace the remaining multi-line expression with a valid unreachable value. source = source.replace( " reply=ReplyPlan.text(f'hello {event.sender} {arguments}'.strip()),\n" " facts=(Fact(kind='topic', text='used hello command'),),\n" " )", "", ) runtime = self.runtime(source) with self.assertRaisesRegex(ExtensionExecutionError, "invalid result"): runtime.invoke_command("/hello", self.event(), "") def test_unconfigured_http_capability_fails_closed(self): source = "\n".join( ( "from jwebot_sdk import ExtensionResult, ReplyPlan", "", "def plugin(context):", " def handle(event, arguments):", " context.http.request('missing', method='GET', path='/')", " return ExtensionResult(reply=ReplyPlan.no_reply())", " context.commands.register('/hello', handle)", ) ) runtime = self.runtime(source) with self.assertRaisesRegex(ExtensionExecutionError, "command failed"): runtime.invoke_command("/hello", self.event(), "") def test_unconfigured_service_capability_fails_closed(self): source = "\n".join( ( "from jwebot_sdk import ExtensionResult, ReplyPlan", "", "def plugin(context):", " def handle(event, arguments):", " context.services.call(", " 'missing', operation='execute', payload={}", " )", " return ExtensionResult(reply=ReplyPlan.no_reply())", " context.commands.register('/hello', handle)", ) ) runtime = self.runtime(source) with self.assertRaisesRegex(ExtensionExecutionError, "command failed"): runtime.invoke_command("/hello", self.event(), "") def test_declared_service_profile_uses_kernel_broker(self): source = "\n".join( ( "from jwebot_sdk import ExtensionResult, ReplyPlan", "", "def plugin(context):", " def handle(event, arguments):", " response = context.services.call(", " 'hello-service',", " operation='execute',", " payload={'sender': event.sender, 'value': arguments},", " )", " return ExtensionResult(", " reply=ReplyPlan.text(response['text'])", " )", " context.commands.register('/hello', handle)", ) ) calls = [] def broker(profile, operation, payload): calls.append((profile, operation, dict(payload))) return {"text": "brokered reply"} runtime = self.runtime( source, plugin_manifest=manifest(service_profiles=["hello-service"]), service_broker=broker, ) result = runtime.invoke_command("/hello", self.event(), "value") self.assertEqual(result.reply.message, "brokered reply") self.assertEqual( calls, [ ( "hello-service", "execute", {"sender": "Alice", "value": "value"}, ) ], ) def test_logger_does_not_emit_plugin_text_or_field_names(self): runtime = self.runtime(normal_source()) with self.assertLogs(level="INFO") as captured: runtime.context.logger.info( "private message body", **{"private-field-name": "private value"}, ) output = "\n".join(captured.output) self.assertNotIn("private message body", output) self.assertNotIn("private-field-name", output) self.assertNotIn("private value", output) def test_outgoing_event_is_rejected_by_request_protocol(self): runtime = self.runtime(normal_source()) event = self.event().to_mapping() event["side"] = "outgoing" with self.assertRaisesRegex(HostError, "SDK contract"): process_request( runtime, { "op": "invoke_command", "command": "/hello", "event": event, }, ) if __name__ == "__main__": unittest.main()