feat(scripts/resources): align list_scripts visibility with list_resources

69a9a48 把 data resources 的可见性从 list 恒真改成
workspace-wide + visibility 过滤 + admin 短路,但 scripts 端
没动。两端不对称,导致:

* 非 admin 调 list_scripts 走 user_relative_path →
  workspace/{当前用户ID}/...,永远拿不到别人的脚本
* count_scripts 同样 user-scoped,dashboard "全部脚本"
  只统计自己
* list_resources 响应没带 owner_display_name,data-only
  owner 的目录名回退到 userId.slice(-6),显示不友好

修复:
* list_scripts / count_scripts 改 workspace-wide(新建
  _build_list_scripts_workspace_descendant_prefix helper;
  旧 _build_list_scripts_descendant_prefix 保留标 deprecated
  避免破坏其它调用方);非 admin 追加
  or_(owner_user_id = me, visibility in {workspace, public})
  与 list_resources 完全对称;admin 短路
* resource_payload 加 owner_display_name 字段(与
  script_payload 对称);list_resources SELECT 加
  Users.display_name + outerjoin
* 前端 ScriptExplorer displayName 回退链:scripts 的
  owner_display_name → data resources 的 owner_display_name
  → 本人 user.display_name → userId.slice(-6) 占位
  ;ResourceItem 类型同步加 owner_display_name?: string|null

