Orchestrator/bahn/teamlandkarte-mcp/tests/test_confirmation_gate.py

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from datetime import date
from typing import Any

from teamlandkarte_mcp.mcp_server import build_server


@dataclass
class FakeDBClient:
    def test_connection(self) -> None:
        return

    def get_table_columns(self, table: str) -> list[str]:
        if table == "teamlandkarte_v_capacity_roles_latest":
            return ["name", "active", "staffing_board_relevant"]
        if table == "teamlandkarte_v_capacities_latest":
            return ["creation_date"]
        if table == "teamlandkarte_v_teams_latest":
            return [
                "team_id",
                "ouid",
                "about_us",
                "offerings",
                "interests",
                "focus_name",
            ]
        if table == "teamlandkarte_v_teammeter_organizational_units_latest":
            return ["id", "name"]
        if table == "teamlandkarte_v_teammeter_team_competences_latest":
            return ["ouid", "competence_id", "top_competency"]
        if table == "teamlandkarte_v_team_references_latest":
            return ["ouid", "partner_id", "projects"]
        return []

    def get_all_role_names(self) -> list[str]:
        return ["Cloud Engineer"]

    def get_all_competence_names(self) -> list[str]:
        return ["AWS"]

    def get_open_tasks(self, limit: int = 20):
        return []

    def get_all_capacities_with_competences(self):  # pragma: no cover
        return []


class FakeMatcher:
    async def match(self, capacities, requirements):  # pragma: no cover
        raise AssertionError("match should not be called when gate blocks")


def _result_to_text(result: Any) -> str:
    # FastMCP.call_tool returns (content, structured) in this project.
    if isinstance(result, tuple) and len(result) == 2:
        content, structured = result
        if isinstance(structured, dict) and isinstance(structured.get("result"), str):
            return structured["result"]
        if isinstance(content, list) and content:
            # Typically TextContent(type='text', text='...')
            text = getattr(content[0], "text", None)
            if isinstance(text, str):
                return text
    if isinstance(result, str):
        return result
    return str(result)


def _call_tool(
    srv: Any,
    name: str,
    arguments: dict[str, Any] | None = None,
) -> str:
    if arguments is None:
        arguments = {}
    coro = srv.call_tool(name, arguments)
    return _result_to_text(asyncio.run(coro))


def _get_tools(srv: Any) -> dict[str, Any]:
    """Return a mapping of tool-name -> callable.

    The MCP library's internal structure can vary across versions.
    We try a few known locations and fall back to asking the server object.
    """

    for attr in ("_tools", "tools"):
        tools = getattr(srv, attr, None)
        if isinstance(tools, dict):
            return tools

    # Some MCP implementations expose tools via a method.
    for method_name in ("get_tools", "list_tools"):
        maybe = getattr(srv, method_name, None)
        if callable(maybe):
            tools = maybe()
            if isinstance(tools, dict):
                return tools

    raise AssertionError("Unable to locate tool registry on server instance")


def test_find_matching_capacities_requires_confirmation(
    monkeypatch,
    tmp_path,
) -> None:
    cfg = tmp_path / "config.toml"
    cfg.write_text(
        """
[database]
host='x'
port=1
username='u'
password='p'
backend='trino'

[matching]
competence_weight=0.8
role_weight=0.2
require_confirmation=true

[cache]
db_ttl_hours=1
search_ttl_minutes=10
max_size=10
""".strip(),
        encoding="utf-8",
    )

    # Patch DB client creation inside server builder.
    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: FakeDBClient())

    srv = build_server(str(cfg))

    msg = _call_tool(
        srv,
        "find_matching_capacities",
        {
            "role_name": "Cloud Engineer",
            "competences": ["AWS"],
            "date_start": str(date(2025, 1, 1)),
            "date_end": str(date(2025, 1, 31)),
        },
    )

    assert "must be confirmed" in msg


def test_find_matching_capacities_auto_skips_when_disabled(
    monkeypatch, tmp_path
) -> None:
    cfg = tmp_path / "config.toml"
    cfg.write_text(
        """
[database]
host='x'
port=1
username='u'
password='p'
backend='trino'

[matching]
competence_weight=0.8
role_weight=0.2
require_confirmation=false

[cache]
db_ttl_hours=1
search_ttl_minutes=10
max_size=10
""".strip(),
        encoding="utf-8",
    )

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: FakeDBClient())

    srv = build_server(str(cfg))

    # If auto-skip is enabled, the gate should not block. We don't assert the
    # full output (it depends on matcher/cache); just that it isn't blocked.
    # The call will still fail later because capacities list is empty, but it
    # must not fail with the confirmation message.
    msg = _call_tool(
        srv,
        "find_matching_capacities",
        {"role_name": "Cloud Engineer", "competences": ["AWS"]},
    )

    assert "must be confirmed" not in msg


def test_confirm_requires_review_after_latest_requirement_update(
    monkeypatch, tmp_path
) -> None:
    cfg = tmp_path / "config.toml"
    cfg.write_text(
        """
[database]
host='x'
port=1
username='u'
password='p'
backend='trino'

[matching]
competence_weight=0.8
role_weight=0.2
require_confirmation=true

[cache]
db_ttl_hours=1
search_ttl_minutes=10
max_size=10
""".strip(),
        encoding="utf-8",
    )

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: FakeDBClient())

    srv = build_server(str(cfg))

    # Wrong order: review before any requirements exist.
    msg = _call_tool(srv, "show_pending_requirements")
    assert "No pending requirements" in msg

    # Collect requirements afterwards.
    _call_tool(
        srv,
        "collect_structured_requirement_data",
        {"role_name": "Backend Developer", "competences": ["Python"]},
    )

    # Confirm must still be blocked: latest requirements were not reviewed.
    msg = _call_tool(srv, "confirm_requirements", {"confirm": True})
    assert "User confirmation has not been requested" in msg

    # Correct order: review after collecting, then confirm.
    _call_tool(srv, "show_pending_requirements")
    msg = _call_tool(srv, "confirm_requirements", {"confirm": True})
    assert "Requirements confirmed" in msg