tao.chen
69a9a48a0b
回退 5483d19 的 workspace-wide 改动。用户真实语义：
- owner 永远可见自己的资源（含 private）
- 非 owner 只见别人 visibility in {workspace, public} 的资源
- admin 全部可见
can_view 与 list_resources 共享同一谓词。list 端点恢复
owner == me OR visibility in {workspace, public} 的过滤，
admin 跳过。get_resource / download_url / jupyter-relative-path
/ delete_resource 全部经 get_visible_resource → can_view，自然
收敛到同一语义。补 1 个 owner-看自己-private 测试，4 个
visibility 测试保持。
725 lines
25 KiB
Python
725 lines
25 KiB
Python
"""Unit tests for data-resource path derivation and helpers.
|
|
|
|
These tests do not need a database because they exercise pure helpers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import Column, MetaData, String, Table, create_engine, select
|
|
from sqlalchemy.dialects import mysql as mysql_dialect
|
|
from backend.resources import (
|
|
_build_list_resources_descendant_prefix,
|
|
can_view,
|
|
compute_jupyter_relative_path,
|
|
resource_directory,
|
|
resource_payload,
|
|
)
|
|
from backend.services.storage import (
|
|
_object_key_tail,
|
|
_safe_file_name,
|
|
_safe_path_segment,
|
|
_strip_uniqueness_suffix,
|
|
)
|
|
|
|
|
|
_BIND_WS = "01WS0000000000000000000A"
|
|
_BIND_USER = "01USR0000000000000000000A"
|
|
|
|
|
|
class _ExecuteResult:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
|
|
def all(self):
|
|
return self._rows
|
|
|
|
|
|
class _BindSessionMock:
|
|
"""Mocked session for bind_resource.
|
|
|
|
``new_object_key`` is the StorageObjects row of the upload being bound;
|
|
``same_name_rows`` are (DataResources, StorageObjects) candidates already
|
|
in the workspace with the same resource_name.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
new_object_key: str,
|
|
same_name_rows=(),
|
|
reused_resource=None,
|
|
usage_type: str = "data_resource",
|
|
):
|
|
self._new_object_key = new_object_key
|
|
self._same_name_rows = list(same_name_rows)
|
|
self._reused_resource = reused_resource
|
|
self._usage_type = usage_type
|
|
self._scalar_calls = 0
|
|
|
|
async def scalar(self, _stmt):
|
|
from sqlalchemy import TextClause
|
|
|
|
if isinstance(_stmt, TextClause):
|
|
return 1 # GET_LOCK 成功
|
|
self._scalar_calls += 1
|
|
# 1st scalar: UploadSessions lookup; 2nd: active 绑定行复用检查。
|
|
if self._scalar_calls == 1:
|
|
return SimpleNamespace(
|
|
upload_id="01UPL0000000000000000000B",
|
|
storage_object_id="01OBJ0000000000000000000B",
|
|
workspace_id=_BIND_WS,
|
|
user_id=_BIND_USER,
|
|
usage_type=self._usage_type,
|
|
)
|
|
if self._scalar_calls == 2 and self._reused_resource is not None:
|
|
return self._reused_resource
|
|
return None
|
|
|
|
async def get(self, _model, _pk):
|
|
return _make_storage_object(self._new_object_key)
|
|
|
|
async def execute(self, _stmt):
|
|
from sqlalchemy import TextClause
|
|
|
|
if isinstance(_stmt, TextClause):
|
|
return _ExecuteResult([]) # RELEASE_LOCK
|
|
# 模拟 SQL 的 owner 过滤与 self-exclusion:只返回属于当前用户且
|
|
# 不是当前 upload 已绑定对象的同名候选。
|
|
rows = [
|
|
(resource, storage_object)
|
|
for resource, storage_object in self._same_name_rows
|
|
if resource.owner_user_id == _BIND_USER
|
|
and resource.storage_object_id != "01OBJ0000000000000000000B"
|
|
]
|
|
return _ExecuteResult(rows)
|
|
|
|
def add(self, _obj):
|
|
pass
|
|
|
|
async def flush(self):
|
|
pass
|
|
|
|
async def refresh(self, _obj, attribute_names=None):
|
|
_obj.created_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
|
|
_obj.updated_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
|
|
|
|
|
|
def _bind_context():
|
|
return SimpleNamespace(
|
|
request_id="01REQ0000000000000000000A",
|
|
user=SimpleNamespace(user_id=_BIND_USER),
|
|
workspace=SimpleNamespace(workspace_id=_BIND_WS),
|
|
)
|
|
|
|
|
|
def _bind_payload():
|
|
return SimpleNamespace(
|
|
resource_name="data.csv", description=None, visibility="private"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_rejects_duplicate_name_in_same_directory() -> None:
|
|
"""Same resource_name in the same directory raises 409."""
|
|
from backend.resources import bind_resource
|
|
|
|
existing_rows = [
|
|
(
|
|
_make_resource(_BIND_WS, _BIND_USER),
|
|
_make_storage_object(f"{_BIND_WS}/{_BIND_USER}/data.csv"),
|
|
)
|
|
]
|
|
session = _BindSessionMock(
|
|
new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
|
|
same_name_rows=existing_rows,
|
|
)
|
|
with pytest.raises(Exception) as exc_info:
|
|
await bind_resource(
|
|
upload_id="01UPL0000000000000000000B",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
assert "already exists" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_allows_same_name_in_different_directory() -> None:
|
|
"""Same resource_name in a different directory binds successfully."""
|
|
from backend.resources import bind_resource
|
|
|
|
existing_rows = [
|
|
(
|
|
_make_resource(_BIND_WS, _BIND_USER),
|
|
_make_storage_object(f"{_BIND_WS}/{_BIND_USER}/data.csv"),
|
|
)
|
|
]
|
|
session = _BindSessionMock(
|
|
new_object_key=f"{_BIND_WS}/{_BIND_USER}/subdir/data.csv",
|
|
same_name_rows=existing_rows,
|
|
)
|
|
result = await bind_resource(
|
|
upload_id="01UPL0000000000000000000B",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert result["data"]["resource_name"] == "data.csv"
|
|
assert result["data"]["jupyter_accessible_path"] == "subdir/data.csv"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_allows_same_name_when_workspace_empty() -> None:
|
|
"""No same-name rows at all: bind succeeds (root directory)."""
|
|
from backend.resources import bind_resource
|
|
|
|
session = _BindSessionMock(
|
|
new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
|
|
)
|
|
result = await bind_resource(
|
|
upload_id="01UPL0000000000000000000A",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert result["data"]["resource_name"] == "data.csv"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_allows_same_name_for_different_owner() -> None:
|
|
"""其他用户在同目录下的同名资源不阻塞当前用户的绑定。"""
|
|
from backend.resources import bind_resource
|
|
|
|
other_user = "01USR0000000000000000000B"
|
|
existing_rows = [
|
|
(
|
|
_make_resource(_BIND_WS, other_user),
|
|
_make_storage_object(f"{_BIND_WS}/{other_user}/data.csv"),
|
|
)
|
|
]
|
|
session = _BindSessionMock(
|
|
new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
|
|
same_name_rows=existing_rows,
|
|
)
|
|
result = await bind_resource(
|
|
upload_id="01UPL0000000000000000000C",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert result["data"]["resource_name"] == "data.csv"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_allows_rebinding_same_storage_object() -> None:
|
|
"""重新绑定同一 upload_id 应走 idempotent 复用路径,不触发 409。"""
|
|
from backend.resources import bind_resource
|
|
|
|
new_object_key = f"{_BIND_WS}/{_BIND_USER}/data.csv"
|
|
existing_resource = _make_resource(_BIND_WS, _BIND_USER)
|
|
existing_resource.resource_name = "data.csv"
|
|
existing_resource.storage_object_id = "01OBJ0000000000000000000B"
|
|
existing_object = _make_storage_object(new_object_key)
|
|
existing_object.storage_object_id = "01OBJ0000000000000000000B"
|
|
same_name_rows = [(existing_resource, existing_object)]
|
|
session = _BindSessionMock(
|
|
new_object_key=new_object_key,
|
|
same_name_rows=same_name_rows,
|
|
reused_resource=existing_resource,
|
|
)
|
|
result = await bind_resource(
|
|
upload_id="01UPL0000000000000000000B",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert result["meta"]["reused"] is True
|
|
assert result["data"]["resource_name"] == "data.csv"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_rejects_non_data_resource_upload() -> None:
|
|
"""其他用途(如 working_copy)的 upload session 不能 bind 成数据资源。"""
|
|
from backend.resources import bind_resource
|
|
|
|
session = _BindSessionMock(
|
|
new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
|
|
usage_type="working_copy",
|
|
)
|
|
with pytest.raises(Exception) as exc_info:
|
|
await bind_resource(
|
|
upload_id="01UPL0000000000000000000B",
|
|
payload=_bind_payload(),
|
|
request=MagicMock(),
|
|
context=_bind_context(),
|
|
session=session,
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
assert "not created for a data resource" in exc_info.value.detail
|
|
|
|
|
|
def test_safe_path_segment_cleans_special_characters():
|
|
assert _safe_path_segment("train") == "train"
|
|
assert _safe_path_segment("train v1") == "train_v1"
|
|
assert _safe_path_segment("../foo") == "foo"
|
|
assert _safe_path_segment("a/b") == "a_b"
|
|
assert _safe_path_segment("...") == "untitled"
|
|
|
|
|
|
def test_safe_file_name_rejects_traversal_and_hidden():
|
|
assert _safe_file_name("data.csv") == "data.csv"
|
|
with pytest.raises(Exception):
|
|
_safe_file_name("../data.csv")
|
|
with pytest.raises(Exception):
|
|
_safe_file_name(".hidden.csv")
|
|
|
|
|
|
def _make_resource(workspace_id: str, owner_user_id: str):
|
|
return SimpleNamespace(
|
|
resource_id="01RES0000000000000000000A",
|
|
workspace_id=workspace_id,
|
|
storage_object_id="01OBJ0000000000000000000A",
|
|
owner_user_id=owner_user_id,
|
|
resource_name="sample",
|
|
description=None,
|
|
visibility="workspace",
|
|
status="active",
|
|
created_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None),
|
|
updated_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None),
|
|
)
|
|
|
|
|
|
def _make_storage_object(object_key: str):
|
|
return SimpleNamespace(
|
|
storage_object_id="01OBJ0000000000000000000A",
|
|
object_key=object_key,
|
|
file_name="data.csv",
|
|
file_extension=".csv",
|
|
mime_type="text/csv",
|
|
size_bytes=42,
|
|
content_hash="a" * 64,
|
|
object_status="available",
|
|
)
|
|
|
|
|
|
def test_resource_payload_legacy_dot_resources():
|
|
ws = "01WS00000000000000000000A"
|
|
user = "01USR000000000000000000A"
|
|
payload = resource_payload(
|
|
_make_resource(ws, user),
|
|
_make_storage_object(f"{ws}/{user}/.resources/data.csv"),
|
|
)
|
|
assert payload["jupyter_accessible_path"] == ".resources/data.csv"
|
|
assert payload["absolute_path"].endswith(f"{ws}/{user}/.resources/data.csv")
|
|
|
|
|
|
def test_resource_payload_new_flat_path():
|
|
ws = "01WS00000000000000000000A"
|
|
user = "01USR000000000000000000A"
|
|
payload = resource_payload(
|
|
_make_resource(ws, user),
|
|
_make_storage_object(f"{ws}/{user}/data.csv"),
|
|
)
|
|
assert payload["jupyter_accessible_path"] == "data.csv"
|
|
assert payload["absolute_path"].endswith(f"{ws}/{user}/data.csv")
|
|
|
|
|
|
def test_resource_payload_new_nested_path():
|
|
ws = "01WS00000000000000000000A"
|
|
user = "01USR000000000000000000A"
|
|
payload = resource_payload(
|
|
_make_resource(ws, user),
|
|
_make_storage_object(f"{ws}/{user}/train/v1/data.csv"),
|
|
)
|
|
assert payload["jupyter_accessible_path"] == "train/v1/data.csv"
|
|
assert payload["absolute_path"].endswith(f"{ws}/{user}/train/v1/data.csv")
|
|
|
|
|
|
def test_compute_jupyter_relative_path_for_legacy_and_new_paths():
|
|
# Legacy .resources path still resolves correctly.
|
|
assert compute_jupyter_relative_path("notebooks/exp.ipynb", ".resources/data.csv") == "../.resources/data.csv"
|
|
# New nested path resolves relative to the script directory.
|
|
assert compute_jupyter_relative_path("notebooks/exp.ipynb", "train/v1/data.csv") == "../train/v1/data.csv"
|
|
# Same-directory script.
|
|
assert compute_jupyter_relative_path("exp.ipynb", "data.csv") == "data.csv"
|
|
|
|
|
|
def test_resource_directory_parses_object_key():
|
|
ws = "01WS00000000000000000000A"
|
|
user = "01USR000000000000000000A"
|
|
# 根目录文件:目录为 ""。
|
|
assert resource_directory(f"{ws}/{user}/data.csv", ws, user) == ""
|
|
# 子目录 / 多级目录。
|
|
assert resource_directory(f"{ws}/{user}/sub/data.csv", ws, user) == "sub"
|
|
assert (
|
|
resource_directory(f"{ws}/{user}/train/v1/data.csv", ws, user)
|
|
== "train/v1"
|
|
)
|
|
# 旧版 .resources 布局:目录为 ".resources"。
|
|
assert (
|
|
resource_directory(f"{ws}/{user}/.resources/data.csv", ws, user)
|
|
== ".resources"
|
|
)
|
|
# 不匹配 ws/user 前缀的键按根目录处理。
|
|
assert resource_directory("other-bucket-key.csv", ws, user) == ""
|
|
|
|
|
|
def test_object_key_tail_and_strip_uniqueness_suffix():
|
|
ws = "01WS00000000000000000000A"
|
|
user = "01USR000000000000000000A"
|
|
# 前缀剥离。
|
|
assert _object_key_tail(f"{ws}/{user}/sub/data.csv", ws, user) == "sub/data.csv"
|
|
assert _object_key_tail("no-prefix.csv", ws, user) == "no-prefix.csv"
|
|
# 带唯一化后缀的 key 恢复为原始路径(26 位 ULID 后缀)。
|
|
suffix = "01arz3ndektsv4rrffq69g5fav" # 26 chars
|
|
assert (
|
|
_strip_uniqueness_suffix(f"sub/data-{suffix}.csv") == "sub/data.csv"
|
|
)
|
|
assert _strip_uniqueness_suffix(f"data-{suffix}.csv") == "data.csv"
|
|
assert _strip_uniqueness_suffix(f"data-{suffix}") == "data"
|
|
# 普通文件名不受影响(短后缀、无连字符、目录含连字符)。
|
|
assert _strip_uniqueness_suffix("sub/data.csv") == "sub/data.csv"
|
|
assert _strip_uniqueness_suffix("data-v1.csv") == "data-v1.csv"
|
|
assert _strip_uniqueness_suffix("my-dir/data.csv") == "my-dir/data.csv"
|
|
assert _strip_uniqueness_suffix("data") == "data"
|
|
|
|
|
|
def test_data_resources_model_allows_duplicate_storage_object_reference() -> None:
|
|
"""With uk_data_resources_object dropped, no unique index covers
|
|
storage_object_id, so multiple DataResources rows may reference the
|
|
same storage object.
|
|
"""
|
|
from common.db.models import DataResources
|
|
|
|
index_names = {idx.name for idx in DataResources.__table__.indexes}
|
|
assert "uk_data_resources_object" not in index_names
|
|
for idx in DataResources.__table__.indexes:
|
|
if idx.unique:
|
|
cols = {c.name for c in idx.columns}
|
|
assert "storage_object_id" not in cols, (
|
|
f"unexpected unique index {idx.name} on storage_object_id"
|
|
)
|
|
|
|
|
|
# ─── parent_path filtering (mirrors test_list_scripts_parent_path.py) ────────
|
|
|
|
|
|
def _resource_ctx(workspace_id: str = "W001") -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
request_id="test",
|
|
user=SimpleNamespace(user_id="U001"),
|
|
workspace=SimpleNamespace(workspace_id=workspace_id),
|
|
is_admin=False,
|
|
)
|
|
|
|
|
|
class TestCanViewVisibility:
|
|
"""同一 workspace 内：owner 永远可见自己的资源（含 private）；
|
|
其他成员只见 visibility in {workspace, public} 的资源；
|
|
admin 全部可见。与 list_resources 的 SQL 谓词一致。"""
|
|
|
|
@staticmethod
|
|
def _viewer() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
request_id="test",
|
|
user=SimpleNamespace(user_id="U002"), # not the owner
|
|
workspace=SimpleNamespace(workspace_id="W001"),
|
|
is_admin=False,
|
|
)
|
|
|
|
@staticmethod
|
|
def _owner() -> SimpleNamespace:
|
|
ctx = TestCanViewVisibility._viewer()
|
|
ctx.user = SimpleNamespace(user_id="U001") # the owner
|
|
return ctx
|
|
|
|
@staticmethod
|
|
def _admin() -> SimpleNamespace:
|
|
ctx = TestCanViewVisibility._viewer()
|
|
ctx.is_admin = True
|
|
return ctx
|
|
|
|
@staticmethod
|
|
def _resource(visibility: str) -> SimpleNamespace:
|
|
res = _make_resource("W001", "U001")
|
|
res.visibility = visibility
|
|
return res
|
|
|
|
def test_owner_can_view_own_private_resource(self) -> None:
|
|
assert (
|
|
can_view(
|
|
self._resource("private"),
|
|
self._owner(),
|
|
)
|
|
is True
|
|
)
|
|
|
|
def test_workspace_member_cannot_view_others_private_resource(self) -> None:
|
|
assert (
|
|
can_view(
|
|
self._resource("private"),
|
|
self._viewer(),
|
|
)
|
|
is False
|
|
)
|
|
|
|
def test_workspace_member_can_view_others_workspace_resource(self) -> None:
|
|
assert (
|
|
can_view(
|
|
self._resource("workspace"),
|
|
self._viewer(),
|
|
)
|
|
is True
|
|
)
|
|
|
|
def test_workspace_member_can_view_others_public_resource(self) -> None:
|
|
assert (
|
|
can_view(
|
|
self._resource("public"),
|
|
self._viewer(),
|
|
)
|
|
is True
|
|
)
|
|
|
|
def test_admin_can_view_others_private_resource(self) -> None:
|
|
assert (
|
|
can_view(
|
|
self._resource("private"),
|
|
self._admin(),
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
class TestBuildListResourcesDescendantPrefix:
|
|
"""Pure-function contract for the escaped object_key prefix. Unlike
|
|
the scripts helper there is NO user scope — data resources are
|
|
workspace-wide, so the prefix is just the normalized+escaped path."""
|
|
|
|
def test_empty_parent_path_returns_empty_prefix(self) -> None:
|
|
assert _build_list_resources_descendant_prefix("") == ""
|
|
|
|
def test_subdir_prefix_appends_trailing_slash(self) -> None:
|
|
assert _build_list_resources_descendant_prefix("foo/bar") == "foo/bar/"
|
|
|
|
def test_escapes_underscore(self) -> None:
|
|
assert _build_list_resources_descendant_prefix("foo_bar") == r"foo\_bar/"
|
|
|
|
def test_escapes_percent(self) -> None:
|
|
assert _build_list_resources_descendant_prefix("100%match") == r"100\%match/"
|
|
|
|
def test_normalizes_leading_trailing_slashes(self) -> None:
|
|
assert _build_list_resources_descendant_prefix("/foo/bar/") == "foo/bar/"
|
|
|
|
def test_rejects_traversal(self) -> None:
|
|
with pytest.raises(HTTPException) as exc:
|
|
_build_list_resources_descendant_prefix("foo/../bar")
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
# ─── layer 2: SQL contract (mock session + mysql dialect compile) ────────────
|
|
|
|
|
|
def _compile_sql(stmt) -> str:
|
|
return str(
|
|
stmt.compile(
|
|
dialect=mysql_dialect.dialect(),
|
|
compile_kwargs={"literal_binds": True},
|
|
)
|
|
)
|
|
|
|
|
|
class _ListResourcesMockResult:
|
|
def all(self):
|
|
return []
|
|
|
|
|
|
def _list_resources_capturing_session(captured_sql: list[str]) -> MagicMock:
|
|
mock_session = MagicMock()
|
|
mock_session.execute = AsyncMock(
|
|
side_effect=lambda stmt: (
|
|
captured_sql.append(_compile_sql(stmt)) or _ListResourcesMockResult()
|
|
)
|
|
)
|
|
return mock_session
|
|
|
|
|
|
async def test_list_resources_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
|
|
from backend.resources import list_resources
|
|
|
|
captured_sql: list[str] = []
|
|
mock_session = _list_resources_capturing_session(captured_sql)
|
|
|
|
await list_resources(
|
|
parent_path="foo/bar",
|
|
context=_resource_ctx(),
|
|
session=mock_session,
|
|
visibility=None,
|
|
keyword=None,
|
|
)
|
|
|
|
assert len(captured_sql) == 1
|
|
sql = captured_sql[0].lower()
|
|
# 中间 ``%`` 是 owner 段通配符(跨所有 owner),父路径本身按字面匹配。
|
|
assert "like 'w001/%%/foo/bar/%%'" in sql
|
|
assert "not like 'w001/%%/foo/bar/%%/%%'" in sql
|
|
|
|
|
|
async def test_list_resources_where_clause_escapes_underscore() -> None:
|
|
"""Regression: parent_path containing ``_`` MUST be escaped in the
|
|
compiled LIKE pattern, otherwise sibling-path leak (``fooXbar``) returns."""
|
|
from backend.resources import list_resources
|
|
|
|
captured_sql: list[str] = []
|
|
mock_session = _list_resources_capturing_session(captured_sql)
|
|
|
|
await list_resources(
|
|
parent_path="foo_bar",
|
|
context=_resource_ctx(),
|
|
session=mock_session,
|
|
visibility=None,
|
|
keyword=None,
|
|
)
|
|
sql = captured_sql[0]
|
|
sql_lower = sql.lower()
|
|
# SQLAlchemy doubles the escape char inside the SQL string literal, so
|
|
# the helper's ``foo\_bar`` renders as ``foo\\_bar`` in the SQL text.
|
|
assert r"like 'w001/%%/foo\\_bar/%%'" in sql_lower
|
|
assert r"not like 'w001/%%/foo\\_bar/%%/%%'" in sql_lower
|
|
# Both LIKE clauses declare ESCAPE '\\' (two in total).
|
|
assert sql.count("ESCAPE '\\\\'") == 2, sql
|
|
|
|
|
|
async def test_list_resources_without_parent_path_adds_no_like_clause() -> None:
|
|
"""Empty parent_path keeps the legacy workspace-wide behaviour — no
|
|
object_key LIKE filter at all."""
|
|
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].lower()
|
|
assert " like " not in sql
|
|
assert " not like " not in sql
|
|
|
|
|
|
# ─── layer 3: behavioral test on real LIKE execution (SQLite) ────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def sqlite_object_key_table():
|
|
"""SQLite in-memory stand-in for ``storage_objects.object_key`` rows.
|
|
|
|
Three-layer structure: direct children under ``data`` owned by two
|
|
different users (cross-owner), a deeper descendant, a sibling folder,
|
|
a root file, plus ``data_x`` and its ``dataXx`` / ``data2x`` decoys.
|
|
"""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
metadata = MetaData()
|
|
table = Table(
|
|
"object_keys",
|
|
metadata,
|
|
Column("object_key", String(1024), nullable=False),
|
|
)
|
|
metadata.create_all(engine)
|
|
rows = [
|
|
"W001/U001/data/alpha.csv", # direct child (owner U001)
|
|
"W001/U002/data/beta.csv", # direct child (owner U002)
|
|
"W001/U001/data/deep/nested.csv", # deeper descendant
|
|
"W001/U001/database/gamma.csv", # sibling folder
|
|
"W001/U001/root.csv", # root file
|
|
"W001/U001/data_x/delta.csv", # target for parent_path=data_x
|
|
"W001/U001/dataXx/decoy.csv", # sibling decoy (unescaped match)
|
|
"W001/U001/data2x/decoy2.csv", # sibling decoy (unescaped match)
|
|
]
|
|
with engine.begin() as conn:
|
|
conn.execute(table.insert(), [{"object_key": key} for key in rows])
|
|
yield engine, table
|
|
engine.dispose()
|
|
|
|
|
|
def test_sqlite_direct_children_across_owners(sqlite_object_key_table) -> None:
|
|
"""parent_path='data' returns exactly the direct children under data,
|
|
across every owner, excluding deeper/sibling/root paths."""
|
|
engine, table = sqlite_object_key_table
|
|
prefix = _build_list_resources_descendant_prefix("data")
|
|
like = f"W001/%/{prefix}%"
|
|
not_like = f"W001/%/{prefix}%/%"
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(
|
|
select(table.c.object_key).where(
|
|
table.c.object_key.like(like, escape="\\"),
|
|
~table.c.object_key.like(not_like, escape="\\"),
|
|
)
|
|
).fetchall()
|
|
matched = sorted(r[0] for r in rows)
|
|
assert matched == [
|
|
"W001/U001/data/alpha.csv",
|
|
"W001/U002/data/beta.csv",
|
|
], matched
|
|
|
|
|
|
def test_sqlite_escaped_underscore_does_not_match_sibling(
|
|
sqlite_object_key_table,
|
|
) -> None:
|
|
"""parent_path='data_x' must match only the literal data_x folder, not
|
|
the dataXx / data2x siblings an unescaped pattern would match."""
|
|
engine, table = sqlite_object_key_table
|
|
prefix = _build_list_resources_descendant_prefix("data_x")
|
|
like = f"W001/%/{prefix}%"
|
|
not_like = f"W001/%/{prefix}%/%"
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(
|
|
select(table.c.object_key).where(
|
|
table.c.object_key.like(like, escape="\\"),
|
|
~table.c.object_key.like(not_like, escape="\\"),
|
|
)
|
|
).fetchall()
|
|
matched = sorted(r[0] for r in rows)
|
|
assert matched == ["W001/U001/data_x/delta.csv"], matched
|
|
|
|
|
|
def test_sqlite_parent_path_with_slash_direct_children(sqlite_object_key_table) -> None:
|
|
"""parent_path='data/deep' returns only data/deep/* — never data/*."""
|
|
engine, table = sqlite_object_key_table
|
|
prefix = _build_list_resources_descendant_prefix("data/deep")
|
|
like = f"W001/%/{prefix}%"
|
|
not_like = f"W001/%/{prefix}%/%"
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(
|
|
select(table.c.object_key).where(
|
|
table.c.object_key.like(like, escape="\\"),
|
|
~table.c.object_key.like(not_like, escape="\\"),
|
|
)
|
|
).fetchall()
|
|
matched = sorted(r[0] for r in rows)
|
|
assert matched == ["W001/U001/data/deep/nested.csv"], matched
|
|
|
|
|
|
def test_sqlite_empty_parent_path_has_no_like_filter(sqlite_object_key_table) -> None:
|
|
"""Empty parent_path → no LIKE filter → the endpoint's base WHERE only
|
|
(workspace-wide active resources). Stand-in: every row is returned."""
|
|
engine, table = sqlite_object_key_table
|
|
with engine.connect() as conn:
|
|
rows = conn.execute(select(table.c.object_key)).fetchall()
|
|
matched = sorted(r[0] for r in rows)
|
|
assert len(matched) == 8, matched
|