存储物理布局仍是 workspace/{user_id}/...,仅读取侧
listing/count 跨 owner。
This commit is contained in:
tao.chen
2026-08-21 14:06:45 +08:00
parent 190c672e42
commit d68055a96c
7 changed files with 288 additions and 46 deletions
+61 -20
View File
@@ -1,10 +1,11 @@
"""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.
Verifies the count endpoint matches the (workspace-wide) listing scope of
``list_scripts(parent_path="")``: workspace + active scripts across every
owner's ``StorageObjects.relative_path``, narrowed by visibility for
non-admin (admin short-circuits). 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
@@ -17,13 +18,23 @@ import pytest
from backend.scripts import count_scripts
def _ctx(user_id: str = "U001", workspace_id: str = "W001") -> SimpleNamespace:
def _ctx(
user_id: str = "U001",
workspace_id: str = "W001",
*,
is_admin: bool = False,
is_system_admin: bool = False,
) -> 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,
role=SimpleNamespace(role_code="admin" if is_admin else "developer"),
is_system_admin=is_system_admin,
# ``count_scripts`` now consults ``context.is_admin`` directly
# (matching list_scripts / list_resources); SimpleNamespace needs
# it as a plain attribute.
is_admin=is_admin or is_system_admin,
)
@@ -57,10 +68,13 @@ async def test_count_scripts_returns_scalar_int() -> None:
# 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.
# Scope: workspace_id + active status + workspace-wide prefix.
assert "scripts.workspace_id" in sql
assert "scripts.status" in sql
assert "workspace/u001/%" in sql
assert "like 'workspace/%%'" in sql
# Non-admin (default) narrows by visibility.
assert "scripts.owner_user_id = 'u001'" in sql
assert "scripts.visibility in ('workspace', 'public')" in sql
async def test_count_scripts_handles_null_result() -> None:
@@ -72,10 +86,10 @@ async def test_count_scripts_handles_null_result() -> None:
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."""
async def test_count_scripts_workspace_wide_not_user_scoped() -> None:
"""The prefix is workspace-wide (``workspace/%`` — no embedded user_id),
so different users count the same physical tree; the only per-user
difference is the non-admin visibility predicate (owner_user_id = me)."""
captured = []
mock_session = MagicMock()
@@ -84,14 +98,41 @@ async def test_count_scripts_uses_user_specific_subtree() -> None:
)
await count_scripts(context=_ctx(user_id="alice"), session=mock_session)
sql_alice = _compile(captured[-1])
sql_alice = _compile(captured[-1]).lower()
await count_scripts(context=_ctx(user_id="bob"), session=mock_session)
sql_bob = _compile(captured[-1])
sql_bob = _compile(captured[-1]).lower()
assert "workspace/alice/%" in sql_alice
assert "workspace/alice/%" not in sql_bob
assert "workspace/bob/%" in sql_bob
# Both count the same workspace-wide subtree.
assert "like 'workspace/%%'" in sql_alice
assert "like 'workspace/%%'" in sql_bob
# Neither embeds the user_id in the path prefix.
assert "workspace/alice/%" not in sql_alice
assert "workspace/bob/%" not in sql_bob
# Per-user narrowing happens via the visibility predicate.
assert "scripts.owner_user_id = 'alice'" in sql_alice
assert "scripts.owner_user_id = 'bob'" in sql_bob
async def test_count_scripts_admin_skips_visibility_filter() -> None:
"""Admin short-circuits the visibility predicate and counts every
active script in the workspace (dashboard '全部脚本' / '工作副本')."""
captured = []
mock_session = MagicMock()
mock_session.scalar = AsyncMock(
side_effect=lambda stmt: (captured.append(stmt), 42)[1]
)
result = await count_scripts(
context=_ctx(user_id="alice", is_admin=True), session=mock_session
)
assert result["data"] == {"total": 42}
sql = _compile(captured[0]).lower()
assert "like 'workspace/%%'" in sql
# visibility / owner_user_id still appear in the SELECT projection, but
# the visibility WHERE predicate must be absent for admins.
assert "scripts.visibility in ('workspace', 'public')" not in sql
async def test_count_scripts_route_declared_before_script_id_route() -> None:
@@ -101,4 +142,4 @@ async def test_count_scripts_route_declared_before_script_id_route() -> None:
from backend.scripts import count_scripts, get_script
assert callable(count_scripts)
assert callable(get_script)
assert callable(get_script)
+112 -8
View File
@@ -29,18 +29,24 @@ from sqlalchemy.dialects import mysql as mysql_dialect
from backend.scripts import (
_build_list_scripts_descendant_prefix,
_build_list_scripts_workspace_descendant_prefix,
_escape_like_pattern,
normalize_user_path,
)
def _ctx(user_id: str = "U001") -> SimpleNamespace:
def _ctx(
user_id: str = "U001", *, is_admin: bool = False, is_system_admin: bool = False
) -> SimpleNamespace:
return SimpleNamespace(
request_id="test",
user=SimpleNamespace(user_id=user_id),
workspace=SimpleNamespace(workspace_id="W001"),
role=SimpleNamespace(role_code="admin"),
is_system_admin=False,
role=SimpleNamespace(role_code="admin" if is_admin else "developer"),
is_system_admin=is_system_admin,
# ``list_scripts`` now consults ``context.is_admin`` directly (matching
# ``list_resources``); SimpleNamespace needs it as a plain attribute.
is_admin=is_admin or is_system_admin,
)
@@ -115,6 +121,51 @@ def test_normalize_user_path_strips() -> None:
assert normalize_user_path("a\\b") == "a/b"
# ─── layer 1.6: workspace-wide prefix helper ─────────────────────
def test_workspace_descendant_prefix_root() -> None:
"""Empty parent_path → workspace-wide root prefix (cross-owner)."""
assert _build_list_scripts_workspace_descendant_prefix("") == "workspace/"
def test_workspace_descendant_prefix_subdir() -> None:
"""Non-empty parent_path → appended under workspace root, no user_id."""
assert (
_build_list_scripts_workspace_descendant_prefix("foo/bar")
== "workspace/foo/bar/"
)
def test_workspace_descendant_prefix_escapes_metachars() -> None:
r"""Folder ``foo_bar`` must produce ``foo\_bar`` so the trailing ``%``
doesn't become 'match any single char'."""
assert (
_build_list_scripts_workspace_descendant_prefix("foo_bar")
== r"workspace/foo\_bar/"
)
def test_workspace_descendant_prefix_escapes_percent() -> None:
assert (
_build_list_scripts_workspace_descendant_prefix("100%match")
== r"workspace/100\%match/"
)
def test_workspace_descendant_prefix_normalizes_leading_trailing_slashes() -> None:
assert (
_build_list_scripts_workspace_descendant_prefix("/foo/bar/")
== "workspace/foo/bar/"
)
def test_workspace_descendant_prefix_rejects_traversal() -> None:
with pytest.raises(HTTPException) as exc:
_build_list_scripts_workspace_descendant_prefix("foo/../bar")
assert exc.value.status_code == 422
# ─── layer 2: SQL contract ────────────────────────────────────────
@@ -147,8 +198,8 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
assert len(captured_sql) == 1
sql = captured_sql[0].lower()
assert "like 'workspace/alice/foo/bar/%%'" in sql
assert "not like 'workspace/alice/foo/bar/%%/%%'" in sql
assert "like 'workspace/foo/bar/%%'" in sql
assert "not like 'workspace/foo/bar/%%/%%'" in sql
async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
@@ -177,9 +228,9 @@ async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
# doubles the escape char inside the SQL string literal, so what
# the helper emits as `foo\_bar` renders as `foo\\_bar` here
# (2 backslash chars in the actual SQL string).
assert r"like 'workspace/alice/foo\\_bar/%%'" in sql_lower
assert r"like 'workspace/foo\\_bar/%%'" in sql_lower
# NOT LIKE clause also escaped.
assert r"not like 'workspace/alice/foo\\_bar/%%/%%'" in sql_lower
assert r"not like 'workspace/foo\\_bar/%%/%%'" in sql_lower
# And both declare ESCAPE '\\'.
assert sql.count("ESCAPE '\\\\'") == 2, sql
@@ -206,7 +257,60 @@ async def test_list_scripts_where_clause_escapes_percent_pattern() -> None:
sql_lower = sql.lower()
# SQLAlchemy doubles the escape char so `%` → `\%` becomes `\\%`
# in the SQL string literal.
assert r"workspace/alice/100\\%%match/%%" in sql_lower
assert r"workspace/100\\%%match/%%" in sql_lower
async def test_list_scripts_non_admin_adds_visibility_filter() -> None:
"""Workspace-wide listing is narrowed by visibility for non-admin:
owner_user_id = me OR visibility IN (workspace, public) — exactly like
list_resources. The workspace prefix contains NO user_id (cross-owner)."""
from backend.scripts import list_scripts
captured_sql: list[str] = []
class _MockResult:
def all(self):
return []
mock_session = MagicMock()
mock_session.execute = AsyncMock(
side_effect=lambda stmt: (
captured_sql.append(_compile_sql(stmt)) or _MockResult()
)
)
await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session)
sql = captured_sql[0].lower()
assert "like 'workspace/%%'" in sql
assert "scripts.owner_user_id = 'alice'" in sql
assert "scripts.visibility in ('workspace', 'public')" in sql
async def test_list_scripts_admin_skips_visibility_filter() -> None:
"""Admin short-circuits the visibility predicate and sees everything."""
from backend.scripts import list_scripts
captured_sql: list[str] = []
class _MockResult:
def all(self):
return []
mock_session = MagicMock()
mock_session.execute = AsyncMock(
side_effect=lambda stmt: (
captured_sql.append(_compile_sql(stmt)) or _MockResult()
)
)
await list_scripts(
parent_path="", context=_ctx("alice", is_admin=True), session=mock_session
)
sql = captured_sql[0].lower()
assert "like 'workspace/%%'" in sql
# owner_user_id / visibility still appear in the SELECT projection; what
# must be absent is the visibility WHERE predicate for non-admins.
assert "scripts.visibility in ('workspace', 'public')" not in sql
async def test_list_workspace_directories_where_clause_escapes_pattern() -> None:
+31
View File
@@ -322,6 +322,8 @@ def test_resource_payload_legacy_dot_resources():
)
assert payload["jupyter_accessible_path"] == ".resources/data.csv"
assert payload["absolute_path"].endswith(f"{ws}/{user}/.resources/data.csv")
# Default None when no Users join row is provided (bind_resource path).
assert payload["owner_display_name"] is None
def test_resource_payload_new_flat_path():
@@ -333,6 +335,13 @@ def test_resource_payload_new_flat_path():
)
assert payload["jupyter_accessible_path"] == "data.csv"
assert payload["absolute_path"].endswith(f"{ws}/{user}/data.csv")
# Explicit owner_display_name is passed through to the payload.
payload = resource_payload(
_make_resource(ws, user),
_make_storage_object(f"{ws}/{user}/data.csv"),
owner_display_name="张三",
)
assert payload["owner_display_name"] == "张三"
def test_resource_payload_new_nested_path():
@@ -344,6 +353,7 @@ def test_resource_payload_new_nested_path():
)
assert payload["jupyter_accessible_path"] == "train/v1/data.csv"
assert payload["absolute_path"].endswith(f"{ws}/{user}/train/v1/data.csv")
assert payload["owner_display_name"] is None
def test_compute_jupyter_relative_path_for_legacy_and_new_paths():
@@ -621,6 +631,27 @@ async def test_list_resources_without_parent_path_adds_no_like_clause() -> None:
assert " not like " not in sql
async def test_list_resources_joins_users_for_display_name() -> None:
"""list_resources must OUTER JOIN users and SELECT users.display_name so
every resource carries owner_display_name (frontend displayName chain)."""
from backend.resources import list_resources
captured_sql: list[str] = []
mock_session = _list_resources_capturing_session(captured_sql)
await list_resources(
parent_path="",
context=_resource_ctx(),
session=mock_session,
visibility=None,
keyword=None,
)
sql = captured_sql[0]
sql_lower = sql.lower()
assert "outer join users" in sql_lower
assert "users.display_name" in sql_lower
# ─── layer 3: behavioral test on real LIKE execution (SQLite) ────────────────