test(gate-a): block A completion proof - imports, lifecycle symmetry, activate-once, contract roundtrip

tests/test_arch_block_a.py

@@ -279,3 +279,132 @@ class TestWikiProviderLifecycle:
 
         await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
         assert registry.get('wiki_article') is None
+
+
+# --- Gate A: Block A completion proof (docs/fix-plan-v3.md) ---
+
+
+class TestGateAImports:
+    def test_every_app_module_imports(self):
+        import importlib
+        import pathlib
+
+        app_root = pathlib.Path('/a0/usr/projects/leocrm/app')
+        failures = []
+        seen = set()
+        for py_file in sorted(app_root.rglob('*.py')):
+            rel = py_file.relative_to(app_root.parent)
+            if py_file.name == '__init__.py':
+                mod_name = str(rel.parent).replace('/', '.')
+            else:
+                mod_name = str(rel.with_suffix('')).replace('/', '.')
+            if not mod_name or mod_name in seen:
+                continue
+            seen.add(mod_name)
+            try:
+                importlib.import_module(mod_name)
+            except Exception as exc:
+                failures.append(f'{mod_name}: {type(exc).__name__}: {exc}')
+        assert failures == [], (
+            f'{len(failures)} modules failed to import: ' + ' | '.join(failures)
+        )
+
+
+def _build_gate_a_plugin():
+    from app.plugins.base import BasePlugin
+    from app.plugins.manifest import PluginManifest
+
+    class GateATestPlugin(BasePlugin):
+        manifest = PluginManifest(
+            name='gate_a_test',
+            version='1.0.0',
+            display_name='Gate A Test',
+            description='Proves lifecycle symmetry.',
+            dependencies=[],
+            routes=[],
+            events=['gate_a.event'],
+            migrations=[],
+            permissions=['gate_a_test:read'],
+        )
+
+        async def on_activate(self, db, service_container, event_bus) -> None:
+            await super().on_activate(db, service_container, event_bus)
+            service_container.register('gate_a_marker', object())
+
+        async def on_deactivate(self, db, service_container, event_bus) -> None:
+            if service_container.has('gate_a_marker'):
+                service_container.remove('gate_a_marker')
+            await super().on_deactivate(db, service_container, event_bus)
+
+    return GateATestPlugin()
+
+
+class TestGateALifecycleSymmetry:
+    @pytest.mark.asyncio
+    async def test_activate_registers_deactivate_deregisters(self):
+        from app.core.event_bus import EventBus
+        from app.core.service_container import ServiceContainer
+
+        plugin = _build_gate_a_plugin()
+        container = ServiceContainer()
+        event_bus = EventBus()
+
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+        assert container.has('gate_a_marker') is True
+        assert len(plugin._event_handlers) == 1
+
+        await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
+        assert container.has('gate_a_marker') is False
+        assert plugin._event_handlers == {}
+
+        # re-activate: no duplicate handlers
+        await plugin.on_activate(db=None, service_container=container, event_bus=event_bus)
+        assert len(plugin._event_handlers) == 1
+        await plugin.on_deactivate(db=None, service_container=container, event_bus=event_bus)
+
+
+class TestGateAActivateOnce:
+    def test_activation_outside_tenant_loop_in_main(self):
+        # ARCH-002 contract: main.py activates each plugin once per process,
+        # never inside a per-tenant loop.
+        import pathlib
+
+        main_py = pathlib.Path('/a0/usr/projects/leocrm/app/main.py')
+        lines = main_py.read_text().splitlines()
+        activate_lines = [i for i, line in enumerate(lines) if 'on_activate(' in line and 'def ' not in line]
+        assert activate_lines, 'no plugin on_activate call found in main.py'
+        for i in activate_lines:
+            # scan upward to the nearest loop header; it must be a plugin
+            # loop, never a tenant loop
+            for j in range(i - 1, max(0, i - 40), -1):
+                stripped = lines[j].strip()
+                if stripped.startswith('for ') or stripped.startswith('async for '):
+                    assert 'tenant' not in stripped.split('in')[0], (
+                        f'on_activate sits inside a tenant loop: main.py line {j + 1}'
+                    )
+                    break
+
+
+class TestGateAContractRoundtrip:
+    def test_register_get_unregister_no_resurrect(self):
+        from app.plugins.builtins.contracts import (
+            get_contract,
+            get_contract_registry,
+            reset_contract_registry_for_testing,
+        )
+
+        reset_contract_registry_for_testing()
+        registry = get_contract_registry()
+
+        class Dummy:
+            contract_name = 'gate_a_dummy'
+
+        registry.register('gate_a_dummy', Dummy())
+        assert isinstance(get_contract('gate_a_dummy'), Dummy)
+
+        registry.unregister('gate_a_dummy')
+        assert get_contract('gate_a_dummy') is None
+        # second read must NOT lazily resurrect the unregistered contract
+        assert get_contract('gate_a_dummy') is None
+
+        reset_contract_registry_for_testing()