test(gate-b): new-plugin-without-core-changes + dependency blockade proofs

tests/test_arch_block_a.py

@@ -408,3 +408,171 @@ class TestGateAContractRoundtrip:
         assert get_contract('gate_a_dummy') is None
 
         reset_contract_registry_for_testing()
+
+
+# --- Gate B: Block B completion proof (docs/fix-plan-v3.md) ---
+
+
+def _install_inline_route_module():
+    """Register a throwaway router module in sys.modules (no file on disk)."""
+    import sys
+    import types
+
+    from fastapi import APIRouter
+
+    mod_name = 'gate_b_inline_routes'
+    if mod_name in sys.modules:
+        return sys.modules[mod_name]
+    mod = types.ModuleType(mod_name)
+    router = APIRouter(prefix='/api/v1/gate-b-inline', tags=['gate-b-inline'])
+
+    @router.get('/ping')
+    async def ping():
+        return {'pong': 'gate-b'}
+
+    mod.router = router
+    sys.modules[mod_name] = mod
+    return mod
+
+
+class TestGateBNewPluginNoCoreChanges:
+    def test_inline_plugin_route_mounted_and_entity_registered(self):
+        import uuid
+
+        from fastapi import Depends, FastAPI
+        from sqlalchemy import Column, Integer, String, Uuid
+
+        from app.core.db import Base
+        from app.deps import require_active_plugin
+        from app.plugins.base import BasePlugin
+        from app.plugins.manifest import PluginManifest, PluginRouteDef
+        from app.plugins.registry import get_registry
+        from app.services.entity_permission_service import (
+            ENTITY_MODELS,
+            register_entity_model,
+            unregister_entity_model,
+        )
+
+        _install_inline_route_module()
+
+        # Classic Column style: Mapped[] annotations cannot be resolved for
+        # classes defined inside a function (no module-level globals).
+        class GateBInlineThing(Base):
+            __tablename__ = 'gate_b_inline_things'
+            id = Column(Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4)
+            tenant_id = Column(Uuid(as_uuid=True), nullable=False)
+            name = Column(String(100))
+            sort_order = Column(Integer)
+
+        class GateBInlinePlugin(BasePlugin):
+            manifest = PluginManifest(
+                name='gate_b_inline',
+                version='1.0.0',
+                display_name='Gate B Inline',
+                description='Proves new-plugin capability without core changes.',
+                dependencies=[],
+                routes=[
+                    PluginRouteDef(
+                        path='/api/v1/gate-b-inline',
+                        module='gate_b_inline_routes',
+                        router_attr='router',
+                    )
+                ],
+                events=[],
+                migrations=[],
+                permissions=['gate_b_inline:read'],
+            )
+
+            def get_entity_models(self):
+                return {'gate_b_thing': GateBInlineThing}
+
+        plugin = GateBInlinePlugin()
+        registry = get_registry()
+        registry._plugins['gate_b_inline'] = plugin
+
+        try:
+            # 1) Route gets mounted by the plugin-route mechanism
+            app = FastAPI()
+            for route_def in plugin.manifest.routes:
+                import importlib
+
+                router_module = importlib.import_module(route_def.module)
+                router = getattr(router_module, route_def.router_attr)
+                app.include_router(
+                    router, dependencies=[Depends(require_active_plugin('gate_b_inline'))]
+                )
+            paths = app.openapi()['paths']
+            assert '/api/v1/gate-b-inline/ping' in paths
+
+            # 2) Entity registers through the standard activation path
+            register_entity_model('gate_b_thing', GateBInlineThing)
+            try:
+                assert ENTITY_MODELS['gate_b_thing'] is GateBInlineThing
+            finally:
+                unregister_entity_model('gate_b_thing')
+        finally:
+            registry._plugins.pop('gate_b_inline', None)
+
+
+class TestGateBDependencyBlockade:
+    @pytest.mark.asyncio
+    async def test_deactivate_blocked_while_dependent_active(self, db_session):
+        from app.models.plugin import Plugin as PluginModel
+        from app.plugins.base import BasePlugin
+        from app.plugins.manifest import PluginManifest
+        from app.plugins.registry import get_registry
+
+        class DepB(BasePlugin):
+            manifest = PluginManifest(
+                name='gate_b_dep_b',
+                version='1.0.0',
+                display_name='Dep B',
+                dependencies=[],
+                routes=[],
+                events=[],
+                migrations=[],
+                permissions=[],
+            )
+
+        class DepA(BasePlugin):
+            manifest = PluginManifest(
+                name='gate_b_dep_a',
+                version='1.0.0',
+                display_name='Dep A',
+                dependencies=['gate_b_dep_b'],
+                routes=[],
+                events=[],
+                migrations=[],
+                permissions=[],
+            )
+
+        registry = get_registry()
+        registry._plugins['gate_b_dep_b'] = DepB()
+        registry._plugins['gate_b_dep_a'] = DepA()
+
+        records = {}
+        try:
+            for name in ('gate_b_dep_b', 'gate_b_dep_a'):
+                rec = PluginModel(
+                    name=name,
+                    display_name=name,
+                    version='1.0.0',
+                    status='active',
+                    installed=True,
+                    active=True,
+                    is_core=False,
+                )
+                db_session.add(rec)
+                records[name] = rec
+            await db_session.commit()
+
+            # Deactivating the dependency must be blocked because gate_b_dep_a
+            # is active and depends on it.
+            with pytest.raises(ValueError, match='depend'):
+                await registry.deactivate(db_session, 'gate_b_dep_b')
+        finally:
+            for rec in records.values():
+                await db_session.delete(rec)
+            await db_session.commit()
+            registry._plugins.pop('gate_b_dep_b', None)
+            registry._plugins.pop('gate_b_dep_a', None)