"""Plugin lifecycle + registry tests. Covers the framework's own core feature - the plugin manager's enable/disable guards and permission seeding, plus the registry's persistence and the equipment->machines rename migration. These paths had no direct coverage; a dropped dependency check or a broken rename migration would let an operator disable a required plugin (breaking boot) with no failing test. """ import json import pytest from shopdb.plugins import PluginManager from shopdb.plugins.loader import PluginLoader from shopdb.plugins.registry import PluginRegistry # --- Registry persistence + rename migration -------------------------------- def test_registry_disable_survives_reload(tmp_path): """A disabled plugin stays disabled when the registry is re-read.""" path = tmp_path / 'plugins.json' reg = PluginRegistry(path) reg.register('demo', '1.0.0', enabled=True) reg.disable('demo') reloaded = PluginRegistry(path) assert reloaded.is_installed('demo') assert not reloaded.is_enabled('demo') def test_registry_recovers_from_corrupt_file(tmp_path): """An unparseable plugins.json loads as empty instead of crashing.""" path = tmp_path / 'plugins.json' path.write_text('{ this is not valid json') reg = PluginRegistry(path) assert reg.get_all() == {} def test_registry_migrates_equipment_to_machines(tmp_path): """A legacy 'equipment' entry is carried over to 'machines' on load (the real plugins/machines dir exists), preserving enabled state, once.""" path = tmp_path / 'plugins.json' seed = PluginRegistry(path) seed.register('equipment', '1.0.0', enabled=True) assert seed.is_installed('equipment') migrated = PluginRegistry(path) # _load runs the rename migration assert migrated.is_installed('machines') assert migrated.is_enabled('machines') assert not migrated.is_installed('equipment') # Persisted: a third read sees machines, and never re-migrates. again = PluginRegistry(path) assert again.is_installed('machines') # --- Manager enable/disable dependency guards + permission seeding ----------- def _write_plugin(plugins_dir, name, deps=None, permissions=None, tier=None): """Write a minimal valid synthetic plugin (manifest.json + plugin.py).""" pdir = plugins_dir / name pdir.mkdir() manifest = { 'name': name, 'version': '1.0.0', 'description': f'synthetic {name}', 'dependencies': deps or [], 'core_version': '>=0.1.0,<1.0.0', 'api_prefix': f'/api/{name}', } if tier: manifest['tier'] = tier (pdir / 'manifest.json').write_text(json.dumps(manifest)) (pdir / '__init__.py').write_text('') (pdir / 'plugin.py').write_text( 'from shopdb.plugins.base import BasePlugin, PluginMeta\n\n' 'class ThePlugin(BasePlugin):\n' ' @property\n' ' def meta(self):\n' f' return PluginMeta(name={name!r}, version="1.0.0",\n' f' description="synthetic", dependencies={deps or []!r})\n' ' def get_blueprint(self):\n' ' return None\n' ' def get_models(self):\n' ' return []\n' ' def get_permissions(self):\n' f' return {permissions or []!r}\n' ) @pytest.fixture def manager(app, db, tmp_path): """A PluginManager wired to a temp plugins dir + registry, with synthetic plugins 'alpha' (owns a permission) and 'beta' (depends on alpha).""" plugins_dir = tmp_path / 'plugins' plugins_dir.mkdir() _write_plugin(plugins_dir, 'alpha', permissions=[('alpha.view', 'View alpha', 'alpha')]) _write_plugin(plugins_dir, 'beta', deps=['alpha']) registry = PluginRegistry(tmp_path / 'plugins.json') registry.register('alpha', '1.0.0', enabled=False) registry.register('beta', '1.0.0', enabled=False) pluginmanager = PluginManager() pluginmanager._app = app pluginmanager._db = db pluginmanager.registry = registry pluginmanager.loader = PluginLoader(plugins_dir, registry) pluginmanager.migration_manager = None return pluginmanager def test_enable_blocked_until_dependency_enabled(manager): """beta depends on alpha: enabling beta first is refused.""" assert manager.enable_plugin('beta') is False assert not manager.registry.is_enabled('beta') assert manager.enable_plugin('alpha') is True assert manager.enable_plugin('beta') is True def test_disable_blocked_while_dependent_enabled(manager): """alpha cannot be disabled while beta (which depends on it) is enabled.""" manager.enable_plugin('alpha') manager.enable_plugin('beta') assert manager.disable_plugin('alpha') is False assert manager.registry.is_enabled('alpha') # Disable the dependent first, then alpha frees up. assert manager.disable_plugin('beta') is True assert manager.disable_plugin('alpha') is True def test_enable_seeds_plugin_permissions(manager, app, db): """Enabling a plugin idempotently seeds its declared RBAC permissions.""" from shopdb.core.models import Permission with app.app_context(): assert Permission.query.filter_by(name='alpha.view').first() is None assert manager.enable_plugin('alpha') is True with app.app_context(): assert Permission.query.filter_by(name='alpha.view').first() is not None # Re-enable is a no-op (already enabled) and never duplicates the row. manager.enable_plugin('alpha') with app.app_context(): assert Permission.query.filter_by(name='alpha.view').count() == 1 # --- ADR-013 Phase 0: reverse-deps from disk, cycles, upgrade-all, profiles -- def test_disable_blocked_by_unloaded_dependent(manager): """alpha cannot be disabled while an enabled dependent exists, even when that dependent was never loaded as an instance (manifest read from disk).""" # Mark both enabled directly in the registry WITHOUT loading beta. manager.registry.enable('alpha') manager.registry.enable('beta') assert manager.loader.get_loaded_plugin('beta') is None # The old loaded-instance-only check would wrongly allow this. assert manager.disable_plugin('alpha') is False assert manager.registry.is_enabled('alpha') def test_uninstall_blocked_by_disabled_dependent(manager): """An installed-but-DISABLED dependent still blocks uninstalling its dependency (uninstall reads all installed manifests, not enabled instances).""" assert manager.registry.is_installed('beta') assert not manager.registry.is_enabled('beta') assert manager.uninstall_plugin('alpha') is False assert manager.registry.is_installed('alpha') def test_dependency_cycle_detected(tmp_path): """A -> B -> A raises instead of looping or silently dropping a plugin.""" from shopdb.exceptions import PluginDependencyError plugins_dir = tmp_path / 'plugins' plugins_dir.mkdir() _write_plugin(plugins_dir, 'aa', deps=['bb']) _write_plugin(plugins_dir, 'bb', deps=['aa']) loader = PluginLoader(plugins_dir, PluginRegistry(tmp_path / 'plugins.json')) with pytest.raises(PluginDependencyError): loader._sort_by_dependencies(['aa', 'bb']) def test_upgrade_all_only_touches_registered(manager, tmp_path): """upgrade-all migrates only ADOPTED plugins; an unregistered folder on disk is never migrated as a side effect.""" _write_plugin(tmp_path / 'plugins', 'gamma') # on disk, not registered class StubMigrationManager: def has_pending_migrations(self, name): return False manager.migration_manager = StubMigrationManager() results = manager.upgrade_all_plugins() assert set(results) == {'alpha', 'beta'} assert 'gamma' not in results def test_apply_profile_enables_dependency_closure(manager): """Naming only the dependent enables it AND its dependency, in order.""" result = manager.apply_profile(['beta']) assert manager.registry.is_enabled('alpha') assert manager.registry.is_enabled('beta') assert 'alpha' in result['enabled'] assert 'beta' in result['enabled'] def test_apply_profile_installs_and_enables_fresh_plugin(manager, tmp_path): """A plugin on disk but not yet registered is installed AND enabled.""" _write_plugin(tmp_path / 'plugins', 'gamma') class StubMigrationManager: def has_pending_migrations(self, name): return False def run_plugin_migrations(self, name, revision='head'): return True def has_unapplied_migrations(self, name): return False manager.migration_manager = StubMigrationManager() result = manager.apply_profile(['gamma']) assert manager.registry.is_installed('gamma') assert manager.registry.is_enabled('gamma') assert 'gamma' in result['installed'] def test_apply_profile_rejects_unknown_plugin(manager): """A profile naming a plugin that is not on disk fails loudly.""" from shopdb.exceptions import PluginError with pytest.raises(PluginError): manager.apply_profile(['doesnotexist']) def test_manifest_schema_check(): """The no-dependency manifest schema check flags missing/mistyped fields and accepts a valid manifest (with unknown extra fields).""" from shopdb.plugins.cli import _check_against_schema, _load_manifest_schema schema = _load_manifest_schema() good = {'name': 'x', 'version': '1.0.0', 'description': 'd', 'dependencies': [], 'tier': 'optional', 'somefuture': 42} assert _check_against_schema(good, schema) == [] bad = {'name': 'x', 'version': '1.0.0', 'description': 'd', 'dependencies': 'notalist', 'tier': 'bogus', 'default_enabled': 'yes'} errors = _check_against_schema(bad, schema) assert any('dependencies' in e for e in errors) assert any('tier' in e for e in errors) assert any('default_enabled' in e for e in errors) assert any('version' in e for e in _check_against_schema({'name': 'x'}, schema)) def test_all_bundled_manifests_match_schema(app): """Every bundled plugin's manifest passes the schema - the schema must describe reality, not an idealized shape.""" from shopdb.plugins.cli import _check_against_schema, _load_manifest_schema schema = _load_manifest_schema() pm = app.extensions['plugin_manager'] for name in pm.loader.discover_plugins(): manifest = pm.loader.load_manifest(name) assert _check_against_schema(manifest, schema) == [], name def test_validate_bundled_plugin_passes(app): """`flask plugin validate ` exits clean on a real plugin.""" runner = app.test_cli_runner() result = runner.invoke(args=['plugin', 'validate', 'printers']) assert result.exit_code in (0, None), result.output assert 'valid' in result.output def test_core_tier_plugin_cannot_be_disabled_or_uninstalled(app, db, tmp_path): """A tier=core plugin refuses disable and uninstall (ADR-013).""" plugins_dir = tmp_path / 'plugins' plugins_dir.mkdir() _write_plugin(plugins_dir, 'kernel', tier='core') registry = PluginRegistry(tmp_path / 'plugins.json') registry.register('kernel', '1.0.0', enabled=True) pm = PluginManager() pm._app = app pm._db = db pm.registry = registry pm.loader = PluginLoader(plugins_dir, registry) pm.migration_manager = None assert pm.disable_plugin('kernel') is False assert pm.registry.is_enabled('kernel') assert pm.uninstall_plugin('kernel') is False assert pm.registry.is_installed('kernel') # --- Seed idempotency ------------------------------------------------------- def test_seed_settings_is_idempotent(app, db): """A second `flask seed settings` pass creates zero net-new rows.""" from shopdb.core.models import Setting runner = app.test_cli_runner() assert runner.invoke(args=['seed', 'settings']).exit_code in (0, None) with app.app_context(): first = Setting.query.count() assert runner.invoke(args=['seed', 'settings']).exit_code in (0, None) with app.app_context(): second = Setting.query.count() assert second == first