mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-09 15:58:30 +00:00
116291a3b6
Rename the import package surfsense_mcp -> mcp_server and remove the src/ layer so the project mirrors the backend's shape (project folder != package name, e.g. surfsense_backend/app). Kills the redundant surfsense_mcp/src/surfsense_mcp nesting. Distribution name and console command (surfsense-mcp) are unchanged; only python -m and internal import paths move to mcp_server. Dockerfile CMD updated; no PYTHONPATH added since the editable install already makes the package importable.
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""Workspace resolution: names, ids, defaults, and the ambiguous cases."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from mcp_server.core.auth import identity
|
|
from mcp_server.core.errors import ToolError
|
|
from mcp_server.core.workspace_context import WorkspaceContext
|
|
|
|
|
|
class FakeClient:
|
|
"""Stands in for SurfSenseClient, serving a fixed workspace list."""
|
|
|
|
def __init__(self, rows: list[dict]) -> None:
|
|
self._rows = rows
|
|
|
|
async def request(self, method: str, path: str, **_kwargs):
|
|
assert (method, path) == ("GET", "/workspaces")
|
|
return self._rows
|
|
|
|
|
|
def _rows(*names_ids: tuple[str, int]) -> list[dict]:
|
|
return [{"id": wid, "name": name} for name, wid in names_ids]
|
|
|
|
|
|
def _context(rows: list[dict], preferred: str | None = None) -> WorkspaceContext:
|
|
return WorkspaceContext(FakeClient(rows), preferred_reference=preferred)
|
|
|
|
|
|
def test_resolves_exact_name():
|
|
ctx = _context(_rows(("Research", 1), ("Marketing", 2)))
|
|
assert asyncio.run(ctx.resolve("Research")).id == 1
|
|
|
|
|
|
def test_resolves_case_insensitively():
|
|
ctx = _context(_rows(("Research", 1)))
|
|
assert asyncio.run(ctx.resolve("research")).id == 1
|
|
|
|
|
|
def test_resolves_unique_substring():
|
|
ctx = _context(_rows(("Research Space", 1), ("Marketing", 2)))
|
|
assert asyncio.run(ctx.resolve("resea")).id == 1
|
|
|
|
|
|
def test_ambiguous_substring_is_rejected():
|
|
ctx = _context(_rows(("Research A", 1), ("Research B", 2)))
|
|
with pytest.raises(ToolError):
|
|
asyncio.run(ctx.resolve("research"))
|
|
|
|
|
|
def test_resolves_by_numeric_id():
|
|
ctx = _context(_rows(("Research", 1), ("Marketing", 2)))
|
|
assert asyncio.run(ctx.resolve(2)).name == "Marketing"
|
|
assert asyncio.run(ctx.resolve("2")).name == "Marketing"
|
|
|
|
|
|
def test_unknown_id_is_rejected():
|
|
ctx = _context(_rows(("Research", 1)))
|
|
with pytest.raises(ToolError):
|
|
asyncio.run(ctx.resolve(99))
|
|
|
|
|
|
def test_unknown_name_is_rejected():
|
|
ctx = _context(_rows(("Research", 1)))
|
|
with pytest.raises(ToolError):
|
|
asyncio.run(ctx.resolve("Nope"))
|
|
|
|
|
|
def test_default_auto_selects_single_workspace():
|
|
ctx = _context(_rows(("Only", 7)))
|
|
assert asyncio.run(ctx.resolve(None)).id == 7
|
|
|
|
|
|
def test_default_with_multiple_requires_a_choice():
|
|
ctx = _context(_rows(("A", 1), ("B", 2)))
|
|
with pytest.raises(ToolError):
|
|
asyncio.run(ctx.resolve(None))
|
|
|
|
|
|
def test_default_with_no_workspaces_is_rejected():
|
|
ctx = _context([])
|
|
with pytest.raises(ToolError):
|
|
asyncio.run(ctx.resolve(None))
|
|
|
|
|
|
def test_default_uses_preferred_reference():
|
|
ctx = _context(_rows(("A", 1), ("Research", 2)), preferred="Research")
|
|
assert asyncio.run(ctx.resolve(None)).id == 2
|
|
|
|
|
|
def test_resolution_is_remembered_as_active():
|
|
ctx = _context(_rows(("A", 1), ("B", 2)))
|
|
asyncio.run(ctx.resolve("B"))
|
|
assert ctx.active is not None and ctx.active.id == 2
|
|
# a later default call reuses the active selection without re-choosing
|
|
assert asyncio.run(ctx.resolve(None)).id == 2
|
|
|
|
|
|
def test_active_workspace_is_isolated_per_identity():
|
|
ctx = _context(_rows(("A", 1), ("B", 2)))
|
|
|
|
async def select_as(key: str, reference: str) -> None:
|
|
token = identity.bind_api_key(key)
|
|
try:
|
|
await ctx.resolve(reference)
|
|
finally:
|
|
identity.unbind_api_key(token)
|
|
|
|
async def active_for(key: str) -> int | None:
|
|
token = identity.bind_api_key(key)
|
|
try:
|
|
return ctx.active.id if ctx.active else None
|
|
finally:
|
|
identity.unbind_api_key(token)
|
|
|
|
asyncio.run(select_as("ss_pat_A", "A"))
|
|
asyncio.run(select_as("ss_pat_B", "B"))
|
|
|
|
# Each caller keeps its own selection; no bleed across identities.
|
|
assert asyncio.run(active_for("ss_pat_A")) == 1
|
|
assert asyncio.run(active_for("ss_pat_B")) == 2
|
|
# An unknown caller has no active selection.
|
|
assert asyncio.run(active_for("ss_pat_C")) is None
|