diff --git a/tests/test_arch_block_a.py b/tests/test_arch_block_a.py index d5ab150..ba6daff 100644 --- a/tests/test_arch_block_a.py +++ b/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)