154 lines
5.3 KiB
Python
154 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import importlib.util
|
|
import sys
|
|
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"
|
|
OFFICIAL_SRC = (
|
|
PROJECT_ROOT
|
|
/ "packages"
|
|
/ "jwebot-official-extensions"
|
|
/ "src"
|
|
)
|
|
sys.path[:0] = [str(SDK_SRC), str(KERNEL_SRC), str(OFFICIAL_SRC)]
|
|
|
|
from jwebot_kernel import parse_manifest, validate_source_tree # noqa: E402
|
|
from jwebot_official_extensions import LOCAL_EXTENSIONS # noqa: E402
|
|
from jwebot_sdk import Event, PlanKind # noqa: E402
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self):
|
|
self.handlers = {}
|
|
|
|
def register(self, command, handler):
|
|
if command in self.handlers:
|
|
raise AssertionError(f"duplicate command: {command}")
|
|
self.handlers[command] = handler
|
|
|
|
|
|
class _Services:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def call(self, profile, *, operation, payload):
|
|
self.calls.append((profile, operation, payload))
|
|
return {"text": f"{profile}:{payload['command']}"}
|
|
|
|
|
|
class _Context:
|
|
def __init__(self):
|
|
self.commands = _Registry()
|
|
self.services = _Services()
|
|
|
|
|
|
class LocalPluginPackageTests(unittest.TestCase):
|
|
def event(self):
|
|
return Event(
|
|
event_id="event-1",
|
|
observed_at="2026-10-14T10:00:00+08:00",
|
|
sender="Alice",
|
|
text="@JWeBot test",
|
|
group_fingerprint="group-a",
|
|
mention_bot=True,
|
|
)
|
|
|
|
def test_every_local_package_validates_registers_and_invokes(self):
|
|
for descriptor in LOCAL_EXTENSIONS:
|
|
with self.subTest(extension=descriptor.extension_id):
|
|
module_name = descriptor.extension_id.replace("-", "_")
|
|
root = (
|
|
PROJECT_ROOT
|
|
/ "extensions"
|
|
/ descriptor.directory
|
|
/ descriptor.extension_id
|
|
)
|
|
manifest = parse_manifest(
|
|
{
|
|
"id": descriptor.extension_id,
|
|
"author": descriptor.author,
|
|
"description": descriptor.description,
|
|
"help": list(descriptor.help_entries),
|
|
"version": "1.0.0",
|
|
"api_version": "1",
|
|
"entrypoint": f"{module_name}:plugin",
|
|
"type": "command",
|
|
"capabilities": {
|
|
"commands": list(descriptor.commands),
|
|
"storage": False,
|
|
"network_profiles": [],
|
|
"service_profiles": [descriptor.extension_id],
|
|
},
|
|
}
|
|
)
|
|
self.assertEqual(manifest.plugin_id, descriptor.extension_id)
|
|
self.assertEqual(manifest.author, descriptor.author)
|
|
self.assertEqual(
|
|
manifest.description,
|
|
descriptor.description,
|
|
)
|
|
self.assertEqual(
|
|
manifest.help_entries,
|
|
descriptor.help_entries,
|
|
)
|
|
self.assertEqual(
|
|
manifest.capabilities.commands,
|
|
descriptor.commands,
|
|
)
|
|
self.assertEqual(
|
|
manifest.capabilities.service_profiles,
|
|
(descriptor.extension_id,),
|
|
)
|
|
validate_source_tree(root, manifest)
|
|
spec = importlib.util.spec_from_file_location(
|
|
f"official_{module_name}",
|
|
root / "src" / f"{module_name}.py",
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
context = _Context()
|
|
module.plugin(context)
|
|
self.assertEqual(
|
|
set(context.commands.handlers),
|
|
set(descriptor.commands),
|
|
)
|
|
first = descriptor.commands[0]
|
|
result = context.commands.handlers[first](
|
|
self.event(),
|
|
"argument",
|
|
)
|
|
self.assertEqual(result.reply.kind, PlanKind.TEXT)
|
|
self.assertEqual(
|
|
context.services.calls[0][0],
|
|
descriptor.extension_id,
|
|
)
|
|
self.assertEqual(
|
|
context.services.calls[0][2]["command"],
|
|
first,
|
|
)
|
|
self.assertEqual(
|
|
context.services.calls[0][2]["arguments"],
|
|
"argument",
|
|
)
|
|
|
|
def test_generated_local_packages_are_in_sync(self):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"sync_local_extensions",
|
|
PROJECT_ROOT / "tools" / "sync-local-extensions.py",
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
self.assertEqual(module.synchronize(check=True), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|