q792602257
725bf4cd9c
- driver/registry.py never actually defined register_driver_type, so the plugin-system's 'driver plugin registered successfully' scenario was unreachable in production (only simulated via test monkeypatch). Added a real implementation wired into the existing driver factory registry. - PluginRegistry.discover() only caught PluginValidationError/ DuplicatePluginError, so a driver-kind manifest that failed wiring (DriverRegistryUnavailableError/PluginTargetResolutionError) aborted the whole scan, silently skipping co-located tool/skill manifests. Now caught and skipped per manifest instead. openspec: plugin-system capability, archived change cloud-runtime
367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""Unit tests for cloud.plugins.PluginRegistry (task 6.9)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.metadata
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from cloud.plugins import (
|
|
DiscoveryResult,
|
|
DriverRegistryUnavailableError,
|
|
DuplicatePluginError,
|
|
PluginManifest,
|
|
PluginRegistry,
|
|
PluginTargetResolutionError,
|
|
PluginValidationError,
|
|
)
|
|
from cloud.store import CloudStore
|
|
import driver.registry as driver_registry_module
|
|
|
|
|
|
def _manifest(
|
|
*,
|
|
name: str = "demo",
|
|
version: str = "1.0.0",
|
|
entry_point_kind: str = "driver",
|
|
target: str = "cloud.store:CloudStore",
|
|
) -> PluginManifest:
|
|
return PluginManifest(
|
|
name=name,
|
|
version=version,
|
|
entry_point_kind=entry_point_kind, # type: ignore[arg-type]
|
|
target=target,
|
|
)
|
|
|
|
|
|
def _fake_driver_factory_builder(connection_info: dict) -> object:
|
|
"""A real ``DriverFactoryBuilder``-shaped function used as a plugin target
|
|
in tests: takes ``connection_info``, returns a zero-arg factory callable
|
|
that "constructs" a driver (here, just the connection info itself, so
|
|
tests can assert on it without needing a real WDA server)."""
|
|
|
|
return lambda: dict(connection_info)
|
|
|
|
|
|
def test_valid_manifest_registers(tmp_path) -> None:
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
|
|
manifest = registry.register(_manifest(entry_point_kind="tool"))
|
|
|
|
assert manifest.name == "demo"
|
|
stored = registry.store.get_plugin("demo")
|
|
assert stored is not None
|
|
assert stored[0].entry_point_kind == "tool"
|
|
|
|
|
|
def test_unrecognized_entry_point_kind_rejected(tmp_path) -> None:
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
|
|
with pytest.raises(PluginValidationError):
|
|
PluginManifest(
|
|
name="bad",
|
|
version="1.0.0",
|
|
entry_point_kind="strategy", # type: ignore[arg-type]
|
|
target="cloud.store:CloudStore",
|
|
)
|
|
|
|
|
|
def test_duplicate_name_rejected(tmp_path) -> None:
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
registry.register(_manifest(entry_point_kind="tool"))
|
|
|
|
with pytest.raises(DuplicatePluginError):
|
|
registry.register(_manifest(entry_point_kind="tool"))
|
|
|
|
|
|
def test_driver_kind_wires_into_register_driver_type(tmp_path, monkeypatch) -> None:
|
|
"""A driver-kind manifest is wired end-to-end into the real driver registry:
|
|
after registration, build_driver_factory() can construct the new type."""
|
|
|
|
monkeypatch.delitem(
|
|
driver_registry_module.SUPPORTED_DRIVER_TYPES, "custom-driver", raising=False
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
registry.register(
|
|
_manifest(
|
|
entry_point_kind="driver",
|
|
name="custom-driver",
|
|
target="tests.test_plugin_registry:_fake_driver_factory_builder",
|
|
)
|
|
)
|
|
|
|
stored = registry.store.get_plugin("custom-driver")
|
|
assert stored is not None
|
|
assert stored[1] is True # wired
|
|
|
|
factory = driver_registry_module.build_driver_factory(
|
|
"custom-driver", {"server_url": "http://example"}
|
|
)
|
|
driver = factory()
|
|
assert driver == {"server_url": "http://example"}
|
|
|
|
|
|
def test_driver_kind_raises_when_extension_point_missing(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
"""When register_driver_type is not importable, registration fails loudly."""
|
|
|
|
# Ensure the attribute is genuinely absent, regardless of future state of driver/registry.py.
|
|
monkeypatch.delattr(
|
|
driver_registry_module,
|
|
"register_driver_type",
|
|
raising=False,
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
|
|
with pytest.raises(DriverRegistryUnavailableError):
|
|
registry.register(_manifest(entry_point_kind="driver"))
|
|
|
|
|
|
def test_tool_and_skill_manifests_register_unwired(tmp_path, monkeypatch) -> None:
|
|
"""tool/skill-kind manifests must not touch any other registry."""
|
|
|
|
touched: list[tuple[str, object]] = []
|
|
|
|
def fail_if_called(name: str, builder) -> None:
|
|
touched.append((name, builder))
|
|
|
|
monkeypatch.setattr(
|
|
driver_registry_module,
|
|
"register_driver_type",
|
|
fail_if_called,
|
|
raising=False,
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
registry.register(_manifest(name="a-tool", entry_point_kind="tool"))
|
|
registry.register(_manifest(name="a-skill", entry_point_kind="skill"))
|
|
|
|
assert touched == []
|
|
|
|
stored_tool = registry.store.get_plugin("a-tool")
|
|
stored_skill = registry.store.get_plugin("a-skill")
|
|
assert stored_tool is not None and stored_tool[1] is False
|
|
assert stored_skill is not None and stored_skill[1] is False
|
|
|
|
|
|
def test_entry_point_discovery_registers_plugin(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
"""discover_entry_points() resolves an installed entry point into a manifest."""
|
|
|
|
fake_ep = SimpleNamespace(
|
|
name="installed-plugin",
|
|
load=lambda: {
|
|
"name": "installed-plugin",
|
|
"version": "0.2.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
importlib.metadata,
|
|
"entry_points",
|
|
lambda **kwargs: [fake_ep],
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
manifests = registry.discover_entry_points()
|
|
|
|
assert [m.name for m in manifests] == ["installed-plugin"]
|
|
registry.register(manifests[0])
|
|
assert registry.store.get_plugin("installed-plugin") is not None
|
|
|
|
|
|
def test_manifest_file_discovery_registers_valid_and_skips_malformed(
|
|
tmp_path,
|
|
) -> None:
|
|
plugins_dir = tmp_path / "plugins"
|
|
(plugins_dir / "good").mkdir(parents=True)
|
|
(plugins_dir / "good" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "good",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(plugins_dir / "bad").mkdir(parents=True)
|
|
(plugins_dir / "bad" / "plugin.json").write_text(
|
|
"{ not valid json",
|
|
encoding="utf-8",
|
|
)
|
|
(plugins_dir / "ugly").mkdir(parents=True)
|
|
(plugins_dir / "ugly" / "plugin.json").write_text(
|
|
json.dumps({"name": "ugly"}), # missing required fields
|
|
encoding="utf-8",
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
manifests = registry.discover_manifest_files(plugins_dir)
|
|
|
|
# Only the well-formed manifest is returned; malformed files are skipped silently.
|
|
assert [m.name for m in manifests] == ["good"]
|
|
|
|
# discover() registers the valid manifest and skips the malformed files
|
|
# without aborting the whole scan.
|
|
result = registry.discover(scan_path=plugins_dir)
|
|
assert any(m.name == "good" for m in result.registered)
|
|
assert isinstance(result, DiscoveryResult)
|
|
|
|
|
|
def test_discover_combines_entry_points_and_manifest_files(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
fake_ep = SimpleNamespace(
|
|
name="via-entry-point",
|
|
load=lambda: PluginManifest(
|
|
name="via-entry-point",
|
|
version="1.0.0",
|
|
entry_point_kind="tool",
|
|
target="cloud.store:CloudStore",
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
importlib.metadata,
|
|
"entry_points",
|
|
lambda **kwargs: [fake_ep],
|
|
)
|
|
|
|
plugins_dir = tmp_path / "plugins"
|
|
(plugins_dir / "via-file").mkdir(parents=True)
|
|
(plugins_dir / "via-file" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "via-file",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "skill",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
result = registry.discover(scan_path=plugins_dir)
|
|
|
|
names = {m.name for m in result.registered}
|
|
assert names == {"via-entry-point", "via-file"}
|
|
assert result.errors == []
|
|
|
|
|
|
def test_discover_does_not_abort_when_driver_wiring_fails(
|
|
tmp_path,
|
|
monkeypatch,
|
|
) -> None:
|
|
"""A driver-kind manifest that fails wiring must not abort the whole
|
|
discover() scan: a co-located tool manifest should still get registered,
|
|
and the failure should surface as a discovery error, not an exception."""
|
|
|
|
# Force the driver-registry extension point to look unavailable so the
|
|
# driver-kind manifest below fails wiring with DriverRegistryUnavailableError.
|
|
monkeypatch.delattr(
|
|
driver_registry_module,
|
|
"register_driver_type",
|
|
raising=False,
|
|
)
|
|
|
|
plugins_dir = tmp_path / "plugins"
|
|
(plugins_dir / "a-broken-driver").mkdir(parents=True)
|
|
(plugins_dir / "a-broken-driver" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "broken-driver",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "driver",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(plugins_dir / "b-ok-tool").mkdir(parents=True)
|
|
(plugins_dir / "b-ok-tool" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "ok-tool",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
result = registry.discover(scan_path=plugins_dir)
|
|
|
|
# The broken driver manifest is recorded as a discovery error, not raised.
|
|
assert any("broken-driver" in err for err in result.errors)
|
|
# The co-located tool manifest still registered despite the driver failure.
|
|
assert any(m.name == "ok-tool" for m in result.registered)
|
|
assert registry.store.get_plugin("ok-tool") is not None
|
|
assert registry.store.get_plugin("broken-driver") is None
|
|
|
|
|
|
def test_discover_reports_target_resolution_failure_without_aborting(
|
|
tmp_path,
|
|
) -> None:
|
|
"""A driver-kind manifest whose target can't resolve to a callable
|
|
(PluginTargetResolutionError) is likewise skipped/logged, not fatal."""
|
|
|
|
plugins_dir = tmp_path / "plugins"
|
|
(plugins_dir / "bad-target-driver").mkdir(parents=True)
|
|
(plugins_dir / "bad-target-driver" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "bad-target-driver",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "driver",
|
|
# os.path.sep exists but is not callable -> PluginTargetResolutionError
|
|
"target": "os.path:sep",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(plugins_dir / "ok-tool-2").mkdir(parents=True)
|
|
(plugins_dir / "ok-tool-2" / "plugin.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "ok-tool-2",
|
|
"version": "1.0.0",
|
|
"entry_point_kind": "tool",
|
|
"target": "cloud.store:CloudStore",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3"))
|
|
|
|
with pytest.raises(PluginTargetResolutionError):
|
|
# Direct register() call still raises loudly for a single manifest...
|
|
registry._wire_driver(
|
|
_manifest(
|
|
name="direct-check",
|
|
entry_point_kind="driver",
|
|
target="os.path:sep",
|
|
)
|
|
)
|
|
|
|
result = registry.discover(scan_path=plugins_dir)
|
|
|
|
assert any("bad-target-driver" in err for err in result.errors)
|
|
assert any(m.name == "ok-tool-2" for m in result.registered)
|
|
assert registry.store.get_plugin("ok-tool-2") is not None
|