304 lines
9.9 KiB
Python
304 lines
9.9 KiB
Python
#!/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()
|