#!/usr/bin/env python3 import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock 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" OFFICIAL_SRC = ( PROJECT_ROOT / "packages" / "jwebot-official-extensions" / "src" ) try: import tomli TOML_EXTRA = [str(Path(tomli.__file__).resolve().parent.parent)] except ModuleNotFoundError: TOML_EXTRA = [] sys.path[:0] = [ str(SDK_SRC), str(OFFICIAL_SRC), str(KERNEL_SRC), ] from jwebot_kernel import ( # noqa: E402 ExtensionHostClient, build_release, install_release, sign_release_record, ) from jwebot_official_extensions import LOCAL_EXTENSIONS # noqa: E402 from tests.test_jwebot_edge import MODULE, config, event # noqa: E402 PUBLISHER = "test-publisher" @unittest.skipUnless( Path("/usr/bin/ssh-keygen").is_file(), "/usr/bin/ssh-keygen is unavailable", ) class HostedCommandParityTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.workspace = tempfile.TemporaryDirectory() cls.root = Path(cls.workspace.name) cls.private_key = cls.root / "publisher" result = subprocess.run( [ "/usr/bin/ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(cls.private_key), ], capture_output=True, check=False, ) if result.returncode != 0: raise RuntimeError("cannot generate test publisher key") cls.allowed_signers = cls.root / "allowed_signers" cls.allowed_signers.write_text( f"{PUBLISHER} " + cls.private_key.with_suffix(".pub").read_text( encoding="utf-8" ), encoding="utf-8", ) cls.wrapper = cls.root / "host-wrapper.py" cls.wrapper.write_text( "\n".join( ( "import runpy, sys", "sys.path[:0] = [", f" {str(SDK_SRC)!r},", f" {str(KERNEL_SRC)!r},", f" {str(HOST_SRC)!r},", *( f" {path!r}," for path in TOML_EXTRA ), "]", "runpy.run_module('jwebot_extension_host', run_name='__main__')", ) ) + "\n", encoding="utf-8", ) cls.state_root = cls.root / "extensions" health_client = cls.client() for descriptor in LOCAL_EXTENSIONS: source = ( PROJECT_ROOT / "extensions" / descriptor.directory / descriptor.extension_id ) artifact, record_path, _ = build_release( source, cls.root / "dist" / descriptor.extension_id, ) signature_path, _ = sign_release_record( record_path, cls.private_key, PUBLISHER, ) install_release( artifact, record_path, cls.state_root / descriptor.extension_id, signature=signature_path, allowed_signers=cls.allowed_signers, expected_publisher=PUBLISHER, health_check=lambda release, client=health_client: client.invoke( release, {"op": "health"}, ), ) @classmethod def tearDownClass(cls): cls.workspace.cleanup() @classmethod def client(cls): return ExtensionHostClient( command=(sys.executable, str(cls.wrapper)), timeout_seconds=5, allowed_signers=cls.allowed_signers, expected_publisher=PUBLISHER, ) @classmethod def factory(cls, plugin_id, release_dir): del plugin_id client = cls.client() return lambda request, service: client.invoke( release_dir, request, service_handler=service, ) def setUp(self): self.hosted_temp = tempfile.TemporaryDirectory() self.compat_temp = tempfile.TemporaryDirectory() self.hosted = MODULE.EdgeApplication( config( Path(self.hosted_temp.name), extension_mode="host-only", extension_transport="process", extension_state_root=self.state_root, extension_allowed_signers=self.allowed_signers, extension_publisher=PUBLISHER, ), host_invoker_factory=self.factory, ) self.compat = MODULE.EdgeApplication( config( Path(self.compat_temp.name), local_extensions_enabled=False, ) ) def tearDown(self): self.hosted.close() self.compat.close() self.hosted_temp.cleanup() self.compat_temp.cleanup() def assert_command_equal( self, event_id, text, *, sender="Alice", ): incoming = event(event_id, text, sender=sender) hosted = self.hosted.accept(dict(incoming)) compat = self.compat.accept(dict(incoming)) self.assertEqual(hosted["route"], compat["route"]) self.assertEqual(hosted["text"], compat["text"]) return hosted def test_all_plugin_families_match_direct_compatibility_path(self): stateless = ( ("dice", "@JWeBot /丢骰子5d20"), ("coin", "@JWeBot /翻硬币"), ("random", "@JWeBot /随机数 -10 100"), ("lot", "@JWeBot /抽签"), ("fortune", "@JWeBot /今日运势"), ("memory-help", "@JWeBot /记忆帮助"), ("scan-empty", "@JWeBot /扫码排行"), ("repeat-empty", "@JWeBot /复读榜"), ("history-empty", "@JWeBot /查记录 测试"), ) for event_id, text in stateless: with self.subTest(command=text): self.assert_command_equal(event_id, text) memory = ( ("remember", "@JWeBot /记住 我喜欢家庭服务器"), ("my-memory", "@JWeBot /我的记忆"), ("person-memory", "@JWeBot /人物记忆 Alice"), ("forget-memory", "@JWeBot /忘记 全部"), ) for event_id, text in memory: with self.subTest(command=text): self.assert_command_equal(event_id, text) challenge = ( ("challenge-start", "@JWeBot /丢骰子挑战 开始 5d20", "Alice"), ("challenge-alice", "@JWeBot /丢骰子挑战", "Alice"), ("challenge-gb", "@JWeBot /丢骰子挑战", "Bob"), ("challenge-state", "@JWeBot /丢骰子挑战 状态", "Alice"), ("challenge-settle", "@JWeBot /丢骰子挑战 结算", "Alice"), ) for event_id, text, sender in challenge: with self.subTest(command=text): self.assert_command_equal(event_id, text, sender=sender) self.assert_command_equal( "guess-start", "@JWeBot /猜数字 开始 50", ) target = self.hosted.ledger.local_game( "group-a", "guess_number", )["target"] self.assertEqual( target, self.compat.ledger.local_game( "group-a", "guess_number", )["target"], ) self.assert_command_equal("guess-status", "@JWeBot /猜数字 状态") self.assert_command_equal("guess-win", f"@JWeBot /猜 {target}") rps = ( ("rps-start", "@JWeBot /石头剪刀布 开始", "Alice"), ("rps-join", "@JWeBot /石头剪刀布 加入", "Bob"), ("rps-first", "@JWeBot /石头剪刀布 出 石头", "Alice"), ("rps-second", "@JWeBot /石头剪刀布 出 剪刀", "Bob"), ("rps-end", "@JWeBot /石头剪刀布 结束", "Alice"), ) for event_id, text, sender in rps: with self.subTest(command=text): self.assert_command_equal(event_id, text, sender=sender) for application in (self.hosted, self.compat): application.local.steam = mock.Mock() application.local.steam.render.return_value = ( "Steam：yr 正在玩 Dota 2" ) self.assert_command_equal("steam", "@JWeBot /steam") hosted_row = self.hosted.ledger.row("steam") compat_row = self.compat.ledger.row("steam") self.assertEqual( self.hosted.generate_steam_reply(hosted_row), self.compat.generate_steam_reply(compat_row), ) self.assert_command_equal( "restart-guess", "@JWeBot /猜数字 开始 30", ) expected_target = self.hosted.ledger.local_game( "group-a", "guess_number", )["target"] self.hosted.close() self.hosted = MODULE.EdgeApplication( config( Path(self.hosted_temp.name), extension_mode="host-only", extension_transport="process", extension_state_root=self.state_root, extension_allowed_signers=self.allowed_signers, extension_publisher=PUBLISHER, ), host_invoker_factory=self.factory, ) restored = self.hosted.ledger.local_game( "group-a", "guess_number", ) self.assertEqual(restored["target"], expected_target) self.assertEqual(restored["upper"], 30) if __name__ == "__main__": unittest.main()