mirror of
https://github.com/Quad4-Software/MeshChatX.git
synced 2026-08-18 09:49:09 -04:00
233 lines
7.9 KiB
Python
233 lines
7.9 KiB
Python
# SPDX-License-Identifier: 0BSD
|
|
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
|
|
|
|
def _make_manager(tmp_path, app=None):
|
|
from meshchatx.src.backend.plugin_manager import PluginManager
|
|
|
|
return PluginManager(str(tmp_path), app=app)
|
|
|
|
|
|
class TestPluginManagerInstall:
|
|
def test_install_bundled_example(self, tmp_path):
|
|
manager = _make_manager(tmp_path)
|
|
manager.install_bundled_examples()
|
|
plugins = manager.list_plugins()
|
|
ids = [plugin["id"] for plugin in plugins]
|
|
assert "com.meshchatx.mcx-bugs" in ids
|
|
assert "com.meshchatx.transport-node-monitor" not in ids
|
|
|
|
def test_enable_disable_plugin(self, tmp_path):
|
|
manager = _make_manager(tmp_path)
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
enabled = manager.enable(plugin_id)
|
|
assert enabled["enabled"] is True
|
|
disabled = manager.disable(plugin_id)
|
|
assert disabled["enabled"] is False
|
|
|
|
def test_storage_roundtrip(self, tmp_path):
|
|
manager = _make_manager(tmp_path)
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
manager.storage_set(plugin_id, "sample_key", json.dumps(["abc123"]))
|
|
value = manager.storage_get(plugin_id, "sample_key")
|
|
assert json.loads(value) == ["abc123"]
|
|
|
|
def test_permission_denied_for_manager_capability(self, tmp_path):
|
|
manager = _make_manager(tmp_path)
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
manager.enable(plugin_id)
|
|
with pytest.raises(PermissionError):
|
|
manager.call_manager(plugin_id, "unknown.capability", {})
|
|
|
|
def test_bug_report_preview_reads_debug_logs(self, tmp_path):
|
|
class FakeLogs:
|
|
def get_logs(self, **_kwargs):
|
|
return [
|
|
{
|
|
"timestamp": 1.0,
|
|
"level": "INFO",
|
|
"module": "meshchat",
|
|
"message": "peer aa" + ("bb" * 15) + " at /home/user1/secret",
|
|
},
|
|
]
|
|
|
|
def get_total_count(self, **_kwargs):
|
|
return 1
|
|
|
|
class FakeDatabase:
|
|
debug_logs = FakeLogs()
|
|
|
|
class FakeApp:
|
|
reticulum = object()
|
|
database = FakeDatabase()
|
|
storage_dir = str(tmp_path)
|
|
current_context = None
|
|
|
|
manager = _make_manager(tmp_path, app=FakeApp())
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
manager.enable(plugin_id)
|
|
preview = manager.call_manager(
|
|
plugin_id,
|
|
"bugReport.preview",
|
|
{"limit": 10},
|
|
)
|
|
assert preview["line_count"] == 1
|
|
assert "/home/user1/secret" not in preview["log_text"]
|
|
assert "[redacted]" in preview["log_text"]
|
|
|
|
def test_rns_link_capabilities_require_manifest_grant(self, tmp_path):
|
|
class FakeLinkManager:
|
|
async def open_link(self, *_args, **_kwargs):
|
|
return object(), False, None
|
|
|
|
def identify(self, *_args, **_kwargs):
|
|
return True, None
|
|
|
|
def close(self, *_args, **_kwargs):
|
|
return True
|
|
|
|
class FakeApp:
|
|
reticulum = object()
|
|
rns_link_manager = FakeLinkManager()
|
|
|
|
manager = _make_manager(tmp_path, app=FakeApp())
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
manager.enable(plugin_id)
|
|
with pytest.raises(PermissionError):
|
|
manager.call_manager(
|
|
plugin_id,
|
|
"rnsLink.open",
|
|
{"destination_hash": "aa" * 16, "aspect": "microrn.mgmt"},
|
|
)
|
|
|
|
record = manager._plugins[plugin_id]
|
|
record.manifest.setdefault("permissions", {})["managers"] = [
|
|
"destinationPath.read",
|
|
"rnsLink.open",
|
|
"rnsLink.close",
|
|
]
|
|
record.granted_permissions = [
|
|
"managers:destinationPath.read",
|
|
"managers:rnsLink.open",
|
|
"managers:rnsLink.close",
|
|
]
|
|
opened = manager.call_manager(
|
|
plugin_id,
|
|
"rnsLink.open",
|
|
{"destination_hash": "aa" * 16, "aspect": "microrn.mgmt"},
|
|
)
|
|
assert opened["ok"] is True
|
|
closed = manager.call_manager(
|
|
plugin_id,
|
|
"rnsLink.close",
|
|
{"destination_hash": "aa" * 16, "aspect": "microrn.mgmt"},
|
|
)
|
|
assert closed["ok"] is True
|
|
|
|
def test_rns_link_event_hook_dispatches(self, tmp_path):
|
|
events = []
|
|
|
|
class FakeApp:
|
|
plugins_enabled = True
|
|
|
|
def websocket_broadcast(self, _message):
|
|
return None
|
|
|
|
manager = _make_manager(tmp_path, app=FakeApp())
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
manager.enable(plugin_id)
|
|
record = manager._plugins[plugin_id]
|
|
record.manifest.setdefault("permissions", {})["hooks"] = [
|
|
"announce.received",
|
|
"rns.link.event",
|
|
]
|
|
record.granted_permissions = [
|
|
"hooks:announce.received",
|
|
"hooks:rns.link.event",
|
|
]
|
|
manager.dispatch_hook = lambda pid, hook, payload: events.append(
|
|
(pid, hook, payload),
|
|
)
|
|
manager.on_rns_link_event(
|
|
{
|
|
"type": "rns.link.event",
|
|
"event": "link_closed",
|
|
"destination_hash": "aa" * 16,
|
|
"aspect": "microrn.mgmt",
|
|
},
|
|
)
|
|
assert events == [
|
|
(
|
|
plugin_id,
|
|
"rns.link.event",
|
|
{
|
|
"event": "link_closed",
|
|
"destination_hash": "aa" * 16,
|
|
"aspect": "microrn.mgmt",
|
|
"payload_b64": None,
|
|
},
|
|
),
|
|
]
|
|
|
|
def test_manifest_validation_rejects_invalid_id(self, tmp_path):
|
|
manager = _make_manager(tmp_path)
|
|
plugin_dir = os.path.join(tmp_path, "bad-plugin")
|
|
os.makedirs(plugin_dir, exist_ok=True)
|
|
with open(
|
|
os.path.join(plugin_dir, "plugin.json"),
|
|
"w",
|
|
encoding="utf-8",
|
|
) as handle:
|
|
json.dump({"id": "bad id", "version": "1.0.0", "apiVersion": 1}, handle)
|
|
with pytest.raises(ValueError):
|
|
manager.install_from_directory(plugin_dir)
|
|
|
|
def test_plugins_disabled_blocks_install_and_enable(self, tmp_path):
|
|
class DisabledApp:
|
|
plugins_enabled = False
|
|
|
|
manager = _make_manager(tmp_path, app=DisabledApp())
|
|
manager.install_bundled_examples()
|
|
assert manager.list_plugins() == []
|
|
source = os.path.join(
|
|
os.path.dirname(__file__),
|
|
"../../meshchatx/src/backend/data/plugins/mcx-bugs",
|
|
)
|
|
with pytest.raises(PermissionError):
|
|
manager.install_from_directory(os.path.abspath(source))
|
|
|
|
def test_bundled_reinstall_skips_unchanged_and_handles_readonly_tree(
|
|
self,
|
|
tmp_path,
|
|
):
|
|
manager = _make_manager(tmp_path)
|
|
manager.install_bundled_examples()
|
|
plugin_id = "com.meshchatx.mcx-bugs"
|
|
record = manager._plugins[plugin_id]
|
|
first_hash = record.integrity_hash
|
|
install_path = record.install_path
|
|
|
|
for dirpath, dirnames, filenames in os.walk(install_path):
|
|
os.chmod(dirpath, 0o555)
|
|
for name in dirnames:
|
|
os.chmod(os.path.join(dirpath, name), 0o555)
|
|
for name in filenames:
|
|
os.chmod(os.path.join(dirpath, name), 0o444)
|
|
|
|
manager.install_bundled_examples()
|
|
assert manager._plugins[plugin_id].integrity_hash == first_hash
|
|
|
|
record.version = "0.0.0-test"
|
|
manager.install_bundled_examples()
|
|
assert manager._plugins[plugin_id].version != "0.0.0-test"
|
|
assert os.path.isdir(manager._plugins[plugin_id].install_path)
|