fix(scripts): workspace-wide parent_path listing + private visibility on reads
1. 同 workspace 互相可见(排除 private):
- list_scripts / count_scripts 已 workspace-wide + visibility 过滤,但
单条读取(get/content/latest-version/versions)不校验 visibility,非
owner 猜 id 即可读他人 private 脚本。新增 script_can_view(与 data
resources 的 can_view 对称)并在 get_script_row / latest_version 强制,
private 对非 owner 返回 404。
2. parent_path 为空默认拉根路径文件:
- 物理存储为 workspace/{user_id}/...,根 prefix 原来是 workspace/ +
NOT LIKE workspace/%/%,所有文件都在两层被整体排除,list_scripts("")
恒空。改为 workspace/%/,配合 LIKE workspace/%/% AND NOT LIKE
workspace/%/%/% 返回各 owner 根级文件。
3. 非空 parent_path 跨 owner 查询:
- 原来 workspace/foo/ 永远匹配不到 workspace/{uid}/foo/...,子目录
懒加载返回空,其他用户目录点击无内容。改为 workspace/%/foo/(owner
段通配,与 list_resources 一致),_ / % 仍按字面转义。
测试:更新前缀契约断言,新增 SQLite 行为测试(跨 owner 根/子目录、转义)
与 script_can_view / get_script 权限测试,133 passed。
This commit is contained in:
@@ -170,23 +170,29 @@ def _build_list_scripts_workspace_descendant_prefix(parent_path: str) -> str:
|
||||
"""Return the escaped materialized-path prefix for direct children of
|
||||
``parent_path`` across **all owners** in the workspace.
|
||||
|
||||
Storage is still physically laid out as ``workspace/{user_id}/...``, but
|
||||
listing is workspace-wide: the prefix starts at ``workspace/`` (no
|
||||
embedded user_id) so the endpoint's ``LIKE '<prefix>/%'`` walks every
|
||||
owner's subtree. Non-admin scoping is handled separately in the SQL via
|
||||
Storage is still physically laid out as ``workspace/{user_id}/...``, so
|
||||
listing a subdirectory across every owner must match the owner segment
|
||||
with an intentional ``%`` wildcard — ``workspace/%/foo`` — exactly like
|
||||
list_resources does against ``object_key``. The endpoint's
|
||||
``LIKE '<prefix>/%' AND NOT LIKE '<prefix>/%/%'`` pair then grabs only
|
||||
DIRECT children of ``parent_path`` per owner. Non-admin scoping is
|
||||
handled separately in the SQL via
|
||||
``owner_user_id = me OR visibility IN (workspace, public)``.
|
||||
|
||||
Empty ``parent_path`` returns ``"workspace/"`` (cross-owner root).
|
||||
The prefix is run through ``_escape_like_pattern`` so folder names
|
||||
containing ``_`` / ``%`` do not act as wildcards. The trailing ``/``
|
||||
is appended AFTER escaping so it remains a literal slash.
|
||||
Empty ``parent_path`` returns ``"workspace/%/"``: the physical root
|
||||
level is each owner's subtree (``workspace/{owner_id}/...``), so the
|
||||
owner segment is wildcarded and the endpoint's direct-child pair
|
||||
keeps every owner's root-level files (``workspace/%/%`` AND NOT
|
||||
``workspace/%/%/%``). The prefix is run through ``_escape_like_pattern``
|
||||
so folder names containing ``_`` / ``%`` do not act as wildcards. The
|
||||
trailing ``/`` is appended AFTER escaping so it remains a literal slash.
|
||||
"""
|
||||
normalized_parent = normalize_user_path(parent_path)
|
||||
if normalized_parent:
|
||||
target_prefix = f"workspace/{normalized_parent}"
|
||||
target_prefix = f"workspace/%/{_escape_like_pattern(normalized_parent)}"
|
||||
else:
|
||||
target_prefix = "workspace"
|
||||
return f"{_escape_like_pattern(target_prefix)}/"
|
||||
target_prefix = "workspace/%"
|
||||
return f"{target_prefix}/"
|
||||
|
||||
|
||||
def safe_script_name(value: str, script_type: str) -> str:
|
||||
@@ -328,6 +334,22 @@ def version_payload(version: Versions) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def script_can_view(script: Scripts, context: RequestContext) -> bool:
|
||||
"""同一 workspace 内:owner 永远可见自己的脚本(含 private);
|
||||
其他成员只见 visibility in {workspace, public} 的脚本;
|
||||
admin 全部可见。与 data resources 的 can_view 完全对称。
|
||||
|
||||
用于单脚本读取(get / content / latest-version / versions 列表)以及
|
||||
所有写路径(update / delete / publish)的前置校验 —— 保证 A 的 private
|
||||
脚本对非 owner 不可见,而不仅是「列表里不出现」。
|
||||
"""
|
||||
if script.owner_user_id == context.user.user_id:
|
||||
return True
|
||||
if script.visibility in {"workspace", "public"}:
|
||||
return True
|
||||
return context.is_admin
|
||||
|
||||
|
||||
async def get_script_row(
|
||||
script_id: str,
|
||||
context: RequestContext,
|
||||
@@ -392,6 +414,11 @@ async def get_script_row(
|
||||
if row is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "script not found")
|
||||
script, storage_object = row
|
||||
if not script_can_view(script, context):
|
||||
# Private scripts are only visible to their owner (and admin);
|
||||
# treat cross-owner access as not-found so the id cannot probe
|
||||
# visibility, matching get_visible_resource for data resources.
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "script not found")
|
||||
return script, storage_object
|
||||
|
||||
|
||||
@@ -1122,8 +1149,10 @@ async def delete_workspace_directory(
|
||||
# 列出当前工作区中用户有权查看的脚本与 Notebook 元数据。
|
||||
#
|
||||
# 可选 ``parent_path`` Query 参数把返回范围限定为 *直接挂在该路径下的* 脚本
|
||||
# (不含更深的子目录)。空字符串等价于用户作用域根目录;这是前端按目录懒加载
|
||||
# 的关键端点,避免 10 万级脚本一次性返回。
|
||||
# (不含更深的子目录)。空字符串等价于工作区根目录(跨所有 owner 根级文件);
|
||||
# 非空路径按 owner 段通配(``workspace/%/<parent>``)跨所有 owner 查询,与
|
||||
# list_resources 一致。这是前端按目录懒加载的关键端点,避免 10 万级脚本
|
||||
# 一次性返回。
|
||||
#
|
||||
# 实现走 ``storage_objects.relative_path`` materialized path(``LIKE prefix%``
|
||||
# + ``NOT LIKE prefix%/%`` 取直接子节点)。MySQL 优化器目前选
|
||||
@@ -1140,10 +1169,12 @@ async def list_scripts(
|
||||
session: AsyncSession = Depends(database_session),
|
||||
) -> dict[str, Any]:
|
||||
# Workspace-wide listing: storage is physically laid out as
|
||||
# ``workspace/{user_id}/...``, but the prefix starts at ``workspace/``
|
||||
# (no embedded user_id) so the LIKE walks every owner's subtree.
|
||||
# Non-admin scoping is applied below via visibility, matching
|
||||
# list_resources (69a9a48).
|
||||
# ``workspace/{user_id}/...``. Empty parent_path wildcards the owner
|
||||
# segment (``workspace/%/``) so each owner's root files are returned;
|
||||
# non-empty parent_path embeds the same owner wildcard
|
||||
# (``workspace/%/foo``) so every owner's ``foo`` subtree matches,
|
||||
# mirroring list_resources. Non-admin scoping is applied below via
|
||||
# visibility, matching list_resources (69a9a48).
|
||||
descendant_prefix = _build_list_scripts_workspace_descendant_prefix(parent_path)
|
||||
|
||||
statement = (
|
||||
@@ -1638,18 +1669,15 @@ async def latest_version(
|
||||
published versions yet (so the frontend can render an empty label
|
||||
without a 404 round-trip).
|
||||
"""
|
||||
script = await session.scalar(
|
||||
select(Scripts.script_id).where(
|
||||
Scripts.script_id == script_id,
|
||||
Scripts.workspace_id == context.workspace.workspace_id,
|
||||
Scripts.status == "active",
|
||||
)
|
||||
# Route through get_script_row so the private-visibility guard applies
|
||||
# here too: other members must not learn about a private script's
|
||||
# versions by guessing its id.
|
||||
_script, _ = await get_script_row(
|
||||
script_id,
|
||||
context,
|
||||
session,
|
||||
allow_missing_storage_object=True,
|
||||
)
|
||||
if script is None:
|
||||
raise HTTPException(
|
||||
status.HTTP_404_NOT_FOUND,
|
||||
"script not found",
|
||||
)
|
||||
latest = await session.scalar(
|
||||
select(Versions)
|
||||
.where(
|
||||
|
||||
@@ -68,10 +68,11 @@ 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 + workspace-wide prefix.
|
||||
# Scope: workspace_id + active status + workspace-wide prefix
|
||||
# (owner segment wildcarded: workspace/%/%).
|
||||
assert "scripts.workspace_id" in sql
|
||||
assert "scripts.status" in sql
|
||||
assert "like 'workspace/%%'" 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
|
||||
@@ -87,9 +88,10 @@ async def test_count_scripts_handles_null_result() -> None:
|
||||
|
||||
|
||||
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)."""
|
||||
"""The prefix is workspace-wide (``workspace/%/%`` — owner segment
|
||||
wildcarded, 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()
|
||||
@@ -104,8 +106,8 @@ async def test_count_scripts_workspace_wide_not_user_scoped() -> None:
|
||||
sql_bob = _compile(captured[-1]).lower()
|
||||
|
||||
# Both count the same workspace-wide subtree.
|
||||
assert "like 'workspace/%%'" in sql_alice
|
||||
assert "like 'workspace/%%'" in sql_bob
|
||||
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
|
||||
@@ -129,7 +131,7 @@ async def test_count_scripts_admin_skips_visibility_filter() -> None:
|
||||
)
|
||||
assert result["data"] == {"total": 42}
|
||||
sql = _compile(captured[0]).lower()
|
||||
assert "like 'workspace/%%'" in sql
|
||||
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
|
||||
|
||||
@@ -125,15 +125,23 @@ def test_normalize_user_path_strips() -> None:
|
||||
|
||||
|
||||
def test_workspace_descendant_prefix_root() -> None:
|
||||
"""Empty parent_path → workspace-wide root prefix (cross-owner)."""
|
||||
assert _build_list_scripts_workspace_descendant_prefix("") == "workspace/"
|
||||
"""Empty parent_path → workspace-wide root prefix (cross-owner).
|
||||
|
||||
Physical root level is each owner's subtree, so the prefix wildcards
|
||||
the owner segment: ``workspace/%/`` (direct children ``workspace/%/%``
|
||||
minus ``workspace/%/%/%``)."""
|
||||
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."""
|
||||
"""Non-empty parent_path → workspace-wide with owner-segment wildcard.
|
||||
|
||||
Physical rows live under ``workspace/{owner_id}/...``, so a subdirectory
|
||||
listing must match ANY owner: ``workspace/%/foo/bar`` (the ``%`` is the
|
||||
intentional owner wildcard, exactly like list_resources)."""
|
||||
assert (
|
||||
_build_list_scripts_workspace_descendant_prefix("foo/bar")
|
||||
== "workspace/foo/bar/"
|
||||
== "workspace/%/foo/bar/"
|
||||
)
|
||||
|
||||
|
||||
@@ -142,21 +150,21 @@ def test_workspace_descendant_prefix_escapes_metachars() -> None:
|
||||
doesn't become 'match any single char'."""
|
||||
assert (
|
||||
_build_list_scripts_workspace_descendant_prefix("foo_bar")
|
||||
== r"workspace/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/"
|
||||
== 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/"
|
||||
== "workspace/%/foo/bar/"
|
||||
)
|
||||
|
||||
|
||||
@@ -198,8 +206,10 @@ 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/foo/bar/%%'" in sql
|
||||
assert "not like 'workspace/foo/bar/%%/%%'" in sql
|
||||
# 中间 ``%`` 是 owner 段通配符(跨所有 owner),父路径本身按字面匹配,
|
||||
# 与 list_resources 对 object_key 的过滤一致。
|
||||
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:
|
||||
@@ -228,9 +238,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/foo\\_bar/%%'" in sql_lower
|
||||
assert r"like 'workspace/%%/foo\\_bar/%%'" in sql_lower
|
||||
# NOT LIKE clause also escaped.
|
||||
assert r"not like 'workspace/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
|
||||
|
||||
@@ -257,7 +267,7 @@ 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/100\\%%match/%%" in sql_lower
|
||||
assert r"workspace/%%/100\\%%match/%%" in sql_lower
|
||||
|
||||
|
||||
async def test_list_scripts_non_admin_adds_visibility_filter() -> None:
|
||||
@@ -281,7 +291,9 @@ async def test_list_scripts_non_admin_adds_visibility_filter() -> None:
|
||||
|
||||
await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session)
|
||||
sql = captured_sql[0].lower()
|
||||
assert "like 'workspace/%%'" in sql
|
||||
# Root listing wildcards the owner segment: LIKE workspace/%/%
|
||||
# (each owner's root files), excluding 3+ segment descendants.
|
||||
assert "like 'workspace/%%/%%'" in sql
|
||||
assert "scripts.owner_user_id = 'alice'" in sql
|
||||
assert "scripts.visibility in ('workspace', 'public')" in sql
|
||||
|
||||
@@ -307,7 +319,7 @@ async def test_list_scripts_admin_skips_visibility_filter() -> None:
|
||||
parent_path="", context=_ctx("alice", is_admin=True), session=mock_session
|
||||
)
|
||||
sql = captured_sql[0].lower()
|
||||
assert "like 'workspace/%%'" in sql
|
||||
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
|
||||
@@ -439,4 +451,273 @@ def test_sqlite_like_with_percent_in_name(sqlite_like_table):
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
assert matched == ["workspace/alice/100%off/x.py"], matched
|
||||
assert matched == ["workspace/alice/100%off/x.py"], matched
|
||||
|
||||
# ─── layer 3.5: workspace-wide subdirectory semantics ──────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sqlite_workspace_table():
|
||||
"""SQLite table of physical relative_path rows (workspace/{owner}/...)."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
metadata = MetaData()
|
||||
table = Table(
|
||||
"paths",
|
||||
metadata,
|
||||
Column("relative_path", String(1024), nullable=False),
|
||||
)
|
||||
metadata.create_all(engine)
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
[
|
||||
# Direct children of `foo/bar` across two owners.
|
||||
{"relative_path": "workspace/alice/foo/bar/a.py"},
|
||||
{"relative_path": "workspace/bob/foo/bar/b.py"},
|
||||
# Deeper than one level under foo/bar → must be excluded.
|
||||
{"relative_path": "workspace/bob/foo/bar/nested/c.py"},
|
||||
# Sibling folder `barX` / `foobar` → must be excluded.
|
||||
{"relative_path": "workspace/alice/foo/barX/decoy.py"},
|
||||
{"relative_path": "workspace/alice/foobar/x.py"},
|
||||
# Root-level files (not under any subdir).
|
||||
{"relative_path": "workspace/alice/root.py"},
|
||||
],
|
||||
)
|
||||
yield engine, table
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _run_like(query_table, prefix: str, escape: str = "\\"):
|
||||
"""Run the endpoint's LIKE + NOT LIKE direct-child pair on SQLite.
|
||||
|
||||
Mirrors list_scripts exactly: LIKE ``prefix + '%'`` and
|
||||
NOT LIKE ``prefix + '%/%'``."""
|
||||
engine, table = query_table
|
||||
stmt = select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(f"{prefix}%", escape=escape),
|
||||
~table.c.relative_path.like(f"{prefix}%/%", escape=escape),
|
||||
)
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(stmt).fetchall()
|
||||
return sorted(r[0] for r in rows)
|
||||
|
||||
|
||||
def test_sqlite_workspace_subdir_prefix_matches_direct_children_across_owners(
|
||||
sqlite_workspace_table,
|
||||
) -> None:
|
||||
"""parent_path='foo/bar' must return each owner's DIRECT children of
|
||||
foo/bar — the workspace-wide contract for lazy directory loading."""
|
||||
prefix = _build_list_scripts_workspace_descendant_prefix("foo/bar")
|
||||
matched = _run_like(sqlite_workspace_table, prefix)
|
||||
assert matched == [
|
||||
"workspace/alice/foo/bar/a.py",
|
||||
"workspace/bob/foo/bar/b.py",
|
||||
], matched
|
||||
|
||||
|
||||
def test_sqlite_workspace_root_prefix_matches_each_owner_root(
|
||||
sqlite_workspace_table,
|
||||
) -> None:
|
||||
"""parent_path='' must return root-level files across all owners —
|
||||
requirement: empty parent_path defaults to pulling root-path files."""
|
||||
prefix = _build_list_scripts_workspace_descendant_prefix("")
|
||||
matched = _run_like(sqlite_workspace_table, prefix)
|
||||
# Direct children of each owner's root = that owner's root-level file.
|
||||
assert matched == ["workspace/alice/root.py"], matched
|
||||
|
||||
|
||||
def test_sqlite_workspace_subdir_escapes_underscore_across_owners() -> None:
|
||||
"""Escaping still works with the owner wildcard added: `foo_bar` must
|
||||
not match `fooXbar` under ANY owner."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
metadata = MetaData()
|
||||
table = Table(
|
||||
"paths",
|
||||
metadata,
|
||||
Column("relative_path", String(1024), nullable=False),
|
||||
)
|
||||
metadata.create_all(engine)
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
[
|
||||
{"relative_path": "workspace/alice/foo_bar/inner.py"},
|
||||
{"relative_path": "workspace/bob/foo_bar/inner.py"},
|
||||
{"relative_path": "workspace/alice/fooXbar/decoy.py"},
|
||||
],
|
||||
)
|
||||
prefix = _build_list_scripts_workspace_descendant_prefix("foo_bar")
|
||||
matched = _run_like((engine, table), prefix)
|
||||
assert matched == [
|
||||
"workspace/alice/foo_bar/inner.py",
|
||||
"workspace/bob/foo_bar/inner.py",
|
||||
], matched
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ─── layer 4: single-script visibility enforcement ────────────────
|
||||
|
||||
|
||||
def _make_script_for_can_view(
|
||||
*, owner_user_id: str = "U001", visibility: str = "private"
|
||||
) -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
script_id="S1",
|
||||
workspace_id="W001",
|
||||
current_object_id="O1",
|
||||
owner_user_id=owner_user_id,
|
||||
script_name="x.py",
|
||||
script_type="python",
|
||||
visibility=visibility,
|
||||
status="active",
|
||||
is_locked=0,
|
||||
)
|
||||
|
||||
|
||||
class TestScriptCanView:
|
||||
"""同一 workspace 内:owner 永远可见自己的脚本(含 private);
|
||||
其他成员只见 visibility in {workspace, public} 的脚本;
|
||||
admin 全部可见——与 data resources 的 can_view 对称。"""
|
||||
|
||||
@staticmethod
|
||||
def _viewer() -> SimpleNamespace:
|
||||
return _ctx("U002") # not the owner
|
||||
|
||||
@staticmethod
|
||||
def _owner() -> SimpleNamespace:
|
||||
return _ctx("U001")
|
||||
|
||||
@staticmethod
|
||||
def _admin() -> SimpleNamespace:
|
||||
return _ctx("U002", is_admin=True)
|
||||
|
||||
def test_owner_can_view_own_private_script(self) -> None:
|
||||
from backend.api.scripts import script_can_view
|
||||
|
||||
assert (
|
||||
script_can_view(
|
||||
_make_script_for_can_view(visibility="private"),
|
||||
self._owner(),
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
def test_workspace_member_cannot_view_others_private_script(self) -> None:
|
||||
from backend.api.scripts import script_can_view
|
||||
|
||||
assert (
|
||||
script_can_view(
|
||||
_make_script_for_can_view(visibility="private"),
|
||||
self._viewer(),
|
||||
)
|
||||
is False
|
||||
)
|
||||
|
||||
def test_workspace_member_can_view_others_workspace_script(self) -> None:
|
||||
from backend.api.scripts import script_can_view
|
||||
|
||||
assert (
|
||||
script_can_view(
|
||||
_make_script_for_can_view(visibility="workspace"),
|
||||
self._viewer(),
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
def test_workspace_member_can_view_others_public_script(self) -> None:
|
||||
from backend.api.scripts import script_can_view
|
||||
|
||||
assert (
|
||||
script_can_view(
|
||||
_make_script_for_can_view(visibility="public"),
|
||||
self._viewer(),
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
def test_admin_can_view_others_private_script(self) -> None:
|
||||
from backend.api.scripts import script_can_view
|
||||
|
||||
assert (
|
||||
script_can_view(
|
||||
_make_script_for_can_view(visibility="private"),
|
||||
self._admin(),
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
|
||||
def _script_row_for_get(script: SimpleNamespace) -> SimpleNamespace:
|
||||
"""Build the (script, storage_object) tuple get_script_row returns."""
|
||||
storage = SimpleNamespace(
|
||||
relative_path=f"workspace/{script.owner_user_id}/x.py",
|
||||
object_key=f"W001/{script.owner_user_id}/x.py",
|
||||
content_hash="h",
|
||||
size_bytes=1,
|
||||
)
|
||||
return SimpleNamespace(data=(script, storage))
|
||||
|
||||
|
||||
async def test_get_script_non_owner_private_returns_404() -> None:
|
||||
"""Non-owner must NOT read another member's private script by id
|
||||
(mirrors get_visible_resource for data resources)."""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from backend.api.scripts import get_script
|
||||
|
||||
script = _make_script_for_can_view(visibility="private")
|
||||
script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
return_value=MagicMock(one_or_none=MagicMock(return_value=(script, None)))
|
||||
)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await get_script("S1", context=_ctx("U002"), session=mock_session)
|
||||
assert exc.value.status_code == 404
|
||||
|
||||
|
||||
async def test_get_script_owner_can_read_own_private_script() -> None:
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from backend.api.scripts import get_script
|
||||
|
||||
script = _make_script_for_can_view(visibility="private")
|
||||
script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
storage = SimpleNamespace(
|
||||
relative_path="workspace/U001/x.py",
|
||||
object_key="W001/U001/x.py",
|
||||
content_hash="h",
|
||||
size_bytes=1,
|
||||
)
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
return_value=MagicMock(one_or_none=MagicMock(return_value=(script, storage)))
|
||||
)
|
||||
result = await get_script("S1", context=_ctx("U001"), session=mock_session)
|
||||
assert result["data"]["script_id"] == "S1"
|
||||
assert result["data"]["visibility"] == "private"
|
||||
|
||||
|
||||
async def test_get_script_non_owner_can_read_workspace_visible_script() -> None:
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from backend.api.scripts import get_script
|
||||
|
||||
script = _make_script_for_can_view(visibility="workspace")
|
||||
script.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
script.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
storage = SimpleNamespace(
|
||||
relative_path="workspace/U001/x.py",
|
||||
object_key="W001/U001/x.py",
|
||||
content_hash="h",
|
||||
size_bytes=1,
|
||||
)
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
return_value=MagicMock(one_or_none=MagicMock(return_value=(script, storage)))
|
||||
)
|
||||
result = await get_script("S1", context=_ctx("U002"), session=mock_session)
|
||||
assert result["data"]["script_id"] == "S1"
|
||||
assert result["data"]["visibility"] == "workspace"
|
||||
|
||||
Reference in New Issue
Block a user