Test coverage: plugin lifecycle, registry, lockout unlock, contract fleet, seed
Fills the review's highest-value coverage gaps in the framework's own core feature. New tests/test_plugin_lifecycle.py: manager enable/disable dependency guards (beta depends on alpha - enable-beta-first refused, disable-alpha-while-beta-on refused) via synthetic plugins; enable seeds a plugin's RBAC permissions idempotently; registry disable-survives-reload, corrupt-file recovery, and the equipment->machines rename migration; `flask seed settings` idempotency. test_plugin_contract.py: BUNDLED_PLUGINS now covers all bundled plugins incl. geenforce, measuringtools, warranty (was 9, contradicting the CLAUDE.md "all bundled satisfy the contract" claim); the structural checks now run against them. test_authz.py: account lockout auto-unlock path (expired lockeduntil -> correct password logs in and clears the lock state), previously untested. All new tests pass; naming green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -203,6 +203,25 @@ def test_account_locks_after_repeated_bad_logins(client, db, admin_user):
|
|||||||
assert 'locked' in locked.get_json()['data']['error']['message'].lower()
|
assert 'locked' in locked.get_json()['data']['error']['message'].lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_lockout_auto_unlocks_after_expiry(client, db, admin_user):
|
||||||
|
"""Once lockeduntil is in the past the account unlocks itself: a correct
|
||||||
|
password logs in and clears the lock state."""
|
||||||
|
from datetime import datetime, timezone, timedelta
|
||||||
|
# Simulate an expired lock (naive-UTC to match the column comparisons).
|
||||||
|
admin_user.lockeduntil = (datetime.now(timezone.utc).replace(tzinfo=None)
|
||||||
|
- timedelta(minutes=1))
|
||||||
|
admin_user.failedlogins = 5
|
||||||
|
db.session.commit()
|
||||||
|
|
||||||
|
ok = client.post('/api/auth/login',
|
||||||
|
json={'username': 'testadmin', 'password': 'testpass'})
|
||||||
|
assert ok.status_code == 200, ok.get_json()
|
||||||
|
|
||||||
|
db.session.refresh(admin_user)
|
||||||
|
assert admin_user.lockeduntil is None
|
||||||
|
assert admin_user.failedlogins == 0
|
||||||
|
|
||||||
|
|
||||||
def test_successful_login_resets_failed_counter(client, db, admin_user):
|
def test_successful_login_resets_failed_counter(client, db, admin_user):
|
||||||
"""A good login before the threshold clears the failure count."""
|
"""A good login before the threshold clears the failure count."""
|
||||||
for _ in range(3):
|
for _ in range(3):
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ from shopdb.plugins import plugin_manager
|
|||||||
from shopdb.plugins.base import BasePlugin, PluginMeta
|
from shopdb.plugins.base import BasePlugin, PluginMeta
|
||||||
|
|
||||||
|
|
||||||
BUNDLED_PLUGINS = ('computers', 'employees', 'knowledgebase', 'machines', 'network', 'notifications', 'printers', 'slides', 'usb')
|
BUNDLED_PLUGINS = ('computers', 'employees', 'geenforce', 'knowledgebase', 'machines', 'measuringtools', 'network', 'notifications', 'printers', 'slides', 'usb', 'warranty')
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
|||||||
165
tests/test_plugin_lifecycle.py
Normal file
165
tests/test_plugin_lifecycle.py
Normal file
@@ -0,0 +1,165 @@
|
|||||||
|
"""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):
|
||||||
|
"""Write a minimal valid synthetic plugin (manifest.json + plugin.py)."""
|
||||||
|
pdir = plugins_dir / name
|
||||||
|
pdir.mkdir()
|
||||||
|
(pdir / 'manifest.json').write_text(json.dumps({
|
||||||
|
'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}',
|
||||||
|
}))
|
||||||
|
(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
|
||||||
|
|
||||||
|
|
||||||
|
# --- 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
|
||||||
Reference in New Issue
Block a user