"""Plugin contract compliance tests. Asserts every bundled plugin satisfies the contract surface declared in ADR-001 and the BasePlugin ABC. Run on every change to BasePlugin or any plugin's plugin.py / manifest.json. """ import json import re from pathlib import Path import pytest from flask import Blueprint from shopdb import __contract_version__ from shopdb.plugins import plugin_manager from shopdb.plugins.base import BasePlugin, PluginMeta BUNDLED_PLUGINS = ('computers', 'employees', 'equipment', 'knowledgebase', 'network', 'notifications', 'printers', 'slides', 'usb') @pytest.fixture def plugin_classes(app): """Load each bundled plugin's class without instantiating in app context.""" classes = {} with app.app_context(): loader = plugin_manager.loader for name in BUNDLED_PLUGINS: cls = loader.load_plugin_class(name) assert cls is not None, f'Plugin class for {name} could not be loaded' classes[name] = cls return classes @pytest.fixture def plugin_instances(app, plugin_classes): """Instantiate each bundled plugin in app context.""" return {name: cls() for name, cls in plugin_classes.items()} def test_contract_version_is_set(): """Framework declares a contract version.""" assert __contract_version__ parts = __contract_version__.split('.') assert len(parts) == 3, f'Expected semver X.Y.Z, got {__contract_version__}' for part in parts: assert part.isdigit(), f'Non-numeric semver part in {__contract_version__}' @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_class_subclasses_baseplugin(plugin_classes, name): """Each plugin class extends BasePlugin.""" cls = plugin_classes[name] assert issubclass(cls, BasePlugin), f'{name} does not subclass BasePlugin' @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_manifest_has_required_fields(name): """Each plugin's manifest.json declares the required fields.""" manifestpath = Path('plugins') / name / 'manifest.json' assert manifestpath.exists(), f'{name} is missing manifest.json' manifest = json.loads(manifestpath.read_text()) assert manifest.get('name') == name assert manifest.get('version'), f'{name} manifest missing version' assert manifest.get('description'), f'{name} manifest missing description' assert manifest.get('core_version'), f'{name} manifest missing core_version' @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_meta_returns_pluginmeta(plugin_instances, name): """plugin.meta returns a PluginMeta instance with sane fields.""" plugin = plugin_instances[name] meta = plugin.meta assert isinstance(meta, PluginMeta) assert meta.name == name assert meta.version assert meta.api_prefix and meta.api_prefix.startswith('/api/') @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_get_blueprint_returns_blueprint_or_none(plugin_instances, name): """get_blueprint returns a Flask Blueprint or None.""" plugin = plugin_instances[name] bp = plugin.get_blueprint() assert bp is None or isinstance(bp, Blueprint), ( f'{name}.get_blueprint returned {type(bp).__name__}' ) @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_get_models_returns_list(plugin_instances, name): """get_models returns a list of model classes.""" plugin = plugin_instances[name] models = plugin.get_models() assert isinstance(models, list), f'{name}.get_models did not return a list' for model in models: assert hasattr(model, '__tablename__'), ( f'{name}.get_models returned {model} without __tablename__' ) @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_get_collector_schema_returns_dict_or_none(plugin_instances, name): """get_collector_schema returns None or a dict with identityfield + fields.""" plugin = plugin_instances[name] schema = plugin.get_collector_schema() if schema is None: return assert isinstance(schema, dict) assert 'identityfield' in schema, ( f'{name}.get_collector_schema must declare identityfield' ) assert 'fields' in schema, ( f'{name}.get_collector_schema must declare fields' ) @pytest.mark.parametrize('name', BUNDLED_PLUGINS) def test_plugin_get_navigation_items_is_iterable(plugin_instances, name): """get_navigation_items returns an iterable (default empty list).""" plugin = plugin_instances[name] items = plugin.get_navigation_items() assert isinstance(items, list) def test_baseplugin_has_no_searchable_fields_hook(): """get_searchable_fields was removed: global search is a core concern over the asset model, and no plugin ever implemented the hook (contract 0.4.0).""" assert not hasattr(BasePlugin, 'get_searchable_fields') def test_baseplugin_has_dashboard_widgets_hook(): """The dashboard widgets hook is on the contract surface and consumed.""" assert hasattr(BasePlugin, 'get_dashboard_widgets') def test_get_services_hook_has_consumer(app): """get_services is consumed by plugin_manager.get_service (no dead hook).""" with app.app_context(): pm = app.extensions['plugin_manager'] assert hasattr(pm, 'get_service') # Unknown service name resolves to None, not an error. assert pm.get_service('definitely-not-a-real-service') is None def test_baseplugin_does_not_have_event_handlers_hook(): """The removed get_event_handlers hook must not be on BasePlugin.""" assert not hasattr(BasePlugin, 'get_event_handlers'), ( 'get_event_handlers was removed for v1. See ADR-001 contract surface.' ) def test_baseplugin_has_collector_schema_hook(): """The collector schema hook is on the contract surface.""" assert hasattr(BasePlugin, 'get_collector_schema') def test_baseplugin_has_apply_collector_payload_hook(): """The collector upsert hook is on the contract surface (ADR-006).""" assert hasattr(BasePlugin, 'apply_collector_payload') def test_schema_declaring_plugins_implement_apply(plugin_instances): """Any plugin returning a collector schema must implement the upsert hook.""" for name, plugin in plugin_instances.items(): if plugin.get_collector_schema() is not None: overridden = type(plugin).apply_collector_payload \ is not BasePlugin.apply_collector_payload assert overridden, ( f'Plugin {name} declares a collector schema but does not ' f'override apply_collector_payload' ) # Imports a plugin may make from the core. shopdb.api is the contract surface; # shopdb.plugins.base is the plugin ABC. Anything else (shopdb.core.*, # shopdb.extensions, shopdb.utils.*) is a contract violation per ADR-001. ALLOWED_CORE_IMPORTS = ('shopdb.api', 'shopdb.plugins.base') _PLUGIN_IMPORT_RE = re.compile( r'^\s*(?:from (shopdb[\w.]*) import|import (shopdb[\w.]*))', re.MULTILINE ) def _plugin_source_files(): root = Path(__file__).resolve().parent.parent / 'plugins' return [p for p in root.rglob('*.py') if '__pycache__' not in p.parts] def test_plugins_only_import_contract_surface(): """Plugins must import core code only via shopdb.api / shopdb.plugins.base.""" violations = [] for path in _plugin_source_files(): text = path.read_text() for match in _PLUGIN_IMPORT_RE.finditer(text): module = match.group(1) or match.group(2) if not module.startswith('shopdb'): continue if any(module == a or module.startswith(a + '.') for a in ALLOWED_CORE_IMPORTS): continue line = text[:match.start()].count('\n') + 1 violations.append(f'{path.name}:{line} imports {module}') assert not violations, ( 'Plugins must import core only via shopdb.api or shopdb.plugins.base. ' 'Violations:\n' + '\n'.join(violations) )