apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-06-06 20:15:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

test(agents): cover live filesystem middleware, retire dead twin

Browse source 
The single-agent-era filesystem middleware (app/agents/shared/middleware/
filesystem.py, ~2000 lines) was never instantiated in production, yet three
unit suites validated it — an illusory guardrail while the live decomposed
middleware (multi_agent_chat/middleware/shared/filesystem) was unguarded.

Close the gap before reorganizing the agents module:
- Add 14 integration tests driving live B's tools in desktop mode (real
  on-disk effects) and cloud mode (in-state staging, namespace policy).
- Port all high-value dead-twin assertions onto the live path: cloud rm/rmdir
  staging + guard rails, KBPostgresBackend delete-view filter, mode-scoped
  system prompt, cwd/relative/namespace resolution, multi-root mount
  normalization.
- Delete dead twin filesystem.py, drop its __init__ re-export, and retire its
  3 dead-twin tests.

Verified: test_import_all + middleware unit + FS integration all green.
This commit is contained in:
CREDO23 2026-06-04 17:46:49 +02:00
parent f3484f5a24
commit 1acde6a470
9 changed files with 960 additions and 2492 deletions
Download patch file Download diff file Expand all files Collapse all files

287
surfsense_backend/tests/unit/middleware/test_b_filesystem_path_resolution.py Normal file
View file

