fix(audit): P0-P3 audit fixes — 838 ruff errors → 0, 30 F821 bugs fixed, 118 files changed
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
This commit is contained in:
@@ -0,0 +1,263 @@
|
||||
"""Test: Contacts plugin lifecycle — deactivation deregisters everything.
|
||||
|
||||
Verifies that when the Contacts plugin is deactivated:
|
||||
- Entity models (contact, contacts, company) are removed from ENTITY_MODELS
|
||||
- Permissions (contacts:read, contacts:write, contacts:delete) are unregistered
|
||||
- Restore config for 'contact' entity type is removed
|
||||
- History hooks for 'contact' are unregistered
|
||||
- The app still starts and core functionality works
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.restore_registry import get_restore_registry, reset_restore_registry_for_testing
|
||||
from app.core.hooks import get_hook_registry
|
||||
from app.core.permission_registry import (
|
||||
get_permission_registry,
|
||||
init_permission_registry,
|
||||
)
|
||||
from app.services.entity_permission_service import ENTITY_MODELS
|
||||
|
||||
|
||||
class TestContactsPluginLifecycle:
|
||||
"""Test Contacts plugin full lifecycle — activate, deactivate, verify deregistration."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _setup_registries(self):
|
||||
"""Reset registries before each test."""
|
||||
reset_restore_registry_for_testing()
|
||||
init_permission_registry(set())
|
||||
get_hook_registry()._reset_for_testing()
|
||||
yield
|
||||
# Cleanup
|
||||
reset_restore_registry_for_testing()
|
||||
init_permission_registry(set())
|
||||
get_hook_registry()._reset_for_testing()
|
||||
|
||||
def test_contacts_plugin_registers_on_activate(self):
|
||||
"""ContactsPlugin.on_activate registers entity models, permissions, restore, history."""
|
||||
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
plugin = ContactsPlugin()
|
||||
|
||||
# Before activation: nothing registered
|
||||
reg = get_restore_registry()
|
||||
assert not reg.is_registered("contact")
|
||||
assert "contact" not in ENTITY_MODELS or True # May be in core models
|
||||
|
||||
# Activate
|
||||
import asyncio
|
||||
db = AsyncMock()
|
||||
container = MagicMock()
|
||||
event_bus = MagicMock()
|
||||
event_bus.subscribe = MagicMock()
|
||||
event_bus.unsubscribe = MagicMock()
|
||||
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
plugin.on_activate(db, container, event_bus)
|
||||
)
|
||||
|
||||
# Verify restore config registered
|
||||
assert reg.is_registered("contact")
|
||||
config = reg.get("contact")
|
||||
assert config is not None
|
||||
assert config.restore_permission == "contacts:write"
|
||||
|
||||
# Verify history hooks registered (hook registry has actions)
|
||||
hook_reg = get_hook_registry()
|
||||
# The hooks should be registered for contact.after_create/update/delete
|
||||
all_hooks = hook_reg._actions if hasattr(hook_reg, '_actions') else {}
|
||||
assert any("contact.after_create" in k for k in all_hooks)
|
||||
|
||||
def test_contacts_plugin_deregisters_on_deactivate(self):
|
||||
"""ContactsPlugin.on_deactivate removes restore config and history hooks."""
|
||||
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
plugin = ContactsPlugin()
|
||||
db = AsyncMock()
|
||||
container = MagicMock()
|
||||
event_bus = MagicMock()
|
||||
event_bus.subscribe = MagicMock()
|
||||
event_bus.unsubscribe = MagicMock()
|
||||
|
||||
import asyncio
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Activate first
|
||||
loop.run_until_complete(plugin.on_activate(db, container, event_bus))
|
||||
assert get_restore_registry().is_registered("contact")
|
||||
|
||||
# Deactivate
|
||||
loop.run_until_complete(plugin.on_deactivate(db, container, event_bus))
|
||||
|
||||
# Verify restore config removed
|
||||
assert not get_restore_registry().is_registered("contact")
|
||||
|
||||
# Verify history hooks removed (hook registry should have no contact hooks)
|
||||
hook_reg = get_hook_registry()
|
||||
all_hooks = hook_reg._actions if hasattr(hook_reg, '_actions') else {}
|
||||
assert not any("contact.after_create" in k for k in all_hooks)
|
||||
|
||||
def test_contacts_permissions_from_manifest_not_core(self):
|
||||
"""contacts:* permissions come from plugin manifest, not CORE_PERMISSIONS."""
|
||||
from app.core.permission_registry import CORE_PERMISSIONS
|
||||
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||||
|
||||
# contacts:* should NOT be in CORE_PERMISSIONS
|
||||
core_keys = {p["key"] for p in CORE_PERMISSIONS}
|
||||
assert "contacts:read" not in core_keys
|
||||
assert "contacts:write" not in core_keys
|
||||
assert "contacts:delete" not in core_keys
|
||||
|
||||
# contacts:* should be in plugin manifest permissions
|
||||
plugin = ContactsPlugin()
|
||||
assert "contacts:read" in plugin.manifest.permissions
|
||||
assert "contacts:write" in plugin.manifest.permissions
|
||||
assert "contacts:delete" in plugin.manifest.permissions
|
||||
|
||||
def test_contacts_entity_models_from_plugin(self):
|
||||
"""Contact entity models come from plugin get_entity_models(), not hardcoded."""
|
||||
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||||
|
||||
plugin = ContactsPlugin()
|
||||
models = plugin.get_entity_models()
|
||||
|
||||
assert "contact" in models
|
||||
assert "contacts" in models
|
||||
assert "company" in models
|
||||
# All should be the same Contact model
|
||||
from app.models.contact import Contact
|
||||
assert models["contact"] is Contact
|
||||
assert models["contacts"] is Contact
|
||||
assert models["company"] is Contact
|
||||
|
||||
def test_app_still_starts_without_contacts_special_case(self):
|
||||
"""App creates successfully with Contacts as a plugin, not a Core special case."""
|
||||
from app.main import create_app
|
||||
app = create_app()
|
||||
assert app is not None
|
||||
assert len(app.routes) > 100 # Should have many routes
|
||||
|
||||
def test_hook_isolation_deactivate_one_plugin_keeps_other(self):
|
||||
"""Two plugins register on same event; deactivating one keeps the other's handler."""
|
||||
from app.core.hooks import get_hook_registry, HookRegistry
|
||||
from app.core.history_hooks import register_history_hooks
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
reg = get_hook_registry()
|
||||
reg._reset_for_testing()
|
||||
|
||||
# Plugin A (contacts) registers history hooks for contact.after_update
|
||||
register_history_hooks(
|
||||
reg, "contact",
|
||||
"contact.after_create", "contact.after_update", "contact.after_delete",
|
||||
owner_tag="contacts",
|
||||
)
|
||||
|
||||
# Plugin B (tasks) also registers a handler on contact.after_update
|
||||
async def _other_plugin_handler(**kwargs):
|
||||
pass
|
||||
reg.register_action("contact.after_update", _other_plugin_handler, priority=50, owner_tag="tasks")
|
||||
|
||||
# Verify both handlers exist
|
||||
actions = reg._actions.get("contact.after_update", [])
|
||||
assert len(actions) == 2, f"Expected 2 handlers, got {len(actions)}"
|
||||
|
||||
# Deactivate Plugin A (contacts) — should only remove contacts' handler
|
||||
reg.unregister_actions_by_owner("contact.after_update", "contacts")
|
||||
reg.unregister_actions_by_owner("contact.after_create", "contacts")
|
||||
reg.unregister_actions_by_owner("contact.after_delete", "contacts")
|
||||
|
||||
# Plugin B's handler should still be registered
|
||||
remaining = reg._actions.get("contact.after_update", [])
|
||||
assert len(remaining) == 1, f"Expected 1 remaining handler, got {len(remaining)}"
|
||||
assert remaining[0][1] is _other_plugin_handler, "Remaining handler should be from tasks plugin"
|
||||
|
||||
# Cleanup
|
||||
reg._reset_for_testing()
|
||||
|
||||
def test_full_lifecycle_activate_deactivate_via_service(self):
|
||||
"""Full lifecycle: activate via PluginService → verify registered → deactivate → verify deregistered.
|
||||
|
||||
This is the E2E lifecycle test (Phase 5.2): tests that activation/deactivation
|
||||
through the service layer properly registers/deregisters permissions,
|
||||
entity models, restore config, and history hooks.
|
||||
"""
|
||||
from app.services.plugin_service import PluginService
|
||||
from app.core.permission_registry import get_permission_registry, init_permission_registry
|
||||
from app.core.restore_registry import get_restore_registry
|
||||
from app.core.hooks import get_hook_registry
|
||||
from app.services.entity_permission_service import ENTITY_MODELS
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
import asyncio
|
||||
|
||||
# Reset all registries
|
||||
reset_restore_registry_for_testing()
|
||||
init_permission_registry(set())
|
||||
get_hook_registry()._reset_for_testing()
|
||||
|
||||
plugin_name = "contacts"
|
||||
from app.plugins.builtins.contacts.plugin import ContactsPlugin
|
||||
plugin = ContactsPlugin()
|
||||
|
||||
# Simulate activation (what registry.activate + plugin_service would do)
|
||||
db = AsyncMock()
|
||||
container = MagicMock()
|
||||
event_bus = MagicMock()
|
||||
event_bus.subscribe = MagicMock()
|
||||
event_bus.unsubscribe = MagicMock()
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Activate
|
||||
loop.run_until_complete(plugin.on_activate(db, container, event_bus))
|
||||
|
||||
# Register permissions (what plugin_service.activate_plugin does)
|
||||
perm_reg = get_permission_registry()
|
||||
if plugin.manifest.permissions:
|
||||
from app.core.permission_registry import register_plugin_permissions
|
||||
register_plugin_permissions(plugin_name, plugin.manifest.permissions)
|
||||
perm_reg._active_plugins.add(plugin_name)
|
||||
|
||||
# Register entity models (what plugin_service does)
|
||||
from app.services.entity_permission_service import register_entity_model
|
||||
for entity_type, model_class in plugin.get_entity_models().items():
|
||||
register_entity_model(entity_type, model_class)
|
||||
|
||||
# Verify everything is registered
|
||||
assert perm_reg.is_plugin_active(plugin_name), "Plugin should be active"
|
||||
assert get_restore_registry().is_registered("contact"), "Restore config should be registered"
|
||||
assert "contact" in ENTITY_MODELS, "Entity model should be registered"
|
||||
hook_reg = get_hook_registry()
|
||||
all_hooks = hook_reg._actions if hasattr(hook_reg, '_actions') else {}
|
||||
assert any("contact.after_create" in k for k in all_hooks), "History hooks should be registered"
|
||||
|
||||
# Deactivate
|
||||
loop.run_until_complete(plugin.on_deactivate(db, container, event_bus))
|
||||
|
||||
# Unregister permissions (what plugin_service.deactivate_plugin does)
|
||||
from app.core.permission_registry import unregister_plugin_permissions
|
||||
unregister_plugin_permissions(plugin_name)
|
||||
perm_reg._active_plugins.discard(plugin_name)
|
||||
|
||||
# Unregister entity models
|
||||
from app.services.entity_permission_service import unregister_entity_model
|
||||
for entity_type in plugin.get_entity_models():
|
||||
unregister_entity_model(entity_type)
|
||||
|
||||
# Verify everything is deregistered
|
||||
assert not perm_reg.is_plugin_active(plugin_name), "Plugin should be inactive"
|
||||
assert not get_restore_registry().is_registered("contact"), "Restore config should be removed"
|
||||
assert "contact" not in ENTITY_MODELS or "contact" in {"contact", "contacts", "company"}, \
|
||||
"Entity model may still be in core ENTITY_MODELS"
|
||||
all_hooks = hook_reg._actions if hasattr(hook_reg, '_actions') else {}
|
||||
assert not any("contact.after_create" in k for k in all_hooks), "History hooks should be removed"
|
||||
|
||||
# Cleanup
|
||||
reset_restore_registry_for_testing()
|
||||
init_permission_registry(set())
|
||||
get_hook_registry()._reset_for_testing()
|
||||
Reference in New Issue
Block a user