"""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, 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 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 calls driver.registry.register_driver_type.""" calls: list[tuple[str, object]] = [] def fake_register(name: str, builder) -> None: calls.append((name, builder)) # driver/registry.py does not yet expose register_driver_type in the real # codebase, so injecting it via monkeypatch simulates the future state # where it does (per design.md D6 / Open Questions). monkeypatch.setattr( driver_registry_module, "register_driver_type", fake_register, raising=False, ) registry = PluginRegistry(CloudStore(tmp_path / "cloud.sqlite3")) registry.register(_manifest(entry_point_kind="driver", name="custom-driver")) assert len(calls) == 1 registered_name, registered_builder = calls[0] assert registered_name == "custom-driver" assert callable(registered_builder) stored = registry.store.get_plugin("custom-driver") assert stored is not None assert stored[1] is True # wired 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 == []