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

# Feature: team-profile-matching, Task 9.11
"""Unit tests for the team-related MCP tools.

Validates the following acceptance criteria:

* **Anforderung 1.2**: The existing ``find_matching_capacities`` tool keeps its
  Markdown output structure after the team-profile refactoring (snapshot-style
  assertion on stable substrings).
* **Anforderung 1.3**: ``find_matching_teams`` is registered with ``FastMCP``.
* **Anforderung 9.1**: ``list_teams`` is registered and renders a Markdown
  table whose headers are exactly ``Team Id``, ``Team Name``, ``Schwerpunkt``,
  ``Anzahl Kompetenzen``, ``Anzahl Referenzen``.
* **Anforderung 9.2**: ``get_team_details`` is registered and renders the full
  layout (summary table + ``## Über uns`` / ``## Leistungen`` / ``## Interessen``
  / ``## Kompetenzen`` / ``## Referenzen`` / ``## Next steps``) for an existing
  team.
* **Anforderung 9.3**: ``get_team_details`` returns an error message that
  contains the requested (unknown) ``team_id`` substring when no team matches.
* **Anforderung 9.4**: Top competences are rendered with the ``(Top)`` suffix
  and references with a ``partner_name`` are rendered as
  ``**<partner>**: <projects>`` (bold partner name).

The tests use a minimal fake ``DBClient`` and stub out ``create_db_client`` so
that no real database is touched. The LLM is not exercised here.
"""

from __future__ import annotations

from typing import Any

import pytest

from teamlandkarte_mcp.mcp_server import build_server
from teamlandkarte_mcp.models import (
    Capacity,
    Team,
    TeamCompetence,
    TeamReference,
)


_CONFIG_TEMPLATE = """
[database]
host='x'
port=1
username='u'
password='p'
backend='trino'

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

[matching.team]
top_competency_weight=1.5

[cache]
db_ttl_hours=1
search_ttl_minutes=10
max_size=10
""".strip()


def _make_team(team_id: str = "t1") -> Team:
    """Construct a deterministic team with mixed top/non-top competences and
    a reference with and without partner name."""
    return Team(
        team_id=team_id,
        ouid=f"ou-{team_id}",
        team_name=f"Team {team_id.upper()}",
        focus_name="Backend Engineering",
        about_us="We build distributed backends.",
        offerings="APIs, ETL pipelines, cloud migrations.",
        interests="Event-driven architectures.",
        competences=[
            TeamCompetence(name="Python", top_competency=True),
            TeamCompetence(name="Java", top_competency=False),
        ],
        references=[
            TeamReference(partner_name="DB Cargo", projects="ETL Pipeline"),
            TeamReference(partner_name="", projects="Internal tooling"),
        ],
    )


class _FakeDB:
    """Minimal fake ``DBClient`` covering all tools used by this test
    module (capacities + teams).

    ``get_all_teams`` returns the configured list of teams; ``get_team_by_id``
    looks up by ``team_id`` and returns ``None`` for unknown ids.
    """

    def __init__(
        self,
        *,
        teams: list[Team] | None = None,
        capacities: list[Capacity] | None = None,
    ) -> None:
        self._teams = list(teams or [])
        self._caps = list(capacities or [])

    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 ["Backend Engineering"]

    def get_all_competence_names(self) -> list[str]:
        return ["Python", "Java"]

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

    def get_all_capacities_with_competences(self):
        return list(self._caps)

    def get_recent_free_capacities(self, limit: int = 20):
        return list(self._caps)[:limit]

    def get_capacity_by_id(self, capacity_id):  # pragma: no cover
        return None

    def get_capacity_description(self, capacity_id):  # pragma: no cover
        return None

    def get_capacity_references(self, capacity_id):  # pragma: no cover
        return []

    def get_capacity_certificates(self, capacity_id):  # pragma: no cover
        return []

    def get_all_teams(self) -> list[Team]:
        return list(self._teams)

    def get_team_by_id(self, team_id) -> Team | None:
        for t in self._teams:
            if t.team_id == str(team_id):
                return t
        return None


