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

from __future__ import annotations

from datetime import datetime, date

import pytest

from teamlandkarte_mcp.mcp_server import build_server
from teamlandkarte_mcp.models import Capacity, Task


class _FakeDB:
    def __init__(self, caps: list[Capacity], tasks: list[Task]) -> None:
        self._caps = caps
        self._tasks = tasks

    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) -> list[Task]:
        if limit == 0:
            return list(self._tasks)
        return list(self._tasks)[:limit]

    def get_task_by_id(self, task_id: str):  # pragma: no cover
        for t in self._tasks:
            if t.id == task_id:
                return t
        return None

    def get_task_by_name(self, name: str):  # pragma: no cover
        if not name:
            return None
        for t in self._tasks:
            if getattr(t, "name", None) == name:
                return t
        return None

    def get_all_capacities_with_competences(self):  # pragma: no cover
        return list(self._caps)

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

    def get_capacity_by_id(self, capacity_id: int | str):
        cid = int(capacity_id)
        for c in self._caps:
            if int(c.id) == cid:
                return c
        return None

    def get_capacity_description(self, capacity_id: int | str) -> str | None:
        return None

    def get_capacity_references(self, capacity_id: int | str) -> list[dict]:
        return []

    def get_capacity_certificates(self, capacity_id: int | str) -> list[str]:
        return []


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

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: _FakeDB([], []))

    srv = build_server(str(cfg))
    content, structured = await srv.call_tool("list_free_capacities", {"limit": 5})
    out = (structured or {}).get("result") or getattr(content[0], "text", "")

    assert "| capacity_id |" in out


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

    cap = Capacity(
        id=1,
        owner_name="Team",
        role_name="Cloud Engineer",
        role_level=None,
        begin_date=date(2026, 1, 1),
        end_date=date(2026, 1, 31),
        competences=["AWS"],
    )

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: _FakeDB([cap], []))

    srv = build_server(str(cfg))
    content, structured = await srv.call_tool(
        "get_capacity_details", {"capacity_id": 1}
    )
    out = (structured or {}).get("result") or getattr(content[0], "text", "")

    assert "| capacity_id |" in out
    assert "## Next steps" in out
    assert "find_matching_tasks" in out


@pytest.mark.asyncio
async def test_find_matching_tasks_creates_search_id(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

[matching.inference]
max_competences=2
min_similarity=0.0

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

    cap = Capacity(
        id=1,
        owner_name="Team",
        role_name="Cloud Engineer",
        role_level=None,
        begin_date=None,
        end_date=None,
        competences=["AWS"],
    )

    task = Task(
        id="t1",
        name=None,
        title="Task",
        description="AWS work",
        start_date=None,
        end_date=None,
        created_date=datetime(2026, 1, 1, 10, 0, 0),
        skills=[],
    )

    import teamlandkarte_mcp.mcp_server as mod

    monkeypatch.setattr(mod, "create_db_client", lambda *_: _FakeDB([cap], [task]))

    srv = build_server(str(cfg))
    content, structured = await srv.call_tool("find_matching_tasks", {"capacity_id": 1})
    out = (structured or {}).get("result") or getattr(content[0], "text", "")

    assert "SEARCH_ID=" in out
    assert "## Summary" in out
    assert "| task_id |" in out