Files
model-platform/backend/tests/test_resources.py
T
tao.chen b6eb069849 fix(backend): enhance storage/resource concurrency, idempotency, and deletion safety
Summary of changes:

- Resources Bind Idempotency (High 1+2):
  - Check existing active storage object bindings before duplicate check.
  - Return existing binding (`reused: true`) on retry with same upload_id.
  - Filter by `status == "active"` to bypass dead/deleted rows during reuse check.

- Storage Concurrent Overwrite (High 3):
  - Add `acquire_named_lock` and `release_named_lock` helpers using MySQL `GET_LOCK`/`RELEASE_LOCK` hashed to <= 64 chars.
  - Wrap `upload_bytes_to_session` PUT+INSERT critical section with named lock on `object_key`.
  - Re-check key collision inside lock; append ULID suffix on collision.

- Shared Reference Deletion Protection (Medium 4):
  - Check active references before deleting storage objects in `delete_resource`.
  - Delete only `DataResources` record if storage object is still referenced elsewhere.

- Robust Usage Type Fallback (Medium 5):
  - Replace direct dict lookup for `USAGE_TYPE_TO_PURPOSE[item.usage_type]` with `.get(..., "workspace")` default.

- Idempotency Key Path Matching (Medium 6):
  - Move `file_name`/`target_path` validation forward and include path dimension in comparison.
  - Strip uniqueness suffix via `_strip_uniqueness_suffix` before key comparison to avoid false 409s on valid retries.

- Usage Type & Bind Concurrency Control (Low 7 & 8):
  - Reject bind requests with 409 if upload session purpose is not `data_resource`.
  - Wrap resource duplicate check and creation in named lock using `(owner, directory, name)`.

- Trash Key Uniqueness & Restore Compatibility (Low 9):
  - Update `trash_key` format to `{purpose}/{object_key}-{storage_object_id}` to prevent collisions.
  - Update `object_key_hash` on trash move.
  - Update restore logic in `storage_api.py` to strip suffix while maintaining backward compatibility with legacy keys.

- Dead Code Removal (Low 10):
  - Remove unreachable `upload_status = "failed"` and redundant `session.rollback()` in `IntegrityError` block.

- Tests & Mocks:
  - Add/update 5 test cases covering non-data_resource bind rejection, suffix stripping, and path recovery.
  - Add named lock statement mocks for DB testing.
2026-08-14 20:53:07 +08:00

408 lines
14 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 MagicMock
import pytest
from backend.resources import (
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"
)