def _result_to_text(result: Any) -> str:
    """Coerce a FastMCP ``call_tool`` result to a plain text string."""
    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:
            text = getattr(content[0], "text", None)
            if isinstance(text, str):
                return text
    if isinstance(result, str):
        return result
    return str(result)


async def _call_tool(srv: Any, name: str, args: dict[str, Any]) -> str:
    return _result_to_text(await srv.call_tool(name, args))


def _build_server(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path,
    *,
    teams: list[Team] | None = None,
    capacities: list[Capacity] | None = None,
):
    cfg = tmp_path / "config.toml"
    cfg.write_text(_CONFIG_TEMPLATE, encoding="utf-8")

    fake_db = _FakeDB(teams=teams, capacities=capacities)

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_a, **_k: fake_db)

    return build_server(str(cfg))


# ---------------------------------------------------------------------------
# Tool registration tests (Anforderungen 1.3, 9.1, 9.2)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_find_matching_teams_tool_is_registered(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``find_matching_teams`` must be registered with FastMCP."""
    srv = _build_server(monkeypatch, tmp_path, teams=[_make_team()])
    tool_names = {t.name for t in await srv.list_tools()}
    assert "find_matching_teams" in tool_names, (
        f"expected 'find_matching_teams' to be registered; got {sorted(tool_names)!r}"
    )


