leocrm/tests/test_gate_h.py

"""Gate H — prove that a plugin contributes ALL extension points without
changing a single core file.

Covers:
1. Agent tool  -> app.ai.tool_registry (core layer)
2. Workflow step type -> app.workflows.step_handlers
3. Fallback intent -> app.ai.action_mapper
4. Deactivate symmetry -> everything removed again
5. Re-activate -> present again, NO duplicates

The plugin under test is defined INLINE in this file — if this test passes,
no core file had to be touched to extend the platform.
"""

from __future__ import annotations

from typing import Any

import pytest

# ─── The Gate-H test plugin (inline — zero core changes) ─────────────────────


def _build_gate_h_plugin():
    from app.plugins.base import BasePlugin
    from app.plugins.manifest import PluginManifest

    class GateHTestPlugin(BasePlugin):
        """Minimal plugin exercising every Block-H contribution point."""

        manifest = PluginManifest(
            name="gate_h_test",
            version="1.0.0",
            display_name="Gate H Test",
            description="Proves plugin contributions without core changes.",
            dependencies=[],
            routes=[],
            events=[],
            migrations=[],
            permissions=["gate_h_test:read"],
        )

        async def on_activate(self, db, service_container, event_bus) -> None:
            await super().on_activate(db, service_container, event_bus)

            # 1) Agent tool via CORE registry
            from app.ai.tool_registry import get_tool_registry

            async def _echo_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
                return f"gate-h says: {arguments.get('text', '')}"

            get_tool_registry().register(
                name="gate_h_echo",
                description="Echo text back (Gate H proof).",
                parameters={
                    "type": "object",
                    "properties": {"text": {"type": "string"}},
                    "required": ["text"],
                },
                handler=_echo_handler,
                plugin_name=self.manifest.name,
                required_permission=None,
                category="test",
            )

            # 2) Workflow step type
            from app.workflows.step_handlers import StepResult, register_step_type

            @register_step_type("gate_h_noop")
            async def _handle_gate_h_noop(db, tenant_id, instance, step):  # noqa: ANN001
                return StepResult(output={"gate_h": True})

            # 3) Fallback intent pattern
            from app.ai.action_mapper import register_intent_pattern

            def _invoice_intent(query: str, context: dict[str, Any] | None):
                return [
                    {
                        "method": "POST",
                        "path": "/api/v1/gate-h/invoices",
                        "body": {},
                        "description": "Create invoice (Gate H proof)",
                        "confidence": 0.9,
                    }
                ]

            _invoice_intent.owner_tag = self.manifest.name
            register_intent_pattern(r"\b(create|new)\b.*\binvoice\b", _invoice_intent, owner=self.manifest.name)

        async def on_deactivate(self, db, service_container, event_bus) -> None:
            # Symmetry: remove everything registered above
            from app.ai.tool_registry import get_tool_registry
            from app.workflows.step_handlers import unregister_step_type
            from app.ai.action_mapper import unregister_intent_patterns

            get_tool_registry().unregister_plugin(self.manifest.name)
            unregister_step_type("gate_h_noop")
            unregister_intent_patterns(self.manifest.name)
            await super().on_deactivate(db, service_container, event_bus)

    return GateHTestPlugin


# ─── The proof ────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_gate_h_full_contribution_lifecycle():
    """Activate -> contributed everywhere; deactivate -> gone; reactivate -> back, no dupes."""
    from app.ai.tool_registry import get_tool_registry
    from app.workflows.step_handlers import get_step_handler
    from app.ai.action_mapper import map_query_to_actions

    plugin_cls = _build_gate_h_plugin()
    plugin = plugin_cls()

    # ── Before activation: nothing registered ──
    assert get_tool_registry().get("gate_h_echo") is None
    assert get_step_handler("gate_h_noop") is None
    assert map_query_to_actions("create invoice for Acme") == []

    # ── Activate ──
    await plugin.on_activate(db=None, service_container=None, event_bus=None)

    # 1) Agent tool present in CORE registry
    tool = get_tool_registry().get("gate_h_echo")
    assert tool is not None
    assert tool.plugin_name == "gate_h_test"
    result = await tool.handler({"text": "hello"}, {})
    assert result == "gate-h says: hello"

    # 2) Workflow step type resolvable
    handler = get_step_handler("gate_h_noop")
    assert handler is not None

    # 3) Intent contributes actions for a FACHMODUL term unknown to core
    actions = map_query_to_actions("create invoice for Acme")
    assert len(actions) == 1
    assert actions[0]["path"] == "/api/v1/gate-h/invoices"

    # ── Deactivate: full symmetry ──
    await plugin.on_deactivate(db=None, service_container=None, event_bus=None)
    assert get_tool_registry().get("gate_h_echo") is None
    assert get_step_handler("gate_h_noop") is None
    assert map_query_to_actions("create invoice for Acme") == []

    # ── Re-activate: back AND no duplicates ──
    await plugin.on_activate(db=None, service_container=None, event_bus=None)
    tool = get_tool_registry().get("gate_h_echo")
    assert tool is not None  # single entry (dict keyed by name — overwrite, not dupe)
    assert get_step_handler("gate_h_noop") is not None
    actions = map_query_to_actions("create invoice for Acme")
    assert len(actions) == 1  # exactly one contributed intent, not two

    # Cleanup so other tests are unaffected
    await plugin.on_deactivate(db=None, service_container=None, event_bus=None)


@pytest.mark.asyncio
async def test_gate_h_agent_runtime_survives_ai_assistant_absence():
    """The core tool registry works even though ai_assistant plugin module is never imported."""
    from app.ai.tool_registry import ToolRegistry, get_tool_registry

    reg = get_tool_registry()
    assert isinstance(reg, ToolRegistry)
    # And it must be THE same singleton the shim exposes
    from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry as shim_get

    assert shim_get() is reg