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.
154 lines
4.0 KiB
Python
154 lines
4.0 KiB
Python
import re
|
|
|
|
import pytest
|
|
|
|
from teamlandkarte_mcp.database.trino_client import TrinoClient
|
|
|
|
|
|
class _FakeCursor:
|
|
def __init__(self) -> None:
|
|
self.executed: list[tuple[str, tuple[object, ...] | None]] = []
|
|
|
|
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:
|
|
# Trino DB-API accepts sequences; tests only need tuple form.
|
|
params_tuple = tuple(params)
|
|
self.executed.append((sql, params_tuple))
|
|
|
|
def fetchone(self): # pragma: no cover
|
|
return (1,)
|
|
|
|
def fetchall(self): # pragma: no cover
|
|
return []
|
|
|
|
|
|
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())
|
|
|
|
# Avoid connecting in tests; ensure all methods use our fake cursor.
|
|
client._cursor = lambda: _FakeCursorContext(cur) # type: ignore[method-assign]
|
|
return client
|
|
|
|
|
|
def _norm_sql(sql: str) -> str:
|
|
sql = re.sub(r"\s+", " ", sql).strip()
|
|
return sql
|
|
|
|
|
|
def test_get_recent_free_capacities_query_ordering_and_limit() -> None:
|
|
cur = _FakeCursor()
|
|
c = _mk_client(cur)
|
|
|
|
c.get_recent_free_capacities(limit=7)
|
|
|
|
assert cur.executed, "Expected query to execute"
|
|
sql, params = cur.executed[-1]
|
|
sql_n = _norm_sql(sql)
|
|
|
|
assert "FROM teamlandkarte_v_capacities_latest" in sql_n
|
|
assert "WHERE cap.deletion_reason IS NULL" in sql_n
|
|
assert "ORDER BY cap.creation_date DESC" in sql_n
|
|
assert "LIMIT ?" in sql_n
|
|
assert params == (7,)
|
|
|
|
|
|
def test_get_capacity_by_id_query_filters_and_ordering() -> None:
|
|
cur = _FakeCursor()
|
|
c = _mk_client(cur)
|
|
|
|
c.get_capacity_by_id(123)
|
|
|
|
assert cur.executed
|
|
sql, params = cur.executed[-1]
|
|
sql_n = _norm_sql(sql)
|
|
|
|
assert "FROM teamlandkarte_v_capacities_latest" in sql_n
|
|
assert "LEFT JOIN teamlandkarte_v_capacity_competences_latest" in sql_n
|
|
assert "LEFT JOIN teamlandkarte_v_competences_latest" in sql_n
|
|
assert "cap.deletion_reason IS NULL" in sql_n
|
|
assert "cap.id = ?" in sql_n
|
|
assert "ORDER BY comp.name" in sql_n
|
|
assert params == (123,)
|
|
|
|
|
|
def test_get_all_role_names_query_filters_and_ordering() -> None:
|
|
cur = _FakeCursor()
|
|
c = _mk_client(cur)
|
|
|
|
c.get_all_role_names()
|
|
|
|
assert cur.executed
|
|
sql, params = cur.executed[-1]
|
|
sql_n = _norm_sql(sql)
|
|
|
|
assert "FROM teamlandkarte_v_capacity_roles_latest" in sql_n
|
|
assert "WHERE active = true" in sql_n
|
|
assert "AND staffing_board_relevant = true" in sql_n
|
|
assert "ORDER BY name" in sql_n
|
|
assert params is None
|
|
|
|
|
|
def test_get_all_competence_names_query_filters_and_ordering() -> None:
|
|
cur = _FakeCursor()
|
|
c = _mk_client(cur)
|
|
|
|
c.get_all_competence_names()
|
|
|
|
assert cur.executed
|
|
sql, params = cur.executed[-1]
|
|
sql_n = _norm_sql(sql)
|
|
|
|
assert "FROM beschaffungstool_kmp_skill_latest" in sql_n
|
|
assert "WHERE skillname__c IS NOT NULL" in sql_n
|
|
assert "ORDER BY skillname__c" in sql_n
|
|
assert params is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"method,args",
|
|
[
|
|
("get_recent_free_capacities", (1,)),
|
|
("get_capacity_by_id", ("1",)),
|
|
("get_all_role_names", ()),
|
|
("get_all_competence_names", ()),
|
|
],
|
|
)
|
|
def test_phase2_queries_are_select_only(method: str, args: tuple[object, ...]) -> None:
|
|
cur = _FakeCursor()
|
|
c = _mk_client(cur)
|
|
|
|
getattr(c, method)(*args)
|
|
|
|
assert cur.executed
|
|
sql, _params = cur.executed[-1]
|
|
assert sql.strip().upper().startswith("SELECT") or sql.strip().upper().startswith(
|
|
"WITH"
|
|
)
|