81178b7be3
Path.rglob does not descend symlinks, so a symlinked external plugin (the ADR-003 dev loop) silently escaped the contract-purity scan. The scanner now resolves plugin dirs before walking, a regression test plants a symlinked plugin with a real violation and asserts it is flagged, and the known-limitation notes in the external-repo docs are lifted. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""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_baseplugin_has_reports_hook():
|
|
"""The reports hook is on the contract surface (contract 0.6.0)."""
|
|
assert hasattr(BasePlugin, 'get_reports')
|
|
|
|
|
|
@pytest.mark.parametrize('name', BUNDLED_PLUGINS)
|
|
def test_plugin_get_reports_is_iterable(plugin_instances, name):
|
|
"""get_reports returns a list (default empty)."""
|
|
plugin = plugin_instances[name]
|
|
reports = plugin.get_reports()
|
|
assert isinstance(reports, list)
|
|
# Each card carries an id/name/category plus exactly one of route/endpoint.
|
|
for card in reports:
|
|
assert card.get('id') and card.get('name') and card.get('category')
|
|
assert ('route' in card) ^ ('endpoint' in card), (
|
|
f'{name} report card must have exactly one of route/endpoint'
|
|
)
|
|
|
|
|
|
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'
|
|
# Skip migrations/ - Alembic env.py is framework glue that legitimately
|
|
# calls the shared runner in shopdb.plugins.alembic_template (see ADR-008),
|
|
# not plugin domain code. Mirrors the style check excluding versions/.
|
|
# Resolve each plugin dir before rglob: rglob does not descend symlinked
|
|
# directories, and external plugins arrive as symlinks per ADR-003.
|
|
files = [p for p in root.glob('*.py')]
|
|
for plugindir in sorted(root.iterdir()):
|
|
if not plugindir.is_dir():
|
|
continue
|
|
for path in plugindir.resolve().rglob('*.py'):
|
|
if '__pycache__' in path.parts or 'migrations' in path.parts:
|
|
continue
|
|
files.append(path)
|
|
return files
|
|
|
|
|
|
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)
|
|
)
|
|
|
|
|
|
def test_import_scan_covers_symlinked_plugins(tmp_path):
|
|
"""External plugins symlinked into plugins/ ARE scanned for violations.
|
|
|
|
Path.rglob does not descend symlinked directories, so a naive scan would
|
|
silently skip external plugins installed per ADR-003 (clone or symlink
|
|
into plugins/<name>/). The scanner resolves each plugin dir first; this
|
|
test plants a symlinked plugin with a violating import and asserts the
|
|
scan both sees the file and flags the violation.
|
|
"""
|
|
root = Path(__file__).resolve().parent.parent / 'plugins'
|
|
external = tmp_path / 'symlinkdemo'
|
|
external.mkdir()
|
|
# violating import: internal core path, not the contract surface
|
|
(external / 'plugin.py').write_text('from shopdb.extensions import db\n')
|
|
|
|
link = root / 'zzsymlinkdemo'
|
|
link.symlink_to(external, target_is_directory=True)
|
|
try:
|
|
scanned = {str(p) for p in _plugin_source_files()}
|
|
assert any('symlinkdemo' in s for s in scanned), (
|
|
'symlinked plugin sources were not scanned'
|
|
)
|
|
with pytest.raises(AssertionError, match='shopdb.extensions'):
|
|
test_plugins_only_import_contract_surface()
|
|
finally:
|
|
link.unlink()
|