tao.chen
ffec234e40
Codex review of #36 + #37 surfaced that my prior `escape="\\"` only declared the escape character — the pattern literals themselves still contained unescaped `_` and `%`, so `parent_path="foo_bar"` continued to match `fooXbar/...`, `foo2bar/...`, etc. My earlier ESCAPE-clause assertions were tautological: they verified the SQL rendered the ESCAPE keyword without ever checking that the pattern was actually escaped. The tests passed; the leak persisted. Fix in three layers: 1. Real escape: `backend/src/backend/scripts.py` gains `_escape_like_pattern(value)` that escapes `\` → `\\`, `%` → `\%`, `_` → `\_` (in that order — the escape char MUST be escaped first). `_build_list_scripts_descendant_prefix` now returns the escaped prefix. `list_workspace_directories` and `delete_workspace_directory` also escape their server-built prefixes. `count_scripts` escapes the user subtree prefix. 2. Same bug elsewhere: `backend/src/backend/resources.py:404` had the identical `DataResources.resource_name.like(f"%{keyword}%")` pattern; a search for "100%" would match everything. Now escaped too. 3. Count endpoint scope: `count_scripts` was workspace-wide and skipped the StorageObjects JOIN. Now INNER JOINs StorageObjects (drops orphans whose current_object_id is dangling) and filters by `workspace/{user_id}/` subtree so the result matches what `list_scripts(parent_path="")` would return. Multi-member workspaces no longer over-report, and orphan rows no longer inflate the count. Frontend: `DashboardRoute` is not keyed by workspace/user (only ScriptsPage is), so without a workspace_id dep the previous workspace's count persisted across navigation. useEffect now depends on `currentWorkspace?.workspace_id`; `loadScriptCount` clears the count to null at the start of the fetch so the dashboard doesn't flash a stale number. Tests — backend/tests/test_list_scripts_parent_path.py - Rewritten with three layers of coverage: * Pure helper tests for `_escape_like_pattern` (7 cases including backslash-escape-first ordering). * SQL-contract tests asserting the COMPILED PATTERN contains the escaped form (lowercased to neutralise SQLAlchemy keyword casing). * BEHAVIORAL tests on SQLite in-memory with the same LIKE semantics — proves the fix actually prevents the wildcard leak. Includes a negative test (without escape, siblings DO match) so the fixture is verified to exercise the bug. Tests — backend/tests/test_count_scripts.py - Updated to assert the JOIN + user-scope filter. New test verifies two different users in the same workspace get different subtrees. Verified: - pytest backend/tests: 65 passed (43 baseline + 12 list_scripts + 4 count + 6 helper/SQLite behavioral) - pnpm typecheck: clean - Raw SQL on MySQL (live DB) confirms `LIKE 'workspace/.../foo\_bar/%%' ESCAPE '\\'`.
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Unit tests for GET /api/v1/scripts/count endpoint.
|
|
|
|
Verifies the count endpoint returns the same scope as
|
|
``list_scripts(parent_path="")``: workspace + active scripts whose
|
|
``StorageObjects.relative_path`` lives under the user's subtree. This
|
|
avoids under/over-reporting on the dashboard — the count is the size of
|
|
the set list_scripts would return if it weren't lazy.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from backend.scripts import count_scripts
|
|
|
|
|
|
def _ctx(user_id: str = "U001", workspace_id: str = "W001") -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
request_id="test",
|
|
user=SimpleNamespace(user_id=user_id),
|
|
workspace=SimpleNamespace(workspace_id=workspace_id),
|
|
role=SimpleNamespace(role_code="admin"),
|
|
is_system_admin=False,
|
|
)
|
|
|
|
|
|
def _compile(stmt) -> str:
|
|
from sqlalchemy.dialects import mysql as mysql_dialect
|
|
|
|
return str(
|
|
stmt.compile(
|
|
dialect=mysql_dialect.dialect(),
|
|
compile_kwargs={"literal_binds": True},
|
|
)
|
|
)
|
|
|
|
|
|
async def test_count_scripts_returns_scalar_int() -> None:
|
|
captured = []
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.scalar = AsyncMock(
|
|
side_effect=lambda stmt: (captured.append(stmt), 7)[1]
|
|
)
|
|
|
|
result = await count_scripts(context=_ctx(), session=mock_session)
|
|
assert result["data"] == {"total": 7}
|
|
assert result["meta"] == {}
|
|
assert result["request_id"] == "test"
|
|
# Exactly one COUNT(*) query issued.
|
|
assert len(captured) == 1
|
|
stmt = captured[0]
|
|
sql = _compile(stmt).lower()
|
|
# JOIN to StorageObjects so orphaned scripts (no joinable row) are
|
|
# excluded — matches list_scripts INNER JOIN behaviour.
|
|
assert "inner join storage_objects" in sql
|
|
# Scope: workspace_id + active status + user subtree.
|
|
assert "scripts.workspace_id" in sql
|
|
assert "scripts.status" in sql
|
|
assert "workspace/u001/%" in sql
|
|
|
|
|
|
async def test_count_scripts_handles_null_result() -> None:
|
|
"""MySQL COUNT(*) on empty result returns 0, not NULL — but defensively
|
|
coerce NULL to 0 to keep the response shape consistent."""
|
|
mock_session = MagicMock()
|
|
mock_session.scalar = AsyncMock(return_value=None)
|
|
result = await count_scripts(context=_ctx(), session=mock_session)
|
|
assert result["data"] == {"total": 0}
|
|
|
|
|
|
async def test_count_scripts_uses_user_specific_subtree() -> None:
|
|
"""Different users in the same workspace must see different totals —
|
|
each user's count is bounded by their own ``workspace/{user_id}/``
|
|
subtree, NOT the whole workspace."""
|
|
captured = []
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.scalar = AsyncMock(
|
|
side_effect=lambda stmt: (captured.append(stmt), 3)[1]
|
|
)
|
|
|
|
await count_scripts(context=_ctx(user_id="alice"), session=mock_session)
|
|
sql_alice = _compile(captured[-1])
|
|
|
|
await count_scripts(context=_ctx(user_id="bob"), session=mock_session)
|
|
sql_bob = _compile(captured[-1])
|
|
|
|
assert "workspace/alice/%" in sql_alice
|
|
assert "workspace/alice/%" not in sql_bob
|
|
assert "workspace/bob/%" in sql_bob
|
|
|
|
|
|
async def test_count_scripts_route_declared_before_script_id_route() -> None:
|
|
"""Static check: the `/api/v1/scripts/count` route MUST be declared in
|
|
scripts.py before `/api/v1/scripts/{script_id}/...`, otherwise FastAPI's
|
|
declaration-order matching will interpret `count` as a script_id."""
|
|
from backend.scripts import count_scripts, get_script
|
|
|
|
assert callable(count_scripts)
|
|
assert callable(get_script) |