@ -0,0 +1,287 @@
"""Path/cwd/namespace + multi-root mount-normalization tests for LIVE filesystem.
Ported from the dead-twin suites:
* ``tests/unit/middleware/test_filesystem_middleware.py`` (cwd defaults,
relative resolution, cloud write-namespace policy)
* ``tests/unit/middleware/test_filesystem_verification.py`` (desktop
multi-root mount-prefix normalization)
Both exercised ``app.agents.shared.middleware.filesystem`` (dead). This drives
the production free functions in
``app.agents.multi_agent_chat.middleware.shared.filesystem.middleware`` instead.
The functions only touch ``mw._filesystem_mode`` and ``mw._get_backend`` so we
pass a lightweight fake ``mw`` rather than constructing the full middleware.
"""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import pytest
from app.agents.multi_agent_chat.middleware.shared.filesystem.middleware.mode import (
default_cwd,
)
from app.agents.multi_agent_chat.middleware.shared.filesystem.middleware.namespace_policy import (
check_cloud_write_namespace,
)
from app.agents.multi_agent_chat.middleware.shared.filesystem.middleware.path_resolution import (
current_cwd,
get_contract_suggested_path,
normalize_local_mount_path,
resolve_relative,
)
from app.agents.shared.filesystem_selection import FilesystemMode
from app.agents.shared.middleware.multi_root_local_folder_backend import (
MultiRootLocalFolderBackend,
)
pytestmark = pytest.mark.unit
def _mw(mode: FilesystemMode = FilesystemMode.CLOUD, backend=None):
return SimpleNamespace(_filesystem_mode=mode, _get_backend=lambda _rt: backend)
def _runtime(state: dict | None = None) -> SimpleNamespace:
return SimpleNamespace(state=state or {})
# ---------------------------------------------------------------------------
# cwd defaults
# ---------------------------------------------------------------------------
class TestCwdDefaults:
def test_default_cwd_in_cloud_is_documents_root(self):
assert default_cwd(FilesystemMode.CLOUD) == "/documents"
def test_default_cwd_in_desktop_is_root(self):
assert default_cwd(FilesystemMode.DESKTOP_LOCAL_FOLDER) == "/"
def test_current_cwd_uses_state_when_set(self):
assert (
current_cwd(_mw(), _runtime({"cwd": "/documents/notes"}))
== "/documents/notes"
)
def test_current_cwd_falls_back_to_default(self):
assert current_cwd(_mw(), _runtime({})) == "/documents"
def test_current_cwd_ignores_invalid(self):
assert current_cwd(_mw(), _runtime({"cwd": "not-absolute"})) == "/documents"
# ---------------------------------------------------------------------------
# relative resolution
# ---------------------------------------------------------------------------
class TestRelativePathResolution:
def test_relative_path_resolves_against_cwd(self):
assert (
resolve_relative(_mw(), "notes.md", _runtime({"cwd": "/documents/projects"}))
== "/documents/projects/notes.md"
)
def test_relative_path_with_dotdot(self):
assert (
resolve_relative(_mw(), "../c.md", _runtime({"cwd": "/documents/a/b"}))
== "/documents/a/c.md"
)
def test_absolute_path_is_kept(self):
assert (
resolve_relative(_mw(), "/other/x.md", _runtime({"cwd": "/documents"}))
== "/other/x.md"
)
def test_empty_path_returns_cwd(self):
assert (
resolve_relative(_mw(), "", _runtime({"cwd": "/documents/projects"}))
== "/documents/projects"
)
# ---------------------------------------------------------------------------
# contract suggested-path fallback
# ---------------------------------------------------------------------------
class TestContractSuggestedPath:
def test_falls_back_to_documents_notes_md_in_cloud(self):
suggested = get_contract_suggested_path(
_mw(FilesystemMode.CLOUD),
_runtime({"file_operation_contract": {}}),
)
assert suggested == "/documents/notes.md"
def test_falls_back_to_root_notes_md_in_desktop(self):
suggested = get_contract_suggested_path(
_mw(FilesystemMode.DESKTOP_LOCAL_FOLDER),
_runtime({"file_operation_contract": {}}),
)
assert suggested == "/notes.md"
# ---------------------------------------------------------------------------
# cloud write-namespace policy
# ---------------------------------------------------------------------------
class TestCloudWriteNamespacePolicy:
def test_documents_path_allowed(self):
assert (
check_cloud_write_namespace(_mw(), "/documents/foo.md", _runtime()) is None
)
def test_documents_root_allowed(self):
assert check_cloud_write_namespace(_mw(), "/documents", _runtime()) is None
def test_temp_basename_anywhere_allowed(self):
assert (
check_cloud_write_namespace(_mw(), "/temp_scratch.md", _runtime()) is None
)
assert check_cloud_write_namespace(_mw(), "/foo/temp_x.md", _runtime()) is None
assert (
check_cloud_write_namespace(_mw(), "/documents/temp_x.md", _runtime())
is None
)
def test_other_paths_rejected(self):
err = check_cloud_write_namespace(_mw(), "/foo/bar.md", _runtime())
assert err is not None
assert "must target /documents" in err
def test_anon_doc_path_is_read_only(self):
runtime = _runtime(
{
"kb_anon_doc": {
"path": "/documents/uploaded.xml",
"title": "uploaded",
"content": "",
"chunks": [],
}
}
)
err = check_cloud_write_namespace(_mw(), "/documents/uploaded.xml", runtime)
assert err is not None
assert "read-only" in err
def test_desktop_mode_skips_namespace_policy(self):
assert (
check_cloud_write_namespace(
_mw(FilesystemMode.DESKTOP_LOCAL_FOLDER), "/random/path.md", _runtime()
)
is None
)
# ---------------------------------------------------------------------------
# desktop multi-root mount normalization
# ---------------------------------------------------------------------------
def _desktop_mw(backend) -> SimpleNamespace:
return _mw(FilesystemMode.DESKTOP_LOCAL_FOLDER, backend)
class TestNormalizeLocalMountPath:
def test_prefixes_default_mount(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
"/random-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/random-note.md"
def test_keeps_explicit_mount(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
"/pc_backups/notes/random-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_windows_backslashes(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
r"\notes\random-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_normalizes_mixed_separators(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
r"\\notes//nested\\random-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/notes/nested/random-note.md"
def test_keeps_explicit_mount_with_backslashes(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
r"\pc_backups\notes\random-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_prefixes_posix_absolute_path(self, tmp_path: Path):
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
resolved = normalize_local_mount_path(
_desktop_mw(backend),
"/var/log/app.log",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/pc_backups/var/log/app.log"
def test_prefers_unique_existing_parent_mount(self, tmp_path: Path):
root_a = tmp_path / "RootA"
root_b = tmp_path / "RootB"
(root_a / "other").mkdir(parents=True)
(root_b / "nested" / "deep").mkdir(parents=True)
backend = MultiRootLocalFolderBackend(
(("root_a", str(root_a)), ("root_b", str(root_b)))
)
resolved = normalize_local_mount_path(
_desktop_mw(backend),
"/nested/deep/new-note.md",
_runtime({"file_operation_contract": {}}),
)
assert resolved == "/root_b/nested/deep/new-note.md"
def test_uses_suggested_mount_when_ambiguous(self, tmp_path: Path):
root_a = tmp_path / "RootA"
root_b = tmp_path / "RootB"
root_a.mkdir(parents=True)
root_b.mkdir(parents=True)
backend = MultiRootLocalFolderBackend(
(("root_a", str(root_a)), ("root_b", str(root_b)))
)
resolved = normalize_local_mount_path(
_desktop_mw(backend),
"/brand-new-note.md",
_runtime(
{"file_operation_contract": {"suggested_path": "/root_b/notes/context.md"}}
),
)
assert resolved == "/root_b/brand-new-note.md"

308
surfsense_backend/tests/unit/middleware/test_b_filesystem_rm_rmdir_cloud.py Normal file
View file

@ -0,0 +1,308 @@
"""Cloud-mode ``rm``/``rmdir`` staging tests for the LIVE filesystem middleware.
Ported from the former ``tests/unit/agents/new_chat/test_rm_rmdir_cloud.py``,
which exercised the *dead twin* ``app.agents.shared.middleware.filesystem``.
This drives the production decomposed tools
(``app.agents.multi_agent_chat.middleware.shared.filesystem``) instead: it
builds the real middleware via ``build_filesystem_mw``, pulls the real ``rm`` /
``rmdir`` tools off it, and invokes their coroutines with a stubbed
``KBPostgresBackend`` + runtime so we can assert the end-of-turn staging
payloads (``pending_deletes`` / ``pending_dir_deletes``) and the destructive-op
guard rails (root, /documents, anon doc, non-empty, cwd/ancestor, file vs dir).
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock
import pytest
from app.agents.multi_agent_chat.middleware.shared.filesystem import (
build_filesystem_mw,
)
from app.agents.shared.filesystem_backends import build_backend_resolver
from app.agents.shared.filesystem_selection import FilesystemMode, FilesystemSelection
from app.agents.shared.middleware.kb_postgres_backend import KBPostgresBackend
from app.agents.shared.state_reducers import _CLEAR
pytestmark = pytest.mark.unit
def _make_middleware(mode: FilesystemMode = FilesystemMode.CLOUD):
selection = FilesystemSelection(mode=mode)
resolver = build_backend_resolver(selection, search_space_id=1)
return build_filesystem_mw(
backend_resolver=resolver,
filesystem_mode=mode,
search_space_id=1,
user_id="00000000-0000-0000-0000-000000000001",
thread_id=1,
)
def _tool(mw, name: str):
return next(t for t in mw.tools if t.name == name)
def _runtime(state: dict[str, Any] | None = None, *, tool_call_id: str = "tc-abc"):
state = state or {}
state.setdefault("cwd", "/documents")
return SimpleNamespace(state=state, tool_call_id=tool_call_id)
class _KBBackendStub(KBPostgresBackend):
"""Construct-able ``KBPostgresBackend`` subclass for tests.
Bypasses the real ``__init__`` (which expects a runtime + DB session) and
injects only the async methods the rm/rmdir tools touch. The class
inheritance keeps the ``isinstance(backend, KBPostgresBackend)`` checks in
the tools on the cloud path.
"""
def __init__(self, *, children=None, file_data=None) -> None:
self.als_info = AsyncMock(return_value=children or [])
self._load_file_data = AsyncMock(
return_value=(file_data, 17) if file_data is not None else None
)
def _make_backend_stub(*, children=None, file_data=None) -> KBPostgresBackend:
return _KBBackendStub(children=children, file_data=file_data)
def _bind_backend(mw, backend):
mw._get_backend = lambda runtime: backend
return backend
# ---------------------------------------------------------------------------
# rm
# ---------------------------------------------------------------------------
class TestRmStaging:
@pytest.mark.asyncio
async def test_stages_delete_and_tombstones_state(self):
m = _make_middleware()
_bind_backend(m, _make_backend_stub(children=[], file_data={"content": ["x"]}))
runtime = _runtime(
{
"cwd": "/documents",
"files": {"/documents/notes.md": {"content": ["hello"]}},
"doc_id_by_path": {"/documents/notes.md": 17},
},
tool_call_id="tc-1",
)
result = await _tool(m, "rm").coroutine("/documents/notes.md", runtime=runtime)
assert hasattr(result, "update"), f"expected Command, got {result!r}"
update = result.update
assert update["pending_deletes"] == [
{"path": "/documents/notes.md", "tool_call_id": "tc-1"}
]
assert update["files"] == {"/documents/notes.md": None}
assert update["doc_id_by_path"] == {"/documents/notes.md": None}
@pytest.mark.asyncio
async def test_rejects_documents_root(self):
m = _make_middleware()
result = await _tool(m, "rm").coroutine("/documents", runtime=_runtime())
assert isinstance(result, str)
assert "refusing to rm" in result
@pytest.mark.asyncio
async def test_rejects_root(self):
m = _make_middleware()
result = await _tool(m, "rm").coroutine("/", runtime=_runtime())
assert isinstance(result, str)
assert "refusing to rm" in result
@pytest.mark.asyncio
async def test_rejects_directory_via_staged_dirs(self):
m = _make_middleware()
runtime = _runtime({"staged_dirs": ["/documents/team-x"]})
result = await _tool(m, "rm").coroutine("/documents/team-x", runtime=runtime)
assert isinstance(result, str)
assert "directory" in result.lower()
assert "rmdir" in result
@pytest.mark.asyncio
async def test_rejects_directory_via_listing(self):
m = _make_middleware()
_bind_backend(
m,
_make_backend_stub(
children=[{"path": "/documents/foo/x.md", "is_dir": False}]
),
)
result = await _tool(m, "rm").coroutine("/documents/foo", runtime=_runtime())
assert isinstance(result, str)
assert "directory" in result.lower()
@pytest.mark.asyncio
async def test_rejects_anonymous_doc(self):
m = _make_middleware()
runtime = _runtime(
{
"kb_anon_doc": {
"path": "/documents/uploaded.xml",
"title": "uploaded",
"content": "",
"chunks": [],
}
}
)
result = await _tool(m, "rm").coroutine(
"/documents/uploaded.xml", runtime=runtime
)
assert isinstance(result, str)
assert "read-only" in result
@pytest.mark.asyncio
async def test_drops_path_from_dirty_paths(self):
m = _make_middleware()
_bind_backend(m, _make_backend_stub(children=[], file_data={"content": ["x"]}))
runtime = _runtime(
{
"files": {"/documents/notes.md": {"content": ["x"]}},
"doc_id_by_path": {"/documents/notes.md": 17},
"dirty_paths": ["/documents/notes.md"],
}
)
result = await _tool(m, "rm").coroutine("/documents/notes.md", runtime=runtime)
dirty = result.update.get("dirty_paths") or []
# First element is the _CLEAR sentinel; the rm'd path must not survive.
assert "/documents/notes.md" not in dirty[1:]
# ---------------------------------------------------------------------------
# rmdir
# ---------------------------------------------------------------------------
class TestRmdirStaging:
@pytest.mark.asyncio
async def test_stages_dir_delete_when_empty_and_db_backed(self):
m = _make_middleware()
backend = _bind_backend(m, _make_backend_stub(children=[]))
backend._load_file_data = AsyncMock(return_value=None)
backend.als_info = AsyncMock(
side_effect=[
[], # children of /documents/proj
[{"path": "/documents/proj", "is_dir": True}], # parent listing
]
)
runtime = _runtime({"cwd": "/documents"}, tool_call_id="tc-rd")
result = await _tool(m, "rmdir").coroutine("/documents/proj", runtime=runtime)
assert hasattr(result, "update")
assert result.update["pending_dir_deletes"] == [
{"path": "/documents/proj", "tool_call_id": "tc-rd"}
]
@pytest.mark.asyncio
async def test_rejects_non_empty(self):
m = _make_middleware()
_bind_backend(
m,
_make_backend_stub(
children=[{"path": "/documents/proj/x.md", "is_dir": False}]
),
)
result = await _tool(m, "rmdir").coroutine(
"/documents/proj", runtime=_runtime()
)
assert isinstance(result, str)
assert "not empty" in result
@pytest.mark.asyncio
async def test_unstages_same_turn_mkdir(self):
m = _make_middleware()
_bind_backend(m, _make_backend_stub(children=[]))
runtime = _runtime(
{"cwd": "/documents", "staged_dirs": ["/documents/scratch"]},
tool_call_id="tc-rd",
)
result = await _tool(m, "rmdir").coroutine(
"/documents/scratch", runtime=runtime
)
assert hasattr(result, "update")
update = result.update
assert "pending_dir_deletes" not in update
staged_after = update["staged_dirs"]
assert staged_after[0] == _CLEAR
assert "/documents/scratch" not in staged_after[1:]
@pytest.mark.asyncio
async def test_rejects_root_and_documents(self):
m = _make_middleware()
for victim in ("/", "/documents"):
result = await _tool(m, "rmdir").coroutine(victim, runtime=_runtime())
assert isinstance(result, str)
assert "refusing to rmdir" in result
@pytest.mark.asyncio
async def test_rejects_cwd(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/proj"})
result = await _tool(m, "rmdir").coroutine("/documents/proj", runtime=runtime)
assert isinstance(result, str)
assert "cwd" in result.lower()
@pytest.mark.asyncio
async def test_rejects_ancestor_of_cwd(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/proj/sub"})
result = await _tool(m, "rmdir").coroutine("/documents/proj", runtime=runtime)
assert isinstance(result, str)
assert "cwd" in result.lower()
@pytest.mark.asyncio
async def test_rejects_files(self):
m = _make_middleware()
_bind_backend(m, _make_backend_stub(children=[], file_data={"content": ["x"]}))
result = await _tool(m, "rmdir").coroutine(
"/documents/notes.md", runtime=_runtime()
)
assert isinstance(result, str)
assert "is a file" in result
# ---------------------------------------------------------------------------
# KBPostgresBackend staged-delete view filter (already the live backend)
# ---------------------------------------------------------------------------
class TestKBPostgresBackendDeleteFilter:
"""``als_info`` / glob / grep must suppress paths queued for delete."""
def _make_backend(self, state: dict[str, Any]) -> KBPostgresBackend:
runtime = SimpleNamespace(state=state)
return KBPostgresBackend(search_space_id=1, runtime=runtime)
def test_pending_filesystem_view_returns_deleted_paths(self):
backend = self._make_backend(
{
"pending_deletes": [{"path": "/documents/x.md", "tool_call_id": "t1"}],
"pending_dir_deletes": [
{"path": "/documents/d1", "tool_call_id": "t2"}
],
}
)
removed, alias, deleted_dirs = backend._pending_filesystem_view({})
assert "/documents/x.md" in removed
assert "/documents/d1" in deleted_dirs
assert alias == {}
def test_dir_suppressed_covers_descendants(self):
backend = self._make_backend({})
deleted_dirs = {"/documents/d"}
assert backend._is_dir_suppressed("/documents/d", deleted_dirs)
assert backend._is_dir_suppressed("/documents/d/x.md", deleted_dirs)
assert backend._is_dir_suppressed("/documents/d/sub/y.md", deleted_dirs)
assert not backend._is_dir_suppressed("/documents/other.md", deleted_dirs)

54
surfsense_backend/tests/unit/middleware/test_b_filesystem_system_prompt.py Normal file
View file

@ -0,0 +1,54 @@
"""Mode-specific system-prompt assembly tests for the LIVE filesystem middleware.
Ported from ``TestModeSpecificPrompts`` in the former
``tests/unit/middleware/test_filesystem_middleware.py`` (which exercised the
dead twin ``app.agents.shared.middleware.filesystem._build_filesystem_system_prompt``).
These drive the production ``build_system_prompt`` so the prompt the model
actually receives stays mode-scoped: cloud rules don't leak into desktop
sessions and vice-versa, and the sandbox section appears only when available.
The per-tool *description* assertions from the old suite are intentionally NOT
ported: they assert exact prompt copy (tightly coupled to the old wording) and
guard prompt token hygiene rather than the code-movement refactor this suite
protects.
"""
from __future__ import annotations
import pytest
from app.agents.multi_agent_chat.middleware.shared.filesystem.system_prompt import (
build_system_prompt,
)
from app.agents.shared.filesystem_selection import FilesystemMode
pytestmark = pytest.mark.unit
class TestModeSpecificPrompts:
def test_cloud_prompt_omits_desktop_section(self):
prompt = build_system_prompt(FilesystemMode.CLOUD, sandbox_available=False)
assert "Local Folder Mode" not in prompt
assert "mount-prefixed" not in prompt
assert "Persistence Rules" in prompt
assert "/documents" in prompt
assert "temp_" in prompt
def test_desktop_prompt_omits_cloud_persistence_rules(self):
prompt = build_system_prompt(
FilesystemMode.DESKTOP_LOCAL_FOLDER, sandbox_available=False
)
assert "Persistence Rules" not in prompt
assert "Workspace Tree" not in prompt
assert "Local Folder Mode" in prompt
assert "mount-prefixed" in prompt
def test_sandbox_addendum_appended_when_available(self):
prompt = build_system_prompt(FilesystemMode.CLOUD, sandbox_available=True)
assert "execute_code" in prompt
assert "Code Execution" in prompt
def test_sandbox_addendum_absent_when_unavailable(self):
prompt = build_system_prompt(FilesystemMode.CLOUD, sandbox_available=False)
assert "execute_code" not in prompt

220
surfsense_backend/tests/unit/middleware/test_filesystem_middleware.py
View file

@ -1,220 +0,0 @@
"""Unit tests for the SurfSense filesystem middleware new behaviors.
Covers:
* cloud cwd defaults to ``/documents`` and relative paths resolve under it
* cloud writes outside ``/documents/`` are rejected unless basename starts
with ``temp_``
* cloud writes/edits to the anonymous document are rejected (read-only)
* helper methods on the middleware (``_resolve_relative``,
``_check_cloud_write_namespace``, ``_default_cwd``)
These tests use ``__new__`` to bypass the heavy ``__init__`` and exercise
the helper methods directly so the test surface stays narrow and fast.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from app.agents.shared.filesystem_selection import FilesystemMode
from app.agents.shared.middleware.filesystem import (
SurfSenseFilesystemMiddleware,
_build_filesystem_system_prompt,
_build_tool_descriptions,
)
pytestmark = pytest.mark.unit
def _make_middleware(mode: FilesystemMode = FilesystemMode.CLOUD):
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._filesystem_mode = mode
return middleware
def _runtime(state: dict | None = None) -> SimpleNamespace:
return SimpleNamespace(state=state or {})
class TestCloudCwdDefaults:
def test_default_cwd_in_cloud_is_documents_root(self):
m = _make_middleware()
assert m._default_cwd() == "/documents"
def test_default_cwd_in_desktop_is_root(self):
m = _make_middleware(FilesystemMode.DESKTOP_LOCAL_FOLDER)
assert m._default_cwd() == "/"
def test_current_cwd_uses_state_when_set(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/notes"})
assert m._current_cwd(runtime) == "/documents/notes"
def test_current_cwd_falls_back_to_default(self):
m = _make_middleware()
runtime = _runtime({})
assert m._current_cwd(runtime) == "/documents"
def test_current_cwd_ignores_invalid(self):
m = _make_middleware()
runtime = _runtime({"cwd": "not-absolute"})
assert m._current_cwd(runtime) == "/documents"
class TestRelativePathResolution:
def test_relative_path_resolves_against_cwd(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/projects"})
assert (
m._resolve_relative("notes.md", runtime) == "/documents/projects/notes.md"
)
def test_relative_path_with_dotdot(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/a/b"})
assert m._resolve_relative("../c.md", runtime) == "/documents/a/c.md"
def test_absolute_path_is_kept(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents"})
assert m._resolve_relative("/other/x.md", runtime) == "/other/x.md"
def test_empty_path_returns_cwd(self):
m = _make_middleware()
runtime = _runtime({"cwd": "/documents/projects"})
assert m._resolve_relative("", runtime) == "/documents/projects"
class TestCloudWriteNamespacePolicy:
def test_documents_path_allowed(self):
m = _make_middleware()
runtime = _runtime()
assert m._check_cloud_write_namespace("/documents/foo.md", runtime) is None
def test_documents_root_allowed(self):
m = _make_middleware()
runtime = _runtime()
assert m._check_cloud_write_namespace("/documents", runtime) is None
def test_temp_basename_anywhere_allowed(self):
m = _make_middleware()
runtime = _runtime()
assert m._check_cloud_write_namespace("/temp_scratch.md", runtime) is None
assert m._check_cloud_write_namespace("/foo/temp_x.md", runtime) is None
assert m._check_cloud_write_namespace("/documents/temp_x.md", runtime) is None
def test_other_paths_rejected(self):
m = _make_middleware()
runtime = _runtime()
err = m._check_cloud_write_namespace("/foo/bar.md", runtime)
assert err is not None
assert "must target /documents" in err
def test_anon_doc_path_is_read_only(self):
m = _make_middleware()
runtime = _runtime(
{
"kb_anon_doc": {
"path": "/documents/uploaded.xml",
"title": "uploaded",
"content": "",
"chunks": [],
}
}
)
err = m._check_cloud_write_namespace("/documents/uploaded.xml", runtime)
assert err is not None
assert "read-only" in err
def test_desktop_mode_skips_namespace_policy(self):
m = _make_middleware(FilesystemMode.DESKTOP_LOCAL_FOLDER)
runtime = _runtime()
assert m._check_cloud_write_namespace("/random/path.md", runtime) is None
class TestModeSpecificPrompts:
"""The prompt and tool descriptions must only describe the active mode.
Cross-mode noise wastes tokens and confuses the model with rules it
cannot use this session.
"""
def test_cloud_prompt_omits_desktop_section(self):
prompt = _build_filesystem_system_prompt(
FilesystemMode.CLOUD, sandbox_available=False
)
assert "Local Folder Mode" not in prompt
assert "mount-prefixed" not in prompt
assert "Persistence Rules" in prompt
assert "/documents" in prompt
assert "temp_" in prompt
def test_desktop_prompt_omits_cloud_persistence_rules(self):
prompt = _build_filesystem_system_prompt(
FilesystemMode.DESKTOP_LOCAL_FOLDER, sandbox_available=False
)
assert "Persistence Rules" not in prompt
assert "Workspace Tree" not in prompt
assert "<priority_documents>" not in prompt
assert "Local Folder Mode" in prompt
assert "mount-prefixed" in prompt
def test_cloud_tool_descs_omit_desktop_phrases(self):
descs = _build_tool_descriptions(FilesystemMode.CLOUD)
for name in (
"write_file",
"edit_file",
"move_file",
"mkdir",
"rm",
"rmdir",
"list_tree",
"grep",
):
text = descs[name]
assert "Desktop" not in text, f"{name} leaks desktop hints"
assert "Cloud mode:" not in text, f"{name} qualifies a cloud-only desc"
def test_desktop_tool_descs_omit_cloud_phrases(self):
descs = _build_tool_descriptions(FilesystemMode.DESKTOP_LOCAL_FOLDER)
for name in (
"write_file",
"edit_file",
"move_file",
"mkdir",
"rm",
"rmdir",
"list_tree",
"grep",
):
text = descs[name]
assert "Cloud" not in text, f"{name} leaks cloud hints"
assert "/documents/" not in text, f"{name} mentions cloud namespace"
assert "temp_" not in text, f"{name} mentions cloud temp_ semantics"
def test_cloud_descs_include_rm_and_rmdir(self):
descs = _build_tool_descriptions(FilesystemMode.CLOUD)
assert "rm" in descs and "rmdir" in descs
assert "Deletes a single file" in descs["rm"]
assert "Deletes an empty directory" in descs["rmdir"]
assert "rmdir" in descs["rmdir"] and "POSIX" in descs["rmdir"]
def test_desktop_descs_warn_about_irreversibility(self):
descs = _build_tool_descriptions(FilesystemMode.DESKTOP_LOCAL_FOLDER)
assert "NOT reversible" in descs["rm"]
assert "NOT reversible" in descs["rmdir"]
def test_sandbox_addendum_appended_when_available(self):
prompt = _build_filesystem_system_prompt(
FilesystemMode.CLOUD, sandbox_available=True
)
assert "execute_code" in prompt
assert "Code Execution" in prompt
def test_sandbox_addendum_absent_when_unavailable(self):
prompt = _build_filesystem_system_prompt(
FilesystemMode.CLOUD, sandbox_available=False
)
assert "execute_code" not in prompt

173
surfsense_backend/tests/unit/middleware/test_filesystem_verification.py
View file

@ -1,173 +0,0 @@
from pathlib import Path
import pytest
from app.agents.shared.filesystem_selection import FilesystemMode
from app.agents.shared.middleware.filesystem import SurfSenseFilesystemMiddleware
from app.agents.shared.middleware.multi_root_local_folder_backend import (
MultiRootLocalFolderBackend,
)
pytestmark = pytest.mark.unit
class _RuntimeNoSuggestedPath:
state = {"file_operation_contract": {}}
class _RuntimeWithSuggestedPath:
def __init__(self, suggested_path: str) -> None:
self.state = {"file_operation_contract": {"suggested_path": suggested_path}}
def test_contract_suggested_path_falls_back_to_documents_notes_md() -> None:
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._filesystem_mode = FilesystemMode.CLOUD
suggested = middleware._get_contract_suggested_path(_RuntimeNoSuggestedPath()) # type: ignore[arg-type]
# Cloud default cwd is /documents so the fallback lands in the KB.
assert suggested == "/documents/notes.md"
def test_contract_suggested_path_falls_back_to_root_notes_md_in_desktop() -> None:
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._filesystem_mode = FilesystemMode.DESKTOP_LOCAL_FOLDER
suggested = middleware._get_contract_suggested_path(_RuntimeNoSuggestedPath()) # type: ignore[arg-type]
assert suggested == "/notes.md"
def test_normalize_local_mount_path_prefixes_default_mount(tmp_path: Path) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path("/random-note.md", runtime) # type: ignore[arg-type]
assert resolved == "/pc_backups/random-note.md"
def test_normalize_local_mount_path_keeps_explicit_mount(tmp_path: Path) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
"/pc_backups/notes/random-note.md",
runtime,
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_normalize_local_mount_path_windows_backslashes(tmp_path: Path) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
r"\notes\random-note.md",
runtime,
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_normalize_local_mount_path_normalizes_mixed_separators(tmp_path: Path) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
r"\\notes//nested\\random-note.md",
runtime,
)
assert resolved == "/pc_backups/notes/nested/random-note.md"
def test_normalize_local_mount_path_keeps_explicit_mount_with_backslashes(
tmp_path: Path,
) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
r"\pc_backups\notes\random-note.md",
runtime,
)
assert resolved == "/pc_backups/notes/random-note.md"
def test_normalize_local_mount_path_prefixes_posix_absolute_path_for_linux_and_macos(
tmp_path: Path,
) -> None:
root = tmp_path / "PC Backups"
root.mkdir()
backend = MultiRootLocalFolderBackend((("pc_backups", str(root)),))
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path("/var/log/app.log", runtime) # type: ignore[arg-type]
assert resolved == "/pc_backups/var/log/app.log"
def test_normalize_local_mount_path_prefers_unique_existing_parent_mount(
tmp_path: Path,
) -> None:
root_a = tmp_path / "RootA"
root_b = tmp_path / "RootB"
(root_a / "other").mkdir(parents=True)
(root_b / "nested" / "deep").mkdir(parents=True)
backend = MultiRootLocalFolderBackend(
(("root_a", str(root_a)), ("root_b", str(root_b)))
)
runtime = _RuntimeNoSuggestedPath()
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
"/nested/deep/new-note.md",
runtime,
)
assert resolved == "/root_b/nested/deep/new-note.md"
def test_normalize_local_mount_path_uses_suggested_mount_when_ambiguous(
tmp_path: Path,
) -> None:
root_a = tmp_path / "RootA"
root_b = tmp_path / "RootB"
root_a.mkdir(parents=True)
root_b.mkdir(parents=True)
backend = MultiRootLocalFolderBackend(
(("root_a", str(root_a)), ("root_b", str(root_b)))
)
runtime = _RuntimeWithSuggestedPath("/root_b/notes/context.md")
middleware = SurfSenseFilesystemMiddleware.__new__(SurfSenseFilesystemMiddleware)
middleware._get_backend = lambda _runtime: backend # type: ignore[method-assign]
resolved = middleware._normalize_local_mount_path( # type: ignore[arg-type]
"/brand-new-note.md",
runtime,
)
assert resolved == "/root_b/brand-new-note.md"