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.
234 lines
7.6 KiB
Python
234 lines
7.6 KiB
Python
"""SQL-shape snapshot tests for `TrinoClient` team queries.
|
||
|
||
These tests focus exclusively on the *string shape* of the production
|
||
SQL emitted by the team-related methods. They complement the more
|
||
behaviour-oriented tests in:
|
||
|
||
* ``tests/test_trino_client_team_master_data.py``
|
||
* ``tests/test_trino_client_team_competences.py``
|
||
* ``tests/test_trino_client_team_references.py``
|
||
|
||
by acting as a focused snapshot guard that ``INNER JOIN``,
|
||
``ORDER BY``, ``COALESCE`` and the ``LEFT JOIN`` against
|
||
``teamlandkarte_v_partners_latest`` remain correctly formulated.
|
||
|
||
Each test calls a method on a ``TrinoClient`` whose ``_cursor`` is
|
||
swapped for a fake context yielding canned rows, then inspects
|
||
``cur.executed[0][0]`` (whitespace-normalised) for the expected
|
||
substrings. Returning canned rows lets every method run end-to-end
|
||
without touching a real database.
|
||
|
||
Validates: Requirements 2.1, 2.2, 3.1, 4.1, 4.2.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
from teamlandkarte_mcp.database.trino_client import TrinoClient
|
||
|
||
|
||
class _FakeCursor:
|
||
"""Minimal cursor double recording executes and returning canned rows."""
|
||
|
||
def __init__(self, rows: list[tuple] | None = None) -> None:
|
||
self.executed: list[tuple[str, tuple[object, ...] | None]] = []
|
||
self._rows: list[tuple] = rows or []
|
||
|
||
def execute(self, sql: str, params=None) -> None: # noqa: ANN001
|
||
if params is None:
|
||
params_tuple: tuple[object, ...] | None = None
|
||
elif isinstance(params, tuple):
|
||
params_tuple = params
|
||
else:
|
||
params_tuple = tuple(params)
|
||
self.executed.append((sql, params_tuple))
|
||
|
||
def fetchone(self):
|
||
return self._rows[0] if self._rows else None
|
||
|
||
def fetchall(self):
|
||
return list(self._rows)
|
||
|
||
|
||
class _FakeCursorContext:
|
||
def __init__(self, cursor: _FakeCursor):
|
||
self._cursor = cursor
|
||
|
||
def __enter__(self) -> _FakeCursor:
|
||
return self._cursor
|
||
|
||
def __exit__(self, exc_type, exc, tb) -> None: # noqa: ANN001
|
||
return None
|
||
|
||
|
||
class _FakeConfig:
|
||
host = "x"
|
||
port = 1
|
||
username = "u"
|
||
password = "p"
|
||
http_scheme = "http"
|
||
verify_ssl = False
|
||
catalog = "c"
|
||
schema = "s"
|
||
pool_size = 1
|
||
|
||
|
||
def _mk_client(cur: _FakeCursor) -> TrinoClient:
|
||
client = TrinoClient(_FakeConfig())
|
||
client._cursor = lambda: _FakeCursorContext(cur) # type: ignore[method-assign]
|
||
return client
|
||
|
||
|
||
def _norm_sql(sql: str) -> str:
|
||
"""Collapse all whitespace runs into single spaces for stable matching."""
|
||
|
||
return re.sub(r"\s+", " ", sql).strip()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_all_teams – master-data SQL shape
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_all_teams_sql_shape_snapshot() -> None:
|
||
"""`get_all_teams` emits the documented master-data SQL.
|
||
|
||
Validates that the first executed query joins teams with the
|
||
organizational-units view via INNER JOIN, COALESCEs nullable text
|
||
columns and applies the deterministic ORDER BY documented in the
|
||
design.
|
||
"""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.get_all_teams()
|
||
|
||
assert cur.executed, "expected at least one query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
assert (
|
||
"INNER JOIN teamlandkarte_v_teammeter_organizational_units_latest"
|
||
in sql
|
||
)
|
||
assert "CAST(t.team_id AS VARCHAR) = CAST(ou.id AS VARCHAR)" in sql
|
||
assert "COALESCE(t.focus_name, '')" in sql
|
||
assert "COALESCE(t.about_us, '')" in sql
|
||
assert "COALESCE(t.offerings, '')" in sql
|
||
assert "COALESCE(t.interests, '')" in sql
|
||
assert "ORDER BY ou.name ASC, t.team_id ASC" in sql
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_team_by_id – same shape plus the team_id/ouid filter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_team_by_id_sql_shape_snapshot() -> None:
|
||
"""`get_team_by_id` reuses the master-data SQL plus a filter clause.
|
||
|
||
Validates that the same INNER JOIN / COALESCE / ORDER BY scaffolding
|
||
is in place and that the additional
|
||
``CAST(t.team_id AS VARCHAR) = ?`` and
|
||
``CAST(t.ouid AS VARCHAR) = ?`` filter terms are present.
|
||
"""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.get_team_by_id("T1")
|
||
|
||
assert cur.executed, "expected the lookup query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
# Same INNER JOIN scaffolding as get_all_teams.
|
||
assert (
|
||
"INNER JOIN teamlandkarte_v_teammeter_organizational_units_latest"
|
||
in sql
|
||
)
|
||
assert "CAST(t.team_id AS VARCHAR) = CAST(ou.id AS VARCHAR)" in sql
|
||
assert "COALESCE(t.focus_name, '')" in sql
|
||
assert "COALESCE(t.about_us, '')" in sql
|
||
assert "COALESCE(t.offerings, '')" in sql
|
||
assert "COALESCE(t.interests, '')" in sql
|
||
assert "ORDER BY ou.name ASC, t.team_id ASC" in sql
|
||
|
||
# Filter clause specific to the by-id lookup.
|
||
assert "CAST(t.team_id AS VARCHAR) = ?" in sql
|
||
assert "CAST(t.ouid AS VARCHAR) = ?" in sql
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_team_competences / batch_get_team_competences – competence join shape
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_team_competences_sql_shape_snapshot() -> None:
|
||
"""`get_team_competences` joins competences and COALESCEs `top_competency`."""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.get_team_competences("OU1")
|
||
|
||
assert cur.executed, "expected the competences query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
assert "JOIN teamlandkarte_v_competences_latest" in sql
|
||
assert "CAST(tc.competence_id AS VARCHAR) = CAST(c.id AS VARCHAR)" in sql
|
||
assert "COALESCE(tc.top_competency, FALSE)" in sql
|
||
|
||
|
||
def test_batch_get_team_competences_sql_shape_snapshot() -> None:
|
||
"""`batch_get_team_competences` keeps the same join shape."""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.batch_get_team_competences(["OU1", "OU2"])
|
||
|
||
assert cur.executed, "expected the batch competences query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
assert "JOIN teamlandkarte_v_competences_latest" in sql
|
||
assert "CAST(tc.competence_id AS VARCHAR) = CAST(c.id AS VARCHAR)" in sql
|
||
assert "COALESCE(tc.top_competency, FALSE)" in sql
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_team_references / batch_get_team_references – partner LEFT JOIN shape
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_team_references_sql_shape_snapshot() -> None:
|
||
"""`get_team_references` LEFT JOINs partners and COALESCEs `p.name`."""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.get_team_references("OU1")
|
||
|
||
assert cur.executed, "expected the references query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
assert "LEFT JOIN teamlandkarte_v_partners_latest" in sql
|
||
assert "CAST(r.partner_id AS VARCHAR) = CAST(p.id AS VARCHAR)" in sql
|
||
assert "COALESCE(p.name, '')" in sql
|
||
|
||
|
||
def test_batch_get_team_references_sql_shape_snapshot() -> None:
|
||
"""`batch_get_team_references` keeps the same partner LEFT JOIN shape."""
|
||
|
||
cur = _FakeCursor(rows=[])
|
||
client = _mk_client(cur)
|
||
|
||
client.batch_get_team_references(["OU1", "OU2"])
|
||
|
||
assert cur.executed, "expected the batch references query to be executed"
|
||
sql = _norm_sql(cur.executed[0][0])
|
||
|
||
assert "LEFT JOIN teamlandkarte_v_partners_latest" in sql
|
||
assert "CAST(r.partner_id AS VARCHAR) = CAST(p.id AS VARCHAR)" in sql
|
||
assert "COALESCE(p.name, '')" in sql
|