Files
Orchestrator/bahn/teamlandkarte-mcp/tests/test_pbt_similarity.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

107 lines
3.8 KiB
Python

# Feature: remove-embedding-competence-similarity, Property 1: BM25 Zero-Score für fehlenden Token-Overlap
"""Property-based tests for SimilarityEngine (BM25+RRF competence matching)."""
from __future__ import annotations
import re
import pytest
from hypothesis import given, settings
from hypothesis import strategies as st
from teamlandkarte_mcp.matching.bm25 import Bm25Index
from teamlandkarte_mcp.matching.similarity import SimilarityEngine
class _DummyClient:
"""Minimal stub for AzureOpenAIClient (unused for competence similarity)."""
async def chat_completion(self, system: str, user: str) -> str:
return '{"similarity": 0.5}'
def _bm25_engine() -> SimilarityEngine:
return SimilarityEngine(client=_DummyClient()) # type: ignore[arg-type]
def _tokenize(text: str) -> set[str]:
"""Mirror the BM25 tokenizer: lowercase + split on [\\W_]+."""
return {t for t in re.split(r"[\W_]+", text.lower()) if t}
# Strategy: generate competence strings using ASCII letters only (min 3 chars).
# We restrict to ASCII to avoid Unicode casefolding edge-cases (e.g. İ → i̇)
# that produce unexpected token splits and break the prefix-disjointness guarantee.
_competence_str = st.text(
alphabet=st.characters(whitelist_categories=("L",), max_codepoint=127),
min_size=3,
max_size=30,
)
# Strategy: generate lists of competence strings (1-5 items)
_competence_list = st.lists(_competence_str, min_size=1, max_size=5)
def _make_disjoint_pair(
required: list[str], candidate: list[str]
) -> tuple[list[str], list[str]]:
"""Prefix-separate required and candidate to guarantee disjoint tokens.
Required competences get prefixed with 'req' and candidate competences
get prefixed with 'cand'. Since both prefixes are pure letters and
differ, and the original strings are pure letters, the resulting token
sets are guaranteed to be disjoint after the BM25 tokenizer processes them.
"""
prefixed_req = [f"req{s}" for s in required]
prefixed_cand = [f"cand{s}" for s in candidate]
return prefixed_req, prefixed_cand
# **Validates: Requirements 1.1**
@settings(max_examples=100)
@given(required=_competence_list, candidate=_competence_list)
@pytest.mark.asyncio
async def test_bm25_zero_score_on_disjoint_tokens(
required: list[str], candidate: list[str]
) -> None:
"""Property 1: BM25 returns 0.0 for all required competences when there is
no token overlap with any candidate competence.
For any set of required competences and candidate competences where no
candidate shares any token with a required competence,
compute_competence_similarity shall return a score of 0.0 for that
required competence.
"""
# Ensure disjoint tokens via prefix separation
disjoint_req, disjoint_cand = _make_disjoint_pair(required, candidate)
# Verify our disjoint guarantee holds (sanity check on the strategy)
req_tokens = set()
for r in disjoint_req:
req_tokens.update(_tokenize(r))
cand_tokens = set()
for c in disjoint_cand:
cand_tokens.update(_tokenize(c))
assert req_tokens.isdisjoint(cand_tokens), (
f"Tokens not disjoint: {req_tokens & cand_tokens}"
)
# Build a global BM25 index from the candidate competences (realistic usage)
global_index = Bm25Index(corpus=disjoint_cand)
engine = _bm25_engine()
result = await engine.compute_competence_similarity(
required=disjoint_req,
candidate=disjoint_cand,
global_index=global_index,
)
# All required competences must have score 0.0
for req_competence in disjoint_req:
assert req_competence in result, (
f"Missing key {req_competence!r} in result"
)
score = float(result[req_competence]["score"]) # type: ignore[arg-type]
assert score == 0.0, (
f"Expected 0.0 for {req_competence!r} but got {score}"
)