@pytest.mark.asyncio
async def test_list_teams_tool_is_registered(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``list_teams`` must be registered with FastMCP."""
    srv = _build_server(monkeypatch, tmp_path, teams=[_make_team()])
    tool_names = {t.name for t in await srv.list_tools()}
    assert "list_teams" in tool_names, (
        f"expected 'list_teams' to be registered; got {sorted(tool_names)!r}"
    )


@pytest.mark.asyncio
async def test_get_team_details_tool_is_registered(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``get_team_details`` must be registered with FastMCP."""
    srv = _build_server(monkeypatch, tmp_path, teams=[_make_team()])
    tool_names = {t.name for t in await srv.list_tools()}
    assert "get_team_details" in tool_names, (
        f"expected 'get_team_details' to be registered; got {sorted(tool_names)!r}"
    )


# ---------------------------------------------------------------------------
# Markdown layout tests (Anforderungen 9.1, 9.2, 9.4)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_teams_renders_markdown_table(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``list_teams`` must render a Markdown table with the spec headers."""
    teams = [_make_team("t1"), _make_team("t2")]
    srv = _build_server(monkeypatch, tmp_path, teams=teams)

    out = await _call_tool(srv, "list_teams", {"limit": 20})

    for header in (
        "Team Id",
        "Team Name",
        "Schwerpunkt",
        "Anzahl Kompetenzen",
        "Anzahl Referenzen",
    ):
        assert header in out, (
            f"missing header {header!r} in list_teams output:\n{out}"
        )

    # Both team_ids must show up in the body of the table.
    assert "t1" in out, f"missing team_id 't1' in list_teams output:\n{out}"
    assert "t2" in out, f"missing team_id 't2' in list_teams output:\n{out}"


@pytest.mark.asyncio
async def test_get_team_details_renders_full_layout(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``get_team_details`` must render summary table + all sections."""
    team = _make_team("t1")
    srv = _build_server(monkeypatch, tmp_path, teams=[team])

    out = await _call_tool(srv, "get_team_details", {"team_id": "t1"})

    # Summary table headers (same as list_teams).
    for header in (
        "Team Id",
        "Team Name",
        "Schwerpunkt",
        "Anzahl Kompetenzen",
        "Anzahl Referenzen",
    ):
        assert header in out, (
            f"missing summary header {header!r} in get_team_details output:\n{out}"
        )

    # All required sections.
    for section in (
        "## Über uns",
        "## Leistungen",
        "## Interessen",
        "## Kompetenzen",
        "## Referenzen",
        "## Next steps",
    ):
        assert section in out, (
            f"missing section {section!r} in get_team_details output:\n{out}"
        )

    # Top-competence marker (Anforderung 9.4 / 5.5).
    # The team has ``Python`` flagged top, ``Java`` not.
    assert "Python (Top)" in out, (
        "expected 'Python (Top)' marker in Kompetenzen section, "
        f"output was:\n{out}"
    )
    assert "Java" in out, f"expected 'Java' to appear in output:\n{out}"
    assert "Java (Top)" not in out, (
        "Non-top competence 'Java' must not be marked as Top, "
        f"output was:\n{out}"
    )

    # Bold partner_name in references (Anforderung 9.4 / 5.6).
    assert "**DB Cargo**" in out, (
        "expected partner_name to be bolded with ** ** in references, "
        f"output was:\n{out}"
    )
    # The reference without partner_name renders only the projects text.
    assert "Internal tooling" in out, (
        "expected partner-less reference projects text in output:\n" + out
    )


@pytest.mark.asyncio
async def test_get_team_details_unknown_team_id(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """``get_team_details`` must mention the unknown id in its error
    message (Anforderung 9.3)."""
    srv = _build_server(monkeypatch, tmp_path, teams=[_make_team("t1")])

    out = await _call_tool(srv, "get_team_details", {"team_id": "unknown"})

    assert "Team not found" in out, (
        f"expected 'Team not found' in error response, got:\n{out}"
    )
    assert "unknown" in out, (
        f"expected the requested team_id substring 'unknown' in error "
        f"response, got:\n{out}"
    )


# ---------------------------------------------------------------------------
# find_matching_capacities snapshot test (Anforderung 1.2)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_find_matching_capacities_output_structure_unchanged(
    monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
    """The existing ``find_matching_capacities`` tool keeps its Markdown
    output structure after the team-profile refactoring.

    This is a snapshot-style assertion on stable, contract-relevant
    substrings (``SEARCH_ID=``, ``META=``, ``## Summary``, the Category /
    Count summary header, and the per-category Results header). It is
    deliberately not byte-exact because per-line scores depend on the
    SimilarityEngine; the structural envelope is what callers depend on.
    """
    cap = Capacity(
        id=1,
        owner_name="Alice Example",
        role_name="Backend Engineering",
        role_level=None,
        begin_date=None,
        end_date=None,
        competences=["Python"],
    )
    srv = _build_server(
        monkeypatch,
        tmp_path,
        teams=[_make_team("t1")],
        capacities=[cap],
    )

    out = await _call_tool(
        srv,
        "find_matching_capacities",
        {
            "role_name": "Backend Engineering",
            "competences": ["Python"],
        },
    )

    # Stable contract markers that survive any Score/LLM refactoring.
    for marker in (
        "SEARCH_ID=",
        "META=",
        "## Summary",
        "| Category | Count |",
    ):
        assert marker in out, (
            f"missing structural marker {marker!r} in "
            f"find_matching_capacities output:\n{out}"
        )

    # The default-shown category section header must be present (one of
    # the five categories). The exact one depends on score thresholds
    # but at least one Results section header must appear.
    assert any(
        f"## {cat} Results" in out
        for cat in ("Top", "Good", "Partial", "Low", "Irrelevant")
    ), (
        "expected at least one '## <Category> Results' header in "
        f"find_matching_capacities output:\n{out}"
    )

    # The capacity-search column layout must include the legacy headers
    # (the team refactoring must not have leaked team headers in here).
    assert "| ID |" in out, (
        f"expected legacy capacity-search 'ID' column in output:\n{out}"
    )
    assert "Team Name" not in out, (
        "team-search column 'Team Name' must not appear in capacity-search "
        f"output:\n{out}"
    )