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

# Feature: llm-fulltext-matching, Property 9: matching_method-Validierung lehnt unbekannte Werte ab
"""Property test for ``matching_method`` validation.

The MCP tools ``find_matching_capacities`` and ``find_matching_tasks``
must reject any ``matching_method`` value whose ``strip().lower()``
representation is neither ``"score"`` nor ``"llm_fulltext"``. The
rejection must:

1. Produce an error message that mentions both allowed values
   (``score`` and ``llm_fulltext``).
2. Avoid touching the database (no capacity loading) and the LLM (no
   ``chat_completion`` call). The validation must short-circuit BEFORE
   any side-effect.

The fake database and the fake ``chat_completion`` track call counts;
both counters must remain zero for every Hypothesis-generated invalid
input.

**Validates: Requirements 1.5**
"""

from __future__ import annotations

from typing import Any

import pytest
from hypothesis import HealthCheck, given, settings
from hypothesis import strategies as st

import teamlandkarte_mcp.mcp_server as mcp_mod
from teamlandkarte_mcp.mcp_server import build_server


_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

[cache]
db_ttl_hours=1
search_ttl_minutes=60
max_size=100

[azure_openai]
endpoint='https://example.openai.azure.com'
api_version='2024-02-15-preview'
chat_deployment='gpt-4'
""".strip()


class _CountingDB:
    """Fake DB that counts how often ``get_all_capacities_with_competences``
    is invoked. Any non-zero counter after a rejected validation is a bug.
    """

    def __init__(self) -> None:
        self.capacity_calls = 0
        self.capacity_by_id_calls = 0

    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 ["X"]

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

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

    def get_all_capacities_with_competences(self):
        self.capacity_calls += 1
        return []

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

    def get_capacity_by_id(self, capacity_id):  # pragma: no cover
        self.capacity_by_id_calls += 1
        return None


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)


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_test_server(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path,
) -> tuple[Any, _CountingDB, dict[str, int]]:
    """Construct a server backed by counting DB and LLM doubles."""
    cfg = tmp_path / "config.toml"
    cfg.write_text(_CONFIG_TEMPLATE, encoding="utf-8")

    fake_db = _CountingDB()
    monkeypatch.setattr(
        mcp_mod, "create_db_client", lambda *_a, **_k: fake_db
    )

    llm_calls = {"count": 0}

    async def _fake_chat_completion(self, system, user):  # noqa: ARG001
        llm_calls["count"] += 1
        return '{"category":"Top","rationale":"r"}'

    monkeypatch.setattr(
        "teamlandkarte_mcp.azure.openai_client.AzureOpenAIClient.chat_completion",
        _fake_chat_completion,
    )

    monkeypatch.setenv("DATA_LAKE_USERNAME", "u")
    monkeypatch.setenv("DATA_LAKE_PASSWORD", "p")
    monkeypatch.setenv("AZURE_OPENAI_LLM_API_KEY", "k")

    srv = build_server(str(cfg))
    return srv, fake_db, llm_calls


# Reject anything that, after trimming and lower-casing, is one of the
# canonical values (those are valid inputs and would not exercise the
# error path) or the empty string. The empty/whitespace-only string is
# treated by the server as "no value provided" (it then falls back to
# the configured default), so it is not part of the invalid-input space
# this property test targets. Hypothesis is otherwise free to generate
# any text.
def _is_invalid_method(value: str) -> bool:
    norm = value.strip().lower()
    if norm == "":
        return False
    return norm not in ("score", "llm_fulltext")


_INVALID_METHOD = st.text(max_size=20).filter(_is_invalid_method)


@settings(
    max_examples=50,
    deadline=None,
    suppress_health_check=[HealthCheck.function_scoped_fixture],
)
@given(value=_INVALID_METHOD)
@pytest.mark.asyncio
async def test_invalid_matching_method_is_rejected_without_side_effects(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path,
    value: str,
) -> None:
    """Property 9: invalid matching_method values are rejected verbatim.

    The error message must list both allowed values; the database and the
    LLM mock must not be touched.
    """
    srv, fake_db, llm_calls = _build_test_server(monkeypatch, tmp_path)

    res = await _call_tool(
        srv,
        "find_matching_capacities",
        {
            "role_name": "X",
            "competences": ["A"],
            "matching_method": value,
        },
    )

    # The error message references both allowed values.
    assert "score" in res, (
        f"missing 'score' in error response for {value!r}: {res!r}"
    )
    assert "llm_fulltext" in res, (
        f"missing 'llm_fulltext' in error response for {value!r}: {res!r}"
    )

    # The response must signal an error (not a successful search).
    lowered = res.lower()
    assert ("error" in lowered) or ("invalid" in lowered), (
        f"response is not an error for {value!r}: {res!r}"
    )

    # No DB/LLM access for the rejected validation path.
    assert fake_db.capacity_calls == 0, (
        f"DB capacity loader was invoked for invalid value {value!r}"
    )
    assert llm_calls["count"] == 0, (
        f"LLM was invoked for invalid value {value!r}"
    )