3c91e7c314
Fifteen tests failed on this workstation and had been written off as
environmental. Thirteen of them were not.
Those thirteen ran subprocess.run(['bash', ...]). shutil.which('bash') finds
System32\bash.exe - the WSL shim - which cannot execute a Windows path and
failed every call with execvpe(/bin/bash). They now locate Git Bash from
git.exe and probe it with a trivial script before trusting it, so a box with
neither still skips rather than failing.
The other two were genuinely environmental, but were skipping nothing and
failing everything. The symlink test now skips only on WinError 1314, the
absent SeCreateSymbolicLinkPrivilege; any other OSError still fails, because a
symlink that cannot be created for some other reason is a real defect. The
server certificate test skips when openssl is not installed, which is what the
route shells out to.
Being labelled environmental is why these sat for weeks. A test that cannot run
should say so and skip; one that fails should mean something is broken.
322 lines
13 KiB
Python
322 lines
13 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', 'geenforce', 'knowledgebase', 'machines', 'measuringtools', 'network', 'notifications', 'printers', 'slides', 'usb', 'warranty')
|
|
|
|
|
|
@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_baseplugin_has_permissions_hook():
|
|
"""The permissions hook is on the contract surface (contract 0.10.0)."""
|
|
assert hasattr(BasePlugin, 'get_permissions')
|
|
|
|
|
|
@pytest.mark.parametrize('name', BUNDLED_PLUGINS)
|
|
def test_plugin_get_permissions_shape(plugin_instances, name):
|
|
"""get_permissions returns a list of (name, description, category) entries."""
|
|
plugin = plugin_instances[name]
|
|
perms = plugin.get_permissions()
|
|
assert isinstance(perms, list)
|
|
for entry in perms:
|
|
if isinstance(entry, dict):
|
|
pname, category = entry['name'], entry.get('category')
|
|
else:
|
|
assert len(entry) == 3, f'{name}: entry must be a 3-tuple'
|
|
pname, _desc, category = entry
|
|
assert isinstance(pname, str) and '.' in pname, (
|
|
f'{name}: permission name {pname!r} must be dotted')
|
|
assert category, f'{name}: permission {pname} needs a category'
|
|
|
|
|
|
def test_baseplugin_has_frontend_contribution_hooks():
|
|
"""The four ADR-010 frontend-contribution hooks are on the contract (0.7.0)."""
|
|
for hook in ('get_settings_cards', 'get_asset_panels',
|
|
'get_map_overlays', 'get_asset_presentation'):
|
|
assert hasattr(BasePlugin, hook), f'{hook} missing from BasePlugin'
|
|
|
|
|
|
@pytest.mark.parametrize('name', BUNDLED_PLUGINS)
|
|
def test_plugin_frontend_hooks_return_lists(plugin_instances, name):
|
|
"""The four ADR-010 hooks default to a list on every bundled plugin."""
|
|
plugin = plugin_instances[name]
|
|
assert isinstance(plugin.get_settings_cards(), list)
|
|
assert isinstance(plugin.get_asset_panels(), list)
|
|
assert isinstance(plugin.get_map_overlays(), list)
|
|
assert isinstance(plugin.get_asset_presentation(), list)
|
|
|
|
|
|
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'
|
|
try:
|
|
link.symlink_to(external, target_is_directory=True)
|
|
except OSError as error:
|
|
# WinError 1314 - SeCreateSymbolicLinkPrivilege is not held. On Windows
|
|
# that privilege belongs to Administrators, or to any account when
|
|
# Developer Mode is on, so an ordinary developer shell cannot create the
|
|
# fixture this test needs. Skip rather than fail: the scanner behaviour
|
|
# under test is real and is exercised on CI, and a red test here says
|
|
# nothing about the code. Only this specific cause is skipped - any
|
|
# other OSError still fails.
|
|
if getattr(error, 'winerror', None) == 1314:
|
|
pytest.skip('needs symlink privilege (Developer Mode or admin)')
|
|
raise
|
|
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()
|