ankn
a5f8fb49ab
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.
144 lines
4.5 KiB
Python
144 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from teamlandkarte_mcp.cache.search_cache import SearchCache
|
|
|
|
|
|
def test_search_cache_pagination_slice() -> None:
|
|
"""Sanity-check pagination math used by get_results_by_category.
|
|
|
|
This test does not execute the MCP tool. It validates the expected slice
|
|
behavior that the tool should apply: 1-based pages and stable page sizes.
|
|
"""
|
|
|
|
cache = SearchCache(ttl_minutes=60, max_size=10)
|
|
search_id = cache.store_search(
|
|
task_id=None,
|
|
requirements={"role_name": "X", "competences": ["A"]},
|
|
results={"Top": list(range(1, 51))},
|
|
)
|
|
|
|
entry = cache.get(search_id)
|
|
assert entry is not None
|
|
|
|
items = entry.results["Top"]
|
|
|
|
page_size = 20
|
|
page1 = items[(1 - 1) * page_size : 1 * page_size]
|
|
page2 = items[(2 - 1) * page_size : 2 * page_size]
|
|
page3 = items[(3 - 1) * page_size : 3 * page_size]
|
|
|
|
assert page1 == list(range(1, 21))
|
|
assert page2 == list(range(21, 41))
|
|
assert page3 == list(range(41, 51))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"page,page_size,expected",
|
|
[
|
|
(1, 10, list(range(1, 11))),
|
|
(2, 10, list(range(11, 21))),
|
|
(3, 10, list(range(21, 31))),
|
|
],
|
|
)
|
|
def test_pagination_param_matrix(
|
|
page: int, page_size: int, expected: list[int]
|
|
) -> None:
|
|
items = list(range(1, 31))
|
|
got = items[(page - 1) * page_size : page * page_size]
|
|
assert got == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Team-search pagination coverage (Profile_Type="team")
|
|
# ---------------------------------------------------------------------------
|
|
#
|
|
# Pagination is search-type-agnostic: ``get_results_by_category``
|
|
# slices ``entry.results["by_category"][cat]`` regardless of whether
|
|
# the entry is a ``capacity_search`` or a ``team_search``. The tests
|
|
# below exercise the same slicing logic for ``team_search`` payloads
|
|
# in both ``matching_method`` flavours so the routing/pagination
|
|
# surface is covered for both ``Profile_Type`` values.
|
|
#
|
|
# _Requirements: 8.1, 8.4, 12.7_
|
|
|
|
|
|
def _team_item(idx: int, *, matching_method: str) -> dict:
|
|
"""Build a minimal cached team item for pagination assertions."""
|
|
base = {
|
|
"team_id": f"t{idx}",
|
|
"ouid": f"ou-{idx}",
|
|
"team_name": f"Team {idx}",
|
|
"focus_name": "Backend Developer",
|
|
"about_us": "",
|
|
"offerings": "",
|
|
"interests": "",
|
|
"competences": [],
|
|
"references": [],
|
|
"category": "Top",
|
|
}
|
|
if matching_method == "score":
|
|
return {
|
|
**base,
|
|
"competence_score": 1.0,
|
|
"role_score": 1.0,
|
|
"overall_score": 1.0,
|
|
}
|
|
return {**base, "rationale": f"rationale {idx}"}
|
|
|
|
|
|
@pytest.mark.parametrize("matching_method", ["score", "llm_fulltext"])
|
|
def test_team_search_pagination_slice(matching_method: str) -> None:
|
|
"""Slicing of a ``team_search`` ``by_category`` bucket is identical
|
|
to the capacity-search slicing math (1-based pages, stable size).
|
|
"""
|
|
cache = SearchCache(ttl_minutes=60, max_size=10)
|
|
items = [_team_item(i, matching_method=matching_method)
|
|
for i in range(1, 51)]
|
|
|
|
payload = {
|
|
"search_type": "team_search",
|
|
"matching_method": matching_method,
|
|
"reference": {
|
|
"availability_date_start": None,
|
|
"availability_date_end": None,
|
|
},
|
|
"summary": {"Top": len(items)},
|
|
"by_category": {
|
|
"Top": items,
|
|
"Good": [],
|
|
"Partial": [],
|
|
"Low": [],
|
|
"Irrelevant": [],
|
|
},
|
|
}
|
|
if matching_method == "llm_fulltext":
|
|
payload["errors"] = []
|
|
|
|
search_id = cache.store_search(
|
|
task_id=None,
|
|
requirements={"role_name": "Backend Developer",
|
|
"competences": ["python"]},
|
|
results=payload,
|
|
)
|
|
|
|
entry = cache.get(search_id)
|
|
assert entry is not None
|
|
assert entry.results["search_type"] == "team_search"
|
|
assert entry.results["matching_method"] == matching_method
|
|
|
|
bucket = entry.results["by_category"]["Top"]
|
|
|
|
page_size = 20
|
|
page1 = bucket[(1 - 1) * page_size : 1 * page_size]
|
|
page2 = bucket[(2 - 1) * page_size : 2 * page_size]
|
|
page3 = bucket[(3 - 1) * page_size : 3 * page_size]
|
|
|
|
assert [it["team_id"] for it in page1] == [f"t{i}" for i in range(1, 21)]
|
|
assert [it["team_id"] for it in page2] == [f"t{i}" for i in range(21, 41)]
|
|
assert [it["team_id"] for it in page3] == [f"t{i}" for i in range(41, 51)]
|
|
# Page 3 has the partial-tail (10 items) like the capacity case.
|
|
assert len(page3) == 10
|
|
|