ankn/Orchestrator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2e35796f37348c6eb51c77475b59a383da947ab2
Orchestrator/bahn/teamlandkarte-mcp/tests/test_llm_fulltext_integration.py
T
ankn a5f8fb49ab Migrate all repos into monorepo context folders 
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
      Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
      Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)

Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
2026-06-30 20:39:52 +02:00

733 lines
22 KiB
Python
Raw Blame History

"""End-to-end integration tests for LLM-fulltext matching (tasks 12.2 and
12.3 of the ``llm-fulltext-matching`` spec).
These tests exercise ``find_matching_capacities`` and
``find_matching_tasks`` through ``build_server`` using fully mocked
``DBClient`` and ``AzureOpenAIClient`` collaborators. They cover:
* Task 12.2 — End-to-end runs in ``llm_fulltext`` mode for both
directions:
- The persisted ``SearchCache`` payload carries
``matching_method``, the **ungekürzte** rationale and an
``errors`` list.
- The availability prefilter behaves identically in score and
LLM-fulltext mode (out-of-window capacities are dropped).
* Task 12.3 — Markdown table headers are stable across modes:
- Score mode keeps the existing ``Role Score`` / ``Competence
Score`` / ``Overall Score`` columns.
- LLM mode ends with ``... | Category | Begründung |`` and
drops all numeric score columns.
_Requirements: 5.1, 5.2, 6.1, 6.2, 7.1, 7.4, 8.1, 8.2, 8.6, 9.5_
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any
import pytest
import teamlandkarte_mcp.mcp_server as mcp_mod
from teamlandkarte_mcp.azure.openai_client import AzureAPIError
from teamlandkarte_mcp.cache.search_cache import SearchCache
from teamlandkarte_mcp.mcp_server import build_server
from teamlandkarte_mcp.models import Capacity
# ---------------------------------------------------------------------------
# Test config and helpers
# ---------------------------------------------------------------------------
_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()
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))
# ---------------------------------------------------------------------------
# Fake DB
# ---------------------------------------------------------------------------
@dataclass
class _Task:
"""In-test stand-in for the production ``Task`` dataclass."""
id: str
title: str
description: str
skills: list[str] = field(default_factory=list)
name: str | None = None
created_date: datetime = datetime(2025, 1, 1)
start_date: date | None = None
end_date: date | None = None
class _FakeDB:
"""In-memory ``DBClient`` double used across the integration tests.
The fake supports both directions (capacity_search and task_search)
and exposes both the single-row and the batch capacity-fulltext
methods used by the LLM matcher.
"""
def __init__(
self,
*,
capacities: list[Capacity] | None = None,
tasks: list[Any] | None = None,
capacity_extras: dict[
str,
dict[str, Any],
]
| None = None,
) -> None:
self._capacities = capacities or []
self._tasks = tasks or []
# Map capacity_id (str) -> {description, certificates, references}.
self._extras = capacity_extras or {}
# ---- generic stubs used by build_server boot-time --------------------
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 list({c.role_name for c in self._capacities if c.role_name})
def get_all_competence_names(self) -> list[str]:
names: set[str] = set()
for c in self._capacities:
names.update(c.competences)
for t in self._tasks:
names.update(getattr(t, "skills", []) or [])
return sorted(names)
# ---- capacity / task readers ----------------------------------------
def get_all_capacities_with_competences(self) -> list[Capacity]:
return list(self._capacities)
def get_recent_free_capacities(self, limit: int = 20) -> list[Capacity]:
return []
def get_capacity_by_id(self, capacity_id: int | str) -> Capacity | None:
target = str(capacity_id)
for c in self._capacities:
if str(c.id) == target:
return c
return None
def get_open_tasks(self, limit: int = 20) -> list[Any]:
return list(self._tasks) if limit == 0 else list(self._tasks)[:limit]
def get_task_by_id(self, task_id: str): # pragma: no cover
return None
def get_task_by_name(self, name: str): # pragma: no cover
return None
# ---- capacity fulltext extras ---------------------------------------
def get_capacity_description(
self, capacity_id: int | str
) -> str | None:
return self._extras.get(str(capacity_id), {}).get("description")
def get_capacity_certificates(
self, capacity_id: int | str
) -> list[str]:
return list(
self._extras.get(str(capacity_id), {}).get("certificates") or []
)
def get_capacity_references(
self, capacity_id: int | str
) -> list[dict]:
return list(
self._extras.get(str(capacity_id), {}).get("references") or []
)
def batch_get_capacity_descriptions(
self, capacity_ids: list[Any]
) -> dict[str, str | None]:
return {
str(i): self._extras.get(str(i), {}).get("description")
for i in capacity_ids
}
def batch_get_capacity_certificates(
self, capacity_ids: list[Any]
) -> dict[str, list[str]]:
return {
str(i): list(
self._extras.get(str(i), {}).get("certificates") or []
)
for i in capacity_ids
}
def batch_get_capacity_references(
self, capacity_ids: list[Any]
) -> dict[str, list[dict]]:
return {
str(i): list(
self._extras.get(str(i), {}).get("references") or []
)
for i in capacity_ids
}
# ---------------------------------------------------------------------------
# Server builder
# ---------------------------------------------------------------------------
def _build_server_with(
monkeypatch: pytest.MonkeyPatch,
tmp_path,
*,
fake_db: _FakeDB,
chat_response: str | None = None,
chat_side_effect: Any = None,
) -> Any:
"""Build an MCP server backed by ``fake_db`` and a stubbed LLM.
Either ``chat_response`` (constant payload) or ``chat_side_effect``
(callable taking ``(system, user)`` and returning a string or
raising) must be provided.
"""
cfg = tmp_path / "config.toml"
cfg.write_text(_CONFIG_TEMPLATE, encoding="utf-8")
monkeypatch.setattr(
mcp_mod, "create_db_client", lambda *_a, **_k: fake_db
)
if chat_side_effect is not None:
async def _fake_chat(self, system, user): # noqa: ARG001
outcome = chat_side_effect(system, user)
if isinstance(outcome, BaseException):
raise outcome
return outcome
else:
async def _fake_chat(self, system, user): # noqa: ARG001
return chat_response or "{}"
monkeypatch.setattr(
"teamlandkarte_mcp.azure.openai_client."
"AzureOpenAIClient.chat_completion",
_fake_chat,
)
monkeypatch.setenv("DATA_LAKE_USERNAME", "u")
monkeypatch.setenv("DATA_LAKE_PASSWORD", "p")
monkeypatch.setenv("AZURE_OPENAI_LLM_API_KEY", "k")
return build_server(str(cfg))
def _patch_capturing_store(
monkeypatch: pytest.MonkeyPatch,
) -> list[dict[str, Any]]:
"""Patch ``SearchCache.store_search`` to capture its ``results``.
Returns a list that is appended to on every call (most recent
payload last). The original method is still invoked so ``search_id``
bookkeeping continues to work.
"""
captured: list[dict[str, Any]] = []
original = SearchCache.store_search
def _capturing(self, *, task_id, requirements, results):
captured.append(results)
return original(
self,
task_id=task_id,
requirements=requirements,
results=results,
)
monkeypatch.setattr(SearchCache, "store_search", _capturing)
return captured
# ---------------------------------------------------------------------------
# Common fixtures
# ---------------------------------------------------------------------------
def _two_capacities_one_in_window() -> list[Capacity]:
"""Capacity 1 overlaps the test window 2025-03-01..2025-06-30, 2 does not."""
return [
Capacity(
id=1,
owner_name="Alice",
role_name="X",
role_level=None,
begin_date=date(2025, 2, 1),
end_date=date(2025, 7, 31),
competences=["A"],
),
Capacity(
id=2,
owner_name="Bob",
role_name="X",
role_level=None,
begin_date=date(2026, 1, 1),
end_date=date(2026, 12, 31),
competences=["A"],
),
]
# ===========================================================================
# Task 12.2 — End-to-end integration tests
# ===========================================================================
@pytest.mark.asyncio
async def test_e2e_find_matching_capacities_llm_mode_caches_unshortened_rationale(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""LLM mode persists ``matching_method``, full rationale and errors list.
Validates: requirements 5.1, 7.4, 8.6, 9.5.
"""
long_rationale = "a" * 350
fake = _FakeDB(capacities=_two_capacities_one_in_window())
chat_payload = json.dumps(
{"category": "Top", "rationale": long_rationale},
ensure_ascii=False,
)
captured = _patch_capturing_store(monkeypatch)
srv = _build_server_with(
monkeypatch, tmp_path, fake_db=fake, chat_response=chat_payload
)
res = await _call_tool(
srv,
"find_matching_capacities",
{
"role_name": "X",
"competences": ["A"],
"date_start": "2025-03-01",
"date_end": "2025-06-30",
"matching_method": "llm_fulltext",
},
)
assert "SEARCH_ID=" in res, res
assert captured, "store_search was not called"
payload = captured[-1]
assert payload["matching_method"] == "llm_fulltext"
assert isinstance(payload.get("errors"), list)
top_items = payload["by_category"].get("Top", [])
assert len(top_items) == 1, (
f"only the in-window capacity should be categorised, got "
f"{[i.get('id') for i in top_items]}"
)
persisted_rationale = top_items[0]["rationale"]
assert len(persisted_rationale) >= 350, (
f"persisted rationale was truncated: len={len(persisted_rationale)}"
)
assert persisted_rationale == long_rationale
# Capacity 2 is outside the window and must not appear in any
# category (availability prefilter).
all_items = [
item
for cat_items in payload["by_category"].values()
for item in cat_items
]
ids = {str(it.get("id")) for it in all_items}
assert ids == {"1"}, f"expected only capacity 1, got {ids!r}"
@pytest.mark.asyncio
async def test_e2e_find_matching_capacities_availability_filter_same_in_both_modes(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""Score and LLM mode apply identical availability prefilters.
Validates: requirements 5.2, 9.5.
"""
fake = _FakeDB(capacities=_two_capacities_one_in_window())
# Single payload satisfies both consumers (similarity for score path,
# category/rationale for LLM path).
chat_payload = json.dumps(
{
"similarity": 1.0,
"category": "Top",
"rationale": "ok",
},
ensure_ascii=False,
)
captured = _patch_capturing_store(monkeypatch)
srv = _build_server_with(
monkeypatch, tmp_path, fake_db=fake, chat_response=chat_payload
)
args_common = {
"role_name": "X",
"competences": ["A"],
"date_start": "2025-03-01",
"date_end": "2025-06-30",
}
# Score-mode run.
score_res = await _call_tool(
srv,
"find_matching_capacities",
{**args_common, "matching_method": "score"},
)
assert "SEARCH_ID=" in score_res
score_payload = captured[-1]
assert score_payload["matching_method"] == "score"
score_ids = {
str(item.get("id"))
for cat in score_payload["by_category"].values()
for item in cat
}
# LLM-mode run.
llm_res = await _call_tool(
srv,
"find_matching_capacities",
{**args_common, "matching_method": "llm_fulltext"},
)
assert "SEARCH_ID=" in llm_res
llm_payload = captured[-1]
assert llm_payload["matching_method"] == "llm_fulltext"
llm_ids = {
str(item.get("id"))
for cat in llm_payload["by_category"].values()
for item in cat
}
assert score_ids == {"1"}, f"score-mode ids: {score_ids!r}"
assert llm_ids == {"1"}, f"llm-mode ids: {llm_ids!r}"
@pytest.mark.asyncio
async def test_e2e_find_matching_tasks_llm_mode_persists_errors_and_rationale(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""``find_matching_tasks`` persists errors list and full rationale.
The mock alternates between a successful payload (long rationale,
Good) and ``AzureAPIError`` so exactly one task is categorised and
the other ends up in ``errors``.
Validates: requirements 6.1, 6.2, 7.4, 8.6, 9.5.
"""
long_rationale = "b" * 360
capacity = Capacity(
id=1,
owner_name="Alice",
role_name="Frontend Developer",
role_level=None,
begin_date=date(2025, 4, 1),
end_date=date(2025, 8, 31),
competences=["React", "TypeScript"],
)
tasks = [
_Task(
id="T1",
title="Frontend Developer",
description="React and TypeScript",
skills=["React"],
),
_Task(
id="T2",
title="Backend Engineer",
description="Python",
skills=["Python"],
),
]
fake = _FakeDB(capacities=[capacity], tasks=tasks)
state = {"call": 0}
def _side_effect(_system: str, _user: str) -> Any:
n = state["call"]
state["call"] += 1
if n % 2 == 0:
return json.dumps(
{"category": "Good", "rationale": long_rationale},
ensure_ascii=False,
)
return AzureAPIError("LLM call failed")
captured = _patch_capturing_store(monkeypatch)
srv = _build_server_with(
monkeypatch,
tmp_path,
fake_db=fake,
chat_side_effect=_side_effect,
)
res = await _call_tool(
srv,
"find_matching_tasks",
{"capacity_id": 1, "matching_method": "llm_fulltext"},
)
assert "SEARCH_ID=" in res
assert captured, "store_search was not called"
payload = captured[-1]
assert payload["matching_method"] == "llm_fulltext"
good = payload["by_category"].get("Good", [])
assert len(good) == 1, (
f"expected exactly one Good item, got {len(good)}: {good!r}"
)
assert good[0]["rationale"] == long_rationale
assert len(good[0]["rationale"]) >= 360
errors = payload["errors"]
assert isinstance(errors, list)
assert len(errors) == 1, f"expected one error, got {errors!r}"
error_ids = {e["item_id"] for e in errors}
good_ids = {str(item.get("task_id") or item.get("id")) for item in good}
assert error_ids.isdisjoint(good_ids), (
"an item must appear in either by_category or errors, never both"
)
# ===========================================================================
# Task 12.3 — Markdown header snapshot tests
# ===========================================================================
@pytest.mark.asyncio
async def test_snapshot_capacity_search_score_mode_header(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""Score mode keeps the existing 9-column capacity table header.
Validates: requirement 7.1.
"""
fake = _FakeDB(capacities=_two_capacities_one_in_window())
srv = _build_server_with(
monkeypatch,
tmp_path,
fake_db=fake,
chat_response='{"similarity": 1.0}',
)
res = await _call_tool(
srv,
"find_matching_capacities",
{
"role_name": "X",
"competences": ["A"],
"date_start": "2025-03-01",
"date_end": "2025-06-30",
"matching_method": "score",
},
)
expected_header = (
"| ID | Owner | Role | Competences | Availability "
"| Role Score | Competence Score | Overall Score | Category |"
)
assert expected_header in res, res
assert "Begründung" not in res
@pytest.mark.asyncio
async def test_snapshot_capacity_search_llm_mode_header(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""LLM mode header ends with ``... | Category | Begründung |``.
Validates: requirements 8.1, 8.2.
"""
fake = _FakeDB(capacities=_two_capacities_one_in_window())
srv = _build_server_with(
monkeypatch,
tmp_path,
fake_db=fake,
chat_response='{"category":"Top","rationale":"ok"}',
)
res = await _call_tool(
srv,
"find_matching_capacities",
{
"role_name": "X",
"competences": ["A"],
"date_start": "2025-03-01",
"date_end": "2025-06-30",
"matching_method": "llm_fulltext",
},
)
assert "| Category | Begründung |" in res, res
assert "| Role Score |" not in res
assert "| Competence Score |" not in res
assert "| Overall Score |" not in res
@pytest.mark.asyncio
async def test_snapshot_task_search_score_mode_header(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""Score mode for ``find_matching_tasks`` keeps the score columns.
Validates: requirement 7.1.
"""
capacity = Capacity(
id=1,
owner_name="Alice",
role_name="Frontend Developer",
role_level=None,
begin_date=date(2025, 4, 1),
end_date=date(2025, 8, 31),
competences=["React"],
)
tasks = [
_Task(
id="T1",
title="Frontend Developer",
description="React",
skills=["React"],
start_date=date(2025, 5, 1),
end_date=date(2025, 7, 31),
)
]
fake = _FakeDB(capacities=[capacity], tasks=tasks)
srv = _build_server_with(
monkeypatch,
tmp_path,
fake_db=fake,
chat_response='{"similarity": 1.0}',
)
res = await _call_tool(
srv,
"find_matching_tasks",
{"capacity_id": 1, "matching_method": "score"},
)
assert (
"| Role Score | Competence Score | Overall Score | Category |"
in res
), res
assert "Begründung" not in res
@pytest.mark.asyncio
async def test_snapshot_task_search_llm_mode_header(
monkeypatch: pytest.MonkeyPatch, tmp_path
) -> None:
"""LLM mode for ``find_matching_tasks`` ends with Begründung header.
Validates: requirements 8.1, 8.2.
"""
capacity = Capacity(
id=1,
owner_name="Alice",
role_name="Frontend Developer",
role_level=None,
begin_date=date(2025, 4, 1),
end_date=date(2025, 8, 31),
competences=["React"],
)
tasks = [
_Task(
id="T1",
title="Frontend Developer",
description="React",
skills=["React"],
start_date=date(2025, 5, 1),
end_date=date(2025, 7, 31),
)
]
fake = _FakeDB(capacities=[capacity], tasks=tasks)
srv = _build_server_with(
monkeypatch,
tmp_path,
fake_db=fake,
chat_response='{"category":"Good","rationale":"ok"}',
)
res = await _call_tool(
srv,
"find_matching_tasks",
{"capacity_id": 1, "matching_method": "llm_fulltext"},
)
assert "| Category | Begründung |" in res, res
assert "| Role Score |" not in res
assert "| Competence Score |" not in res
assert "| Overall Score |" not in res
Reference in New Issue View Git Blame Copy Permalink