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.
447 lines
14 KiB
Python
447 lines
14 KiB
Python
"""Property-based tests for the capacity details enrichment feature.
|
|
|
|
Uses Hypothesis to verify correctness properties across many random inputs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from hypothesis import given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from teamlandkarte_mcp.mcp_server import build_server
|
|
from teamlandkarte_mcp.models import Capacity, Task
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeDB:
|
|
"""Minimal fake DB client for property-based tests."""
|
|
|
|
def __init__(
|
|
self,
|
|
capacity: Capacity | None,
|
|
*,
|
|
description: str | None = None,
|
|
references: list[dict] | None = None,
|
|
certificates: list[str] | None = None,
|
|
) -> None:
|
|
self._capacity = capacity
|
|
self._description = description
|
|
self._references = references if references is not None else []
|
|
self._certificates = certificates if certificates is not None else []
|
|
# Track calls for Property 1
|
|
self.calls: dict[str, list] = {
|
|
"get_capacity_description": [],
|
|
"get_capacity_references": [],
|
|
"get_capacity_certificates": [],
|
|
}
|
|
|
|
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]:
|
|
return []
|
|
|
|
def get_task_by_id(self, task_id: str):
|
|
return None
|
|
|
|
def get_task_by_name(self, name: str):
|
|
return None
|
|
|
|
def get_all_capacities_with_competences(self) -> list[Capacity]:
|
|
return [self._capacity] if self._capacity else []
|
|
|
|
def get_recent_free_capacities(self, limit: int = 20) -> list[Capacity]:
|
|
return [self._capacity] if self._capacity else []
|
|
|
|
def get_capacity_by_id(self, capacity_id: int | str) -> Capacity | None:
|
|
if self._capacity and int(capacity_id) == self._capacity.id:
|
|
return self._capacity
|
|
return None
|
|
|
|
def get_capacity_description(self, capacity_id: int | str) -> str | None:
|
|
self.calls["get_capacity_description"].append(capacity_id)
|
|
return self._description
|
|
|
|
def get_capacity_references(self, capacity_id: int | str) -> list[dict]:
|
|
self.calls["get_capacity_references"].append(capacity_id)
|
|
return self._references
|
|
|
|
def get_capacity_certificates(self, capacity_id: int | str) -> list[str]:
|
|
self.calls["get_capacity_certificates"].append(capacity_id)
|
|
return self._certificates
|
|
|
|
|
|
def _make_config(tmp_path) -> str:
|
|
"""Write a minimal config.toml and return its path."""
|
|
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",
|
|
)
|
|
return str(cfg)
|
|
|
|
|
|
def _make_capacity(cap_id: int) -> Capacity:
|
|
return Capacity(
|
|
id=cap_id,
|
|
owner_name="Team Alpha",
|
|
role_name="Cloud Engineer",
|
|
role_level=None,
|
|
begin_date=date(2026, 1, 1),
|
|
end_date=date(2026, 6, 30),
|
|
competences=["AWS", "Terraform"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 1: All enrichment data is fetched for any capacity
|
|
# Feature: capacity-details-enrichment
|
|
# Validates: Requirements 1.1, 2.1, 3.1
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@settings(max_examples=100)
|
|
@given(capacity_id=st.integers(min_value=1, max_value=100_000))
|
|
async def test_all_enrichment_data_fetched(capacity_id: int, tmp_path_factory):
|
|
"""Property 1: All enrichment data is fetched for any capacity.
|
|
|
|
For any valid capacity ID that resolves to an existing capacity,
|
|
the tool SHALL invoke get_capacity_description, get_capacity_references,
|
|
and get_capacity_certificates with that capacity ID.
|
|
|
|
**Validates: Requirements 1.1, 2.1, 3.1**
|
|
"""
|
|
tmp_path = tmp_path_factory.mktemp("cfg")
|
|
cap = _make_capacity(capacity_id)
|
|
fake_db = _FakeDB(
|
|
cap,
|
|
description="Some description",
|
|
references=[{"partner_name": "P", "projects": "Proj"}],
|
|
certificates=["Cert A"],
|
|
)
|
|
|
|
import teamlandkarte_mcp.mcp_server as mod
|
|
|
|
with patch.object(mod, "create_db_client", return_value=fake_db):
|
|
srv = build_server(_make_config(tmp_path))
|
|
await srv.call_tool("get_capacity_details", {"capacity_id": capacity_id})
|
|
|
|
# Each enrichment method must be called exactly once with the correct ID
|
|
assert len(fake_db.calls["get_capacity_description"]) == 1
|
|
assert fake_db.calls["get_capacity_description"][0] == capacity_id
|
|
|
|
assert len(fake_db.calls["get_capacity_references"]) == 1
|
|
assert fake_db.calls["get_capacity_references"][0] == capacity_id
|
|
|
|
assert len(fake_db.calls["get_capacity_certificates"]) == 1
|
|
assert fake_db.calls["get_capacity_certificates"][0] == capacity_id
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 2: Non-empty data appears in output
|
|
# Feature: capacity-details-enrichment
|
|
# Validates: Requirements 1.2, 2.2, 3.2
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Strategies for generating test data
|
|
_description_st = st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=200,
|
|
).filter(lambda s: s.strip())
|
|
|
|
_reference_st = st.fixed_dictionaries({
|
|
"partner_name": st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=50,
|
|
).filter(lambda s: s.strip()),
|
|
"projects": st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=100,
|
|
).filter(lambda s: s.strip()),
|
|
})
|
|
|
|
_certificate_st = st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=100,
|
|
).filter(lambda s: s.strip())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@settings(max_examples=100)
|
|
@given(
|
|
description=_description_st,
|
|
references=st.lists(_reference_st, min_size=1, max_size=5),
|
|
certificates=st.lists(_certificate_st, min_size=1, max_size=5),
|
|
)
|
|
async def test_nonempty_data_appears_in_output(
|
|
description: str,
|
|
references: list[dict],
|
|
certificates: list[str],
|
|
tmp_path_factory,
|
|
):
|
|
"""Property 2: Non-empty data appears in output.
|
|
|
|
For any capacity with a non-empty description, a non-empty list of
|
|
references, and a non-empty list of certificates, the formatted output
|
|
SHALL contain the description text, every reference's projects text,
|
|
and every certificate string.
|
|
|
|
**Validates: Requirements 1.2, 2.2, 3.2**
|
|
"""
|
|
tmp_path = tmp_path_factory.mktemp("cfg")
|
|
cap = _make_capacity(1)
|
|
fake_db = _FakeDB(
|
|
cap,
|
|
description=description,
|
|
references=references,
|
|
certificates=certificates,
|
|
)
|
|
|
|
import teamlandkarte_mcp.mcp_server as mod
|
|
|
|
with patch.object(mod, "create_db_client", return_value=fake_db):
|
|
srv = build_server(_make_config(tmp_path))
|
|
content, structured = await srv.call_tool(
|
|
"get_capacity_details", {"capacity_id": 1}
|
|
)
|
|
out = (structured or {}).get("result") or getattr(
|
|
content[0], "text", ""
|
|
)
|
|
|
|
# Description must appear in output
|
|
assert description in out, (
|
|
f"Description not found in output: {description!r}"
|
|
)
|
|
|
|
# Every reference's projects text must appear
|
|
for ref in references:
|
|
assert ref["projects"] in out, (
|
|
f"Reference projects not found: {ref['projects']!r}"
|
|
)
|
|
|
|
# Every certificate must appear
|
|
for cert in certificates:
|
|
assert cert in out, f"Certificate not found: {cert!r}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 3: Section ordering is fixed
|
|
# Feature: capacity-details-enrichment
|
|
# Validates: Requirements 4.1, 4.2
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_optional_description_st = st.one_of(
|
|
st.none(),
|
|
st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=100,
|
|
).filter(lambda s: s.strip()),
|
|
)
|
|
|
|
_optional_references_st = st.lists(
|
|
st.fixed_dictionaries({
|
|
"partner_name": st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=0,
|
|
max_size=30,
|
|
),
|
|
"projects": st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=50,
|
|
).filter(lambda s: s.strip()),
|
|
}),
|
|
min_size=0,
|
|
max_size=5,
|
|
)
|
|
|
|
_optional_certificates_st = st.lists(
|
|
st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P", "Z"),
|
|
blacklist_characters="\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=50,
|
|
).filter(lambda s: s.strip()),
|
|
min_size=0,
|
|
max_size=5,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@settings(max_examples=100, deadline=None)
|
|
@given(
|
|
description=_optional_description_st,
|
|
references=_optional_references_st,
|
|
certificates=_optional_certificates_st,
|
|
)
|
|
async def test_section_ordering_is_fixed(
|
|
description: str | None,
|
|
references: list[dict],
|
|
certificates: list[str],
|
|
tmp_path_factory,
|
|
):
|
|
"""Property 3: Section ordering is fixed.
|
|
|
|
For any capacity (regardless of which fields are empty or populated),
|
|
the output string SHALL contain the section markers in the order:
|
|
capacity table first, then "Beschreibung", then "Referenzen", then
|
|
"Zertifizierungen", then "Next steps" — and each section SHALL be
|
|
separated by at least one blank line.
|
|
|
|
**Validates: Requirements 4.1, 4.2**
|
|
"""
|
|
tmp_path = tmp_path_factory.mktemp("cfg")
|
|
cap = _make_capacity(1)
|
|
fake_db = _FakeDB(
|
|
cap,
|
|
description=description,
|
|
references=references,
|
|
certificates=certificates,
|
|
)
|
|
|
|
import teamlandkarte_mcp.mcp_server as mod
|
|
|
|
with patch.object(mod, "create_db_client", return_value=fake_db):
|
|
srv = build_server(_make_config(tmp_path))
|
|
content, structured = await srv.call_tool(
|
|
"get_capacity_details", {"capacity_id": 1}
|
|
)
|
|
out = (structured or {}).get("result") or getattr(
|
|
content[0], "text", ""
|
|
)
|
|
|
|
# All section headers must be present
|
|
assert "## Beschreibung" in out
|
|
assert "## Referenzen" in out
|
|
assert "## Zertifizierungen" in out
|
|
assert "## Next steps" in out
|
|
|
|
# Section ordering must be fixed
|
|
idx_beschreibung = out.index("## Beschreibung")
|
|
idx_referenzen = out.index("## Referenzen")
|
|
idx_zertifizierungen = out.index("## Zertifizierungen")
|
|
idx_next_steps = out.index("## Next steps")
|
|
|
|
assert idx_beschreibung < idx_referenzen, (
|
|
"Beschreibung must come before Referenzen"
|
|
)
|
|
assert idx_referenzen < idx_zertifizierungen, (
|
|
"Referenzen must come before Zertifizierungen"
|
|
)
|
|
assert idx_zertifizierungen < idx_next_steps, (
|
|
"Zertifizierungen must come before Next steps"
|
|
)
|
|
|
|
# Capacity table must come before all enrichment sections
|
|
assert "| capacity_id |" in out
|
|
idx_table = out.index("| capacity_id |")
|
|
assert idx_table < idx_beschreibung, (
|
|
"Capacity table must come before Beschreibung"
|
|
)
|
|
|
|
# Each section must be separated by at least one blank line
|
|
# A blank line means two consecutive newlines (\n\n)
|
|
between_table_and_beschreibung = out[idx_table:idx_beschreibung]
|
|
assert "\n\n" in between_table_and_beschreibung, (
|
|
"Blank line required between table and Beschreibung"
|
|
)
|
|
|
|
between_beschreibung_and_referenzen = out[idx_beschreibung:idx_referenzen]
|
|
assert "\n\n" in between_beschreibung_and_referenzen, (
|
|
"Blank line required between Beschreibung and Referenzen"
|
|
)
|
|
|
|
between_referenzen_and_zertifizierungen = out[
|
|
idx_referenzen:idx_zertifizierungen
|
|
]
|
|
assert "\n\n" in between_referenzen_and_zertifizierungen, (
|
|
"Blank line required between Referenzen and Zertifizierungen"
|
|
)
|
|
|
|
between_zertifizierungen_and_next = out[idx_zertifizierungen:idx_next_steps]
|
|
assert "\n\n" in between_zertifizierungen_and_next, (
|
|
"Blank line required between Zertifizierungen and Next steps"
|
|
)
|