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

from __future__ import annotations

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

import pytest

from teamlandkarte_mcp.mcp_server import build_server


@dataclass
class FakeTask:
    id: str
    name: str | None = None
    title: str = ""
    description: str = ""
    created_date: datetime = datetime(2026, 1, 1)
    start_date: date | None = None
    end_date: date | None = None
    skills: list[str] = None  # type: ignore[assignment]

    def __post_init__(self):
        if self.skills is None:
            self.skills = []


class FakeDBClient:
    def __init__(self, task: FakeTask):
        self._task = task

    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_open_tasks(self, limit: int = 20):
        return [self._task]

    def get_task_by_id(self, task_id: str):
        if task_id != self._task.id:
            return None
        return self._task

    def get_task_by_name(self, name: str):
        if not name:
            return None
        if getattr(self._task, "name", None) == name:
            return self._task
        return None

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

    def get_recent_free_capacities(self, limit: int = 20):  # pragma: no cover
        return []

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

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

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


def _result_to_text(result: Any) -> str:
    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
    return str(result)


def _call_tool(srv: Any, name: str, args: dict[str, Any] | None = None) -> str:
    if args is None:
        args = {}
    import asyncio

    return _result_to_text(asyncio.run(srv.call_tool(name, args)))


def test_infer_primary_role_table_output(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",
    )

    task = FakeTask(
        id="t1",
        title="Task",
        description="We need AWS and Kubernetes work.",
        created_date=datetime(2026, 1, 1, 10, 0, 0),
        start_date=None,
        end_date=None,
        skills=["AWS"],
    )

    import teamlandkarte_mcp.mcp_server as mod

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

    srv = build_server(str(cfg))
    out = _call_tool(
        srv,
        "infer_primary_role",
        {"task_text": task.description},
    )

    assert "| Role | Similarity |" in out


@pytest.mark.asyncio
async def test_get_task_details_table_first(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",
    )

    task = FakeTask(
        id="t1",
        title="Title",
        description="Desc",
        created_date=datetime(2026, 1, 2, 12, 0, 0),
        start_date=date(2026, 2, 1),
        end_date=date(2026, 2, 28),
        skills=["AWS", "Kubernetes"],
    )

    import teamlandkarte_mcp.mcp_server as mod

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

    srv = build_server(str(cfg))
    content, structured = await srv.call_tool(
        "get_task_details",
        {"task_id": "t1"},
    )
    out = _result_to_text((content, structured))

    assert out.lstrip().startswith("|")
    assert "| Task ID |" in out
    assert "## Description" in out


@pytest.mark.asyncio
async def test_validate_task_requirements_single_table(
    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",
    )

    task = FakeTask(
        id="t1",
        title="Title",
        description="We need AWS.",
        created_date=datetime(2026, 1, 2, 12, 0, 0),
        start_date=None,
        end_date=None,
        skills=["AWS"],
    )

    import teamlandkarte_mcp.mcp_server as mod

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

    srv = build_server(str(cfg))
    content, structured = await srv.call_tool(
        "validate_task_requirements", {"task_id": "t1"}
    )
    out = _result_to_text((content, structured))

    assert "| Task ID |" in out
    assert "| Time range (DB) |" in out
    assert "## Summary" in out
    assert "## Inferred Primary Role" in out
    assert "## Inferred Competences" in out