ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 3 (part 1): manifest-first loader, shopdb.api namespace, auto-register

Browse Source 
Hardens the plugin framework so sister-site adoption is safe.

Loader rewrite (shopdb/plugins/loader.py):
- Reads manifest.json directly. Dependency sort and version checks
  no longer instantiate plugin classes (avoids __init__ side effects).
- Fail-loud policy: in dev/test (DEBUG or TESTING true), plugin
  errors re-raise. In production, errors log with full context and
  the plugin is excluded from registration. Framework keeps booting.
- Contract-version range check via packaging.SpecifierSet. Plugin's
  manifest.core_version must include the framework's
  __contract_version__ or load fails per the policy above.
- Manifest validation: required fields (name, version, description),
  name matches directory, JSON parseable.

Exceptions (shopdb/exceptions.py):
- PluginNotFoundError, PluginContractError, PluginVersionError,
  PluginDependencyError. Specific types replace generic Exception
  swallowing.

Auto-register core blueprints (shopdb/__init__.py):
- CORE_BLUEPRINT_NAMES tuple drives registration. Adding a core
  resource is one entry, not three lines (import + register call).
- Replaces 27 hand-coded register_blueprint calls.
- Asserts each blueprint is exported by shopdb.core.api at boot.

Public API namespace (shopdb/api/__init__.py):
- audit_log: thin wrapper over AuditLog.log() with stable signature.
- resolve_asset_position: implements ADR-001 position resolution
  (asset > related > location). Asset.mapx/mapy and
  AssetRelationship.inheritsposition columns are part of the locked
  contract surface but not yet in models; helper degrades gracefully
  to location-only fallback until the migration lands.

BasePlugin helpers (shopdb/plugins/base.py):
- get_setting(key, default), set_setting(key, value, ...). Settings
  namespaced as plugin.<pluginname>.<key> so two plugins can use the
  same key without colliding.

Manifest version compatibility (plugins/*/manifest.json):
- Bumped core_version from ">=1.0.0" to ">=0.1.0,<1.0.0" so all
  bundled plugins satisfy the new range check.

Contract version bump (shopdb/__init__.py):
- 0.1.0 -> 0.2.0. Additive surface change (Setting helpers,
  shopdb.api namespace) per ADR-002 minor-bump rules.

Tests (tests/test_plugin_loader.py, tests/test_api_namespace.py):
- 13 loader tests: manifest validation failures, version range
  checks, plugin.py import errors, strict-vs-isolate behavior under
  TESTING vs production-like config, manifest-first dependency sort.
- 8 api-namespace tests: audit_log roundtrip, resolve position
  fallback chain, plugin.get_setting/set_setting roundtrip with
  per-plugin namespacing.

Test count: 66 -> 87 passing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-05-08 16:15:28 -04:00
parent 5fefb53bca
commit 6f085a175d
14 changed files with 757 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/PLUGIN-HOOKS.md
View File

@@ -9,7 +9,7 @@ The contract is locked in [ADR-001](../migrations/adr/ADR-001-asset-as-platform-
The framework declares its contract version in `shopdb/__init__.py`:
```python
__contract_version__ = '0.1.0'
__contract_version__ = '0.2.0'
```
Each plugin's `manifest.json` declares the range of contract versions it supports:

2
plugins/computers/manifest.json
View File

@@ -4,7 +4,7 @@
"description": "Computer management plugin for PCs, servers, and workstations with software tracking",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/computers",
"provides": {
"asset_type": "computer",

2
plugins/equipment/manifest.json
View File

@@ -4,7 +4,7 @@
"description": "Equipment management plugin for CNCs, CMMs, lathes, grinders, and other manufacturing equipment",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/equipment",
"provides": {
"asset_type": "equipment",

2
plugins/network/manifest.json
View File

@@ -4,7 +4,7 @@
"description": "Network device management plugin for switches, APs, cameras, and IDFs",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/network",
"provides": {
"asset_type": "network_device",

2
plugins/notifications/manifest.json
View File

@@ -4,7 +4,7 @@
"description": "Notifications and announcements management plugin",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/notifications",
"provides": {
"features": ["notifications", "announcements", "calendar_events"]

2
plugins/printers/manifest.json
View File

@@ -4,7 +4,7 @@
"description": "Printer management plugin with Zabbix integration, supply tracking, and QR codes",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/printers",
"provides": {
"machine_category": "Printer",

2
plugins/usb/manifest.json
View File

@@ -4,6 +4,6 @@
"description": "USB device checkout management",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=1.0.0",
"core_version": ">=0.1.0,<1.0.0",
"api_prefix": "/api/usb"
}

93
shopdb/__init__.py
View File

@@ -12,7 +12,7 @@ from .plugins import plugin_manager
# ADR-002 for the bump rules. Plugins declare a compatible range in
# their manifest.json `core_version` field. Pre-1.0 (0.x) means the
# contract is still settling; sister sites should pin tight ranges.
__contract_version__ = '0.1.0'
__contract_version__ = '0.2.0'
def create_app(config_name: str = None) -> Flask:
@@ -76,57 +76,52 @@ def create_app(config_name: str = None) -> Flask:
return app
CORE_BLUEPRINT_NAMES = (
'auth',
'assets',
'machines',
'machinetypes',
'pctypes',
'statuses',
'vendors',
'models',
'businessunits',
'locations',
'operatingsystems',
'dashboard',
'applications',
'knowledgebase',
'search',
'reports',
'collector',
'employees',
'slides',
'settings',
'auditlogs',
'users',
)
def register_blueprints(app: Flask):
"""Register core API blueprints."""
from .core.api import (
auth_bp,
assets_bp,
machines_bp,
machinetypes_bp,
pctypes_bp,
statuses_bp,
vendors_bp,
models_bp,
businessunits_bp,
locations_bp,
operatingsystems_bp,
dashboard_bp,
applications_bp,
knowledgebase_bp,
search_bp,
reports_bp,
collector_bp,
employees_bp,
slides_bp,
settings_bp,
auditlogs_bp,
users_bp,
)
"""Register core API blueprints from CORE_BLUEPRINT_NAMES.
Each entry maps to an attribute `<name>_bp` exported by
`shopdb.core.api` and a URL prefix `/api/<name>`. Adding a new
core resource is one entry in CORE_BLUEPRINT_NAMES, not a 3-line
edit in this function.
"""
from .core import api as api_module
api_prefix = '/api'
app.register_blueprint(auth_bp, url_prefix=f'{api_prefix}/auth')
app.register_blueprint(assets_bp, url_prefix=f'{api_prefix}/assets')
app.register_blueprint(machines_bp, url_prefix=f'{api_prefix}/machines')
app.register_blueprint(machinetypes_bp, url_prefix=f'{api_prefix}/machinetypes')
app.register_blueprint(pctypes_bp, url_prefix=f'{api_prefix}/pctypes')
app.register_blueprint(statuses_bp, url_prefix=f'{api_prefix}/statuses')
app.register_blueprint(vendors_bp, url_prefix=f'{api_prefix}/vendors')
app.register_blueprint(models_bp, url_prefix=f'{api_prefix}/models')
app.register_blueprint(businessunits_bp, url_prefix=f'{api_prefix}/businessunits')
app.register_blueprint(locations_bp, url_prefix=f'{api_prefix}/locations')
app.register_blueprint(operatingsystems_bp, url_prefix=f'{api_prefix}/operatingsystems')
app.register_blueprint(dashboard_bp, url_prefix=f'{api_prefix}/dashboard')
app.register_blueprint(applications_bp, url_prefix=f'{api_prefix}/applications')
app.register_blueprint(knowledgebase_bp, url_prefix=f'{api_prefix}/knowledgebase')
app.register_blueprint(search_bp, url_prefix=f'{api_prefix}/search')
app.register_blueprint(reports_bp, url_prefix=f'{api_prefix}/reports')
app.register_blueprint(collector_bp, url_prefix=f'{api_prefix}/collector')
app.register_blueprint(employees_bp, url_prefix=f'{api_prefix}/employees')
app.register_blueprint(slides_bp, url_prefix=f'{api_prefix}/slides')
app.register_blueprint(settings_bp, url_prefix=f'{api_prefix}/settings')
app.register_blueprint(auditlogs_bp, url_prefix=f'{api_prefix}/auditlogs')
app.register_blueprint(users_bp, url_prefix=f'{api_prefix}/users')
for name in CORE_BLUEPRINT_NAMES:
attr_name = f'{name}_bp'
if not hasattr(api_module, attr_name):
raise RuntimeError(
f'Core blueprint "{attr_name}" missing from shopdb.core.api. '
f'Either add it or remove "{name}" from CORE_BLUEPRINT_NAMES.'
)
bp = getattr(api_module, attr_name)
app.register_blueprint(bp, url_prefix=f'{api_prefix}/{name}')
def register_cli_commands(app: Flask):

96
shopdb/api/__init__.py Normal file
View File

@@ -0,0 +1,96 @@
"""Public API namespace exposed to plugins.
Plugin authors at sister sites import from this module. The contract is
locked in ADR-001 and versioned per ADR-002. Helpers added here become
part of the platform contract; bumps follow ADR-002 rules.
Currently exposed:
- audit_log: record an audit log entry with consistent schema
- resolve_asset_position: compute an asset's resolved map position
Setting helpers are exposed via BasePlugin instance methods
(plugin.get_setting, plugin.set_setting), not from this namespace.
"""
from typing import Any, Dict, Optional
from shopdb.core.models import AuditLog
def audit_log(
action: str,
entitytype: str,
entityid: int = None,
entityname: str = None,
changes: Dict = None,
details: Dict = None,
) -> AuditLog:
"""Record an audit log entry with the framework's standard schema.
Plugins call this to record state changes on their own assets in a
way that is consistent with core auditing. The function delegates to
AuditLog.log() which already captures the current user, IP address,
and user agent from the Flask request context.
Args:
action: Action verb in past tense ('created', 'updated', 'deleted')
entitytype: Class name of the entity affected ('Computer', 'Printer')
entityid: Primary key of the entity
entityname: Human-readable identifier (hostname, asset number)
changes: Dict with 'before' and 'after' snapshots for updates
details: Arbitrary additional context
Returns:
The created AuditLog instance, already committed to the DB.
"""
return AuditLog.log(
action=action,
entitytype=entitytype,
entityid=entityid,
entityname=entityname,
changes=changes,
details=details,
)
def resolve_asset_position(asset) -> Optional[Dict[str, Any]]:
"""Compute the resolved map position for an asset.
Per ADR-001, position resolution follows this priority:
1. Asset-specific override (asset.mapx, asset.mapy)
2. Walk relationships where inheritsposition is true (partof, then controls)
3. Asset's location coords (asset.location.mapx, .mapy)
4. None (asset is rendered in an unplaced tray)
Returns a dict {'mapx', 'mapy', 'positionsource'} or None if no
position can be resolved.
Note: Asset.mapx/mapy and AssetRelationship.inheritsposition columns
are part of the locked ADR-001 contract surface but have not yet
been added to the models. Until they are, this helper falls back to
the location-only path. The full algorithm activates automatically
once those columns exist.
"""
if hasattr(asset, 'mapx') and hasattr(asset, 'mapy'):
if asset.mapx is not None and asset.mapy is not None:
return {
'mapx': asset.mapx,
'mapy': asset.mapy,
'positionsource': 'self',
}
location = getattr(asset, 'location', None)
if location is not None:
location_mapx = getattr(location, 'mapx', None)
location_mapy = getattr(location, 'mapy', None)
if location_mapx is not None and location_mapy is not None:
return {
'mapx': location_mapx,
'mapy': location_mapy,
'positionsource': 'location',
}
return None
__all__ = ['audit_log', 'resolve_asset_position']

16
shopdb/exceptions.py
View File

@@ -52,3 +52,19 @@ class PluginError(ShopDBException):
def __init__(self, message: str, plugin_name: str = None):
super().__init__(message, 'PLUGIN_ERROR', {'plugin': plugin_name})
class PluginNotFoundError(PluginError):
"""A plugin directory or manifest is missing."""
class PluginContractError(PluginError):
"""A plugin violates the BasePlugin contract (no subclass, missing hooks)."""
class PluginVersionError(PluginError):
"""A plugin's core_version range does not include the framework's __contract_version__."""
class PluginDependencyError(PluginError):
"""A plugin declares a dependency on another plugin that is missing or not enabled."""

30
shopdb/plugins/base.py
View File

@@ -73,6 +73,36 @@ class BasePlugin(ABC):
"""Return dict of service name -> service class."""
return {}
def get_setting(self, key: str, default=None):
"""Read a plugin-scoped setting from the core Setting store.
Settings are namespaced by plugin name to avoid collisions
across plugins. The on-disk key is `plugin.<pluginname>.<key>`.
Returns the typed value (string, integer, boolean, etc.) or
the default if not set.
"""
from shopdb.core.models import Setting
namespaced_key = f'plugin.{self.meta.name}.{key}'
return Setting.get(namespaced_key, default)
def set_setting(self, key: str, value, valuetype: str = 'string',
description: str = None) -> None:
"""Write a plugin-scoped setting to the core Setting store.
Settings are namespaced by plugin name. Persists immediately and
survives restarts.
"""
from shopdb.core.models import Setting
namespaced_key = f'plugin.{self.meta.name}.{key}'
Setting.set(
namespaced_key,
value,
valuetype=valuetype,
category=f'plugin.{self.meta.name}',
description=description,
)
def get_collector_schema(self) -> Optional[Dict]:
"""Return JSON Schema describing the collector payload for this plugin.

230
shopdb/plugins/loader.py
View File

@@ -1,129 +1,232 @@
"""Plugin discovery and loading."""
"""Plugin discovery and loading.
The loader reads each plugin's manifest.json before instantiating the
plugin class. Dependency sorting and contract-version compatibility
checks operate on manifests, not plugin instances. The plugin class is
imported and instantiated only after the manifest passes validation.
Failure policy (per the enforcing-plugin-contract skill):
- In dev/test (app.config.DEBUG or TESTING true): re-raise the original
exception so failures are loud.
- In production: log the failure with full context and exclude the
plugin from registration. The framework keeps booting.
"""
import importlib
import importlib.util
import json
import logging
from pathlib import Path
from typing import Dict, List, Type, Optional
from flask import Flask
import logging
from packaging.specifiers import SpecifierSet, InvalidSpecifier
from packaging.version import Version, InvalidVersion
from .base import BasePlugin
from .registry import PluginRegistry
from ..exceptions import (
PluginError,
PluginNotFoundError,
PluginContractError,
PluginVersionError,
PluginDependencyError,
)
logger = logging.getLogger(__name__)
class PluginLoader:
"""
Discovers and loads plugins from the plugins directory.
"""
"""Discovers and loads plugins from the plugins directory."""
def __init__(self, plugins_dir: Path, registry: PluginRegistry):
self.plugins_dir = plugins_dir
self.registry = registry
self._loaded_plugins: Dict[str, BasePlugin] = {}
self._plugin_classes: Dict[str, Type[BasePlugin]] = {}
self._manifests: Dict[str, dict] = {}
def _is_strict_mode(self, app: Optional[Flask]) -> bool:
"""Return True if loader should re-raise instead of isolating failures."""
if app is None:
return True
return bool(app.config.get('DEBUG') or app.config.get('TESTING'))
def _handle_failure(self, app: Optional[Flask], exc: Exception, name: str) -> None:
"""In strict mode re-raise; in production log with context and continue."""
if self._is_strict_mode(app):
raise exc
logger.exception(f'Plugin {name} failed to load: {exc}')
def discover_plugins(self) -> List[str]:
"""
Discover available plugins in plugins directory.
Returns list of plugin names.
"""
"""Discover available plugins in plugins directory."""
available = []
if not self.plugins_dir.exists():
return available
for item in self.plugins_dir.iterdir():
if item.is_dir() and (item / 'plugin.py').exists():
available.append(item.name)
elif item.is_dir() and (item / '__init__.py').exists():
# Check for plugin.py in package
if not item.is_dir():
continue
if (item / 'plugin.py').exists():
available.append(item.name)
return available
def load_plugin_class(self, name: str) -> Optional[Type[BasePlugin]]:
def load_manifest(self, name: str) -> dict:
"""Load the plugin's manifest.json.
Raises PluginNotFoundError if the manifest does not exist or is
unparseable. The result is cached.
"""
Load plugin class without instantiating.
Used for inspection before installation.
if name in self._manifests:
return self._manifests[name]
manifest_path = self.plugins_dir / name / 'manifest.json'
if not manifest_path.exists():
raise PluginNotFoundError(
f'Plugin {name} has no manifest.json at {manifest_path}',
plugin_name=name,
)
try:
with open(manifest_path) as f:
manifest = json.load(f)
except json.JSONDecodeError as e:
raise PluginContractError(
f'Plugin {name} has invalid manifest.json: {e}',
plugin_name=name,
) from e
for required in ('name', 'version', 'description'):
if not manifest.get(required):
raise PluginContractError(
f'Plugin {name} manifest.json missing required field "{required}"',
plugin_name=name,
)
if manifest['name'] != name:
raise PluginContractError(
f'Plugin directory "{name}" does not match manifest name "{manifest["name"]}"',
plugin_name=name,
)
self._manifests[name] = manifest
return manifest
def check_contract_version(self, name: str, contract_version: str) -> None:
"""Verify the plugin's core_version range includes contract_version.
Raises PluginVersionError on mismatch.
"""
manifest = self.load_manifest(name)
core_version_spec = manifest.get('core_version', '')
if not core_version_spec:
return
try:
specifier = SpecifierSet(core_version_spec)
framework_version = Version(contract_version)
except (InvalidSpecifier, InvalidVersion) as e:
raise PluginContractError(
f'Plugin {name} has invalid version specifier "{core_version_spec}": {e}',
plugin_name=name,
) from e
if framework_version not in specifier:
raise PluginVersionError(
f'Plugin {name} requires core_version {core_version_spec} '
f'but framework is at {contract_version}',
plugin_name=name,
)
def load_plugin_class(self, name: str) -> Type[BasePlugin]:
"""Import the plugin's plugin.py and return the BasePlugin subclass.
Raises PluginNotFoundError if plugin.py is missing.
Raises PluginContractError if the module has no BasePlugin subclass
or import fails.
"""
if name in self._plugin_classes:
return self._plugin_classes[name]
plugin_dir = self.plugins_dir / name
plugin_module_path = plugin_dir / 'plugin.py'
plugin_module_path = self.plugins_dir / name / 'plugin.py'
if not plugin_module_path.exists():
logger.error(f"Plugin {name} not found: {plugin_module_path}")
return None
raise PluginNotFoundError(
f'Plugin {name} plugin.py not found at {plugin_module_path}',
plugin_name=name,
)
try:
# Import the plugin module
spec = importlib.util.spec_from_file_location(
f"plugins.{name}.plugin",
plugin_module_path
f'plugins.{name}.plugin', plugin_module_path,
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
except Exception as e:
raise PluginContractError(
f'Plugin {name} import failed: {e}',
plugin_name=name,
) from e
# Find the plugin class
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (isinstance(attr, type) and
issubclass(attr, BasePlugin) and
attr is not BasePlugin):
if (isinstance(attr, type)
and issubclass(attr, BasePlugin)
and attr is not BasePlugin):
self._plugin_classes[name] = attr
return attr
logger.error(f"No BasePlugin subclass found in {name}")
return None
except Exception as e:
logger.error(f"Error loading plugin {name}: {e}")
return None
raise PluginContractError(
f'Plugin {name} module has no BasePlugin subclass',
plugin_name=name,
)
def load_plugin(self, name: str, app: Flask, db) -> Optional[BasePlugin]:
"""
Load and instantiate a plugin.
"""Load and instantiate a plugin.
Returns the plugin instance on success. In strict mode (dev/test),
any failure raises. In production, returns None and logs the
failure with full context.
"""
if name in self._loaded_plugins:
return self._loaded_plugins[name]
plugin_class = self.load_plugin_class(name)
if not plugin_class:
return None
try:
# Instantiate plugin
plugin = plugin_class()
from shopdb import __contract_version__
# Check dependencies
for dep in plugin.meta.dependencies:
manifest = self.load_manifest(name)
self.check_contract_version(name, __contract_version__)
for dep in manifest.get('dependencies', []):
if not self.registry.is_enabled(dep):
logger.error(
f"Plugin {name} requires {dep} which is not enabled"
raise PluginDependencyError(
f'Plugin {name} requires {dep} which is not enabled',
plugin_name=name,
)
return None
# Initialize plugin
plugin_class = self.load_plugin_class(name)
plugin = plugin_class()
plugin.init_app(app, db)
self._loaded_plugins[name] = plugin
return plugin
except PluginError as e:
self._handle_failure(app, e, name)
return None
except Exception as e:
logger.error(f"Error instantiating plugin {name}: {e}")
wrapped = PluginError(
f'Unexpected failure loading plugin {name}: {e}',
plugin_name=name,
)
self._handle_failure(app, wrapped, name)
return None
def load_enabled_plugins(self, app: Flask, db) -> Dict[str, BasePlugin]:
"""
Load all enabled plugins.
Returns dict of name -> plugin instance.
"""
"""Load all enabled plugins in dependency order."""
loaded = {}
# Sort by dependencies (simple topological sort)
enabled = self.registry.get_enabled_plugins()
sorted_plugins = self._sort_by_dependencies(enabled)
@@ -131,14 +234,16 @@ class PluginLoader:
plugin = self.load_plugin(name, app, db)
if plugin:
loaded[name] = plugin
logger.info(f"Loaded plugin: {name} v{plugin.meta.version}")
else:
logger.warning(f"Failed to load plugin: {name}")
logger.info(f'Loaded plugin: {name} v{plugin.meta.version}')
return loaded
def _sort_by_dependencies(self, plugin_names: List[str]) -> List[str]:
"""Sort plugins so dependencies come first."""
"""Sort plugins so dependencies come first.
Reads dependencies from manifest.json directly; does not
instantiate plugin classes during sort.
"""
sorted_list = []
visited = set()
@@ -147,15 +252,12 @@ class PluginLoader:
return
visited.add(name)
plugin_class = self.load_plugin_class(name)
if plugin_class:
# Create temporary instance to get meta
try:
temp = plugin_class()
for dep in temp.meta.dependencies:
manifest = self.load_manifest(name)
for dep in manifest.get('dependencies', []):
if dep in plugin_names:
visit(dep)
except Exception:
except PluginError:
pass
sorted_list.append(name)

127
tests/test_api_namespace.py Normal file
View File

@@ -0,0 +1,127 @@
"""Tests for the public shopdb.api namespace exposed to plugins.
Pins audit_log, resolve_asset_position, and the BasePlugin
get_setting/set_setting helpers as part of the contract surface.
"""
import pytest
from shopdb.api import audit_log, resolve_asset_position
from shopdb.core.models import AuditLog, Setting
def test_audit_log_creates_row(db, client):
"""audit_log writes an AuditLog row with the standard fields."""
with client.application.test_request_context('/'):
entry = audit_log(
action='created',
entitytype='Computer',
entityid=42,
entityname='PC-1234',
changes={'before': {}, 'after': {'hostname': 'PC-1234'}},
)
assert entry is not None
assert entry.action == 'created'
assert entry.entitytype == 'Computer'
assert entry.entityid == 42
assert entry.entityname == 'PC-1234'
saved = AuditLog.query.filter_by(entityid=42, entitytype='Computer').first()
assert saved is not None
def test_resolve_asset_position_returns_none_when_no_data():
"""An asset with no coords and no location returns None."""
class FakeAsset:
mapx = None
mapy = None
location = None
assert resolve_asset_position(FakeAsset()) is None
def test_resolve_asset_position_uses_self_when_set():
"""Asset-specific coords win over everything else."""
class FakeLocation:
mapx = 10
mapy = 20
class FakeAsset:
mapx = 100
mapy = 200
location = FakeLocation()
result = resolve_asset_position(FakeAsset())
assert result == {'mapx': 100, 'mapy': 200, 'positionsource': 'self'}
def test_resolve_asset_position_falls_back_to_location():
"""When asset has no coords, falls back to location coords."""
class FakeLocation:
mapx = 50
mapy = 75
class FakeAsset:
mapx = None
mapy = None
location = FakeLocation()
result = resolve_asset_position(FakeAsset())
assert result == {'mapx': 50, 'mapy': 75, 'positionsource': 'location'}
def test_resolve_asset_position_handles_asset_without_mapx_attr():
"""Assets that don't yet have mapx/mapy columns degrade gracefully."""
class FakeLocation:
mapx = 1
mapy = 2
class FakeAsset:
location = FakeLocation()
result = resolve_asset_position(FakeAsset())
assert result == {'mapx': 1, 'mapy': 2, 'positionsource': 'location'}
def test_plugin_get_setting_returns_default_when_unset(db, app):
"""A plugin reading an unset setting gets the default."""
from shopdb.plugins import plugin_manager
with app.app_context():
printers = plugin_manager.get_plugin('printers')
if printers is None:
pytest.skip('printers plugin not loaded')
assert printers.get_setting('nonexistentkey', default='fallback') == 'fallback'
def test_plugin_set_and_get_setting_roundtrip(db, app):
"""A plugin can set and read its own setting; key is namespaced."""
from shopdb.plugins import plugin_manager
with app.app_context():
printers = plugin_manager.get_plugin('printers')
if printers is None:
pytest.skip('printers plugin not loaded')
printers.set_setting('zabbix_url', 'http://zabbix.example.com')
assert printers.get_setting('zabbix_url') == 'http://zabbix.example.com'
raw = Setting.query.filter_by(key='plugin.printers.zabbix_url').first()
assert raw is not None
assert raw.value == 'http://zabbix.example.com'
def test_plugin_setting_is_namespaced_per_plugin(db, app):
"""Two plugins using the same key do not collide."""
from shopdb.plugins import plugin_manager
with app.app_context():
printers = plugin_manager.get_plugin('printers')
computers = plugin_manager.get_plugin('computers')
if printers is None or computers is None:
pytest.skip('required plugins not loaded')
printers.set_setting('shared_key', 'printers_value')
computers.set_setting('shared_key', 'computers_value')
assert printers.get_setting('shared_key') == 'printers_value'
assert computers.get_setting('shared_key') == 'computers_value'

261
tests/test_plugin_loader.py Normal file
View File

@@ -0,0 +1,261 @@
"""Tests for the plugin loader behavior.
Pins the manifest-first design and the strict-vs-isolate failure policy
described in the enforcing-plugin-contract skill.
"""
import json
from pathlib import Path
import pytest
from shopdb.plugins.loader import PluginLoader
from shopdb.plugins.registry import PluginRegistry
from shopdb.exceptions import (
PluginNotFoundError,
PluginContractError,
PluginVersionError,
PluginDependencyError,
PluginError,
)
@pytest.fixture
def temp_plugins_dir(tmp_path):
"""An empty temporary plugins directory."""
plugins = tmp_path / 'plugins'
plugins.mkdir()
return plugins
@pytest.fixture
def temp_registry(tmp_path):
"""An empty temporary registry."""
return PluginRegistry(tmp_path / 'plugins.json')
@pytest.fixture
def loader(temp_plugins_dir, temp_registry):
"""A loader pointed at the temporary plugins dir."""
return PluginLoader(temp_plugins_dir, temp_registry)
def write_plugin(plugins_dir, name, manifest=None, plugin_py=None):
"""Helper: write a plugin directory with manifest.json and plugin.py."""
plugin_dir = plugins_dir / name
plugin_dir.mkdir()
if manifest is not None:
(plugin_dir / 'manifest.json').write_text(json.dumps(manifest))
if plugin_py is not None:
(plugin_dir / 'plugin.py').write_text(plugin_py)
return plugin_dir
VALID_PLUGIN_PY = '''
from shopdb.plugins.base import BasePlugin, PluginMeta
class FakePlugin(BasePlugin):
@property
def meta(self):
return PluginMeta(
name='fake',
version='1.0.0',
description='Fake test plugin',
)
def get_blueprint(self):
return None
def get_models(self):
return []
'''
def test_load_manifest_raises_on_missing_manifest(loader, temp_plugins_dir):
"""A plugin without manifest.json raises PluginNotFoundError."""
plugin_dir = temp_plugins_dir / 'noplugin'
plugin_dir.mkdir()
(plugin_dir / 'plugin.py').write_text('# empty')
with pytest.raises(PluginNotFoundError, match='manifest.json'):
loader.load_manifest('noplugin')
def test_load_manifest_raises_on_invalid_json(loader, temp_plugins_dir):
"""An unparseable manifest.json raises PluginContractError."""
plugin_dir = temp_plugins_dir / 'bad'
plugin_dir.mkdir()
(plugin_dir / 'manifest.json').write_text('{not json}')
with pytest.raises(PluginContractError, match='invalid manifest.json'):
loader.load_manifest('bad')
def test_load_manifest_raises_on_missing_required_fields(loader, temp_plugins_dir):
"""Missing name/version/description raises PluginContractError."""
write_plugin(temp_plugins_dir, 'incomplete', manifest={'name': 'incomplete'})
with pytest.raises(PluginContractError, match='missing required field'):
loader.load_manifest('incomplete')
def test_load_manifest_raises_on_name_mismatch(loader, temp_plugins_dir):
"""manifest.name must match the directory name."""
write_plugin(
temp_plugins_dir,
'mydir',
manifest={'name': 'different', 'version': '1.0.0', 'description': 'x'},
)
with pytest.raises(PluginContractError, match='does not match manifest name'):
loader.load_manifest('mydir')
def test_check_contract_version_passes_when_in_range(loader, temp_plugins_dir):
"""A core_version range that covers the framework version passes."""
write_plugin(
temp_plugins_dir,
'compat',
manifest={
'name': 'compat',
'version': '1.0.0',
'description': 'x',
'core_version': '>=0.1.0,<1.0.0',
},
)
loader.check_contract_version('compat', '0.1.0')
def test_check_contract_version_raises_when_out_of_range(loader, temp_plugins_dir):
"""A core_version range that excludes the framework version raises."""
write_plugin(
temp_plugins_dir,
'incompat',
manifest={
'name': 'incompat',
'version': '1.0.0',
'description': 'x',
'core_version': '>=2.0.0',
},
)
with pytest.raises(PluginVersionError, match='requires core_version'):
loader.check_contract_version('incompat', '0.1.0')
def test_check_contract_version_skipped_when_unspecified(loader, temp_plugins_dir):
"""A manifest without core_version does not raise."""
write_plugin(
temp_plugins_dir,
'unspec',
manifest={
'name': 'unspec',
'version': '1.0.0',
'description': 'x',
},
)
loader.check_contract_version('unspec', '0.1.0')
def test_load_plugin_class_raises_on_missing_plugin_py(loader, temp_plugins_dir):
"""A directory with no plugin.py raises PluginNotFoundError."""
plugin_dir = temp_plugins_dir / 'nofile'
plugin_dir.mkdir()
with pytest.raises(PluginNotFoundError, match='plugin.py not found'):
loader.load_plugin_class('nofile')
def test_load_plugin_class_raises_when_no_subclass_found(loader, temp_plugins_dir):
"""A plugin.py without a BasePlugin subclass raises PluginContractError."""
write_plugin(
temp_plugins_dir,
'empty',
manifest={'name': 'empty', 'version': '1.0.0', 'description': 'x'},
plugin_py='# no plugin defined',
)
with pytest.raises(PluginContractError, match='no BasePlugin subclass'):
loader.load_plugin_class('empty')
def test_load_plugin_class_raises_on_import_error(loader, temp_plugins_dir):
"""A plugin.py with a syntax/import error raises PluginContractError."""
write_plugin(
temp_plugins_dir,
'broken',
manifest={'name': 'broken', 'version': '1.0.0', 'description': 'x'},
plugin_py='import nonexistent_module_xyz',
)
with pytest.raises(PluginContractError, match='import failed'):
loader.load_plugin_class('broken')
def test_sort_by_dependencies_uses_manifest_not_instantiation(loader, temp_plugins_dir):
"""Topological sort reads manifest.json directly, never instantiates."""
write_plugin(
temp_plugins_dir,
'a',
manifest={
'name': 'a',
'version': '1.0.0',
'description': 'x',
'dependencies': ['b'],
},
)
write_plugin(
temp_plugins_dir,
'b',
manifest={'name': 'b', 'version': '1.0.0', 'description': 'x'},
)
sorted_names = loader._sort_by_dependencies(['a', 'b'])
assert sorted_names.index('b') < sorted_names.index('a')
def test_load_plugin_strict_mode_reraises(loader, app, temp_plugins_dir):
"""In TESTING/DEBUG mode, load_plugin re-raises plugin errors."""
write_plugin(
temp_plugins_dir,
'incompat',
manifest={
'name': 'incompat',
'version': '1.0.0',
'description': 'x',
'core_version': '>=99.0.0',
},
plugin_py=VALID_PLUGIN_PY,
)
assert app.config.get('TESTING') is True
with pytest.raises(PluginVersionError):
loader.load_plugin('incompat', app, None)
def test_load_plugin_production_mode_isolates(loader, temp_plugins_dir):
"""In production-like config, load_plugin returns None on failure."""
from flask import Flask
fake_app = Flask(__name__)
fake_app.config['TESTING'] = False
fake_app.config['DEBUG'] = False
write_plugin(
temp_plugins_dir,
'incompat',
manifest={
'name': 'incompat',
'version': '1.0.0',
'description': 'x',
'core_version': '>=99.0.0',
},
plugin_py=VALID_PLUGIN_PY,
)
result = loader.load_plugin('incompat', fake_app, None)
assert result is None