519 lines
18 KiB
Python
519 lines
18 KiB
Python
"""Tests for script storage-layer behavior after unique-index removal.
|
||
|
||
These are intentionally unit-level: they mock the async SQLAlchemy session
|
||
and the Jupyter runtime client so the suite stays fast and does not need a
|
||
live database. The tests verify the code-level guarantees that back the
|
||
"delete then re-upload" flow:
|
||
|
||
* StorageObjects is flushed before Scripts on creation.
|
||
* All soft-delete paths flip ``is_deleted = 1`` alongside ``status='deleted'``
|
||
/ ``object_status='deleted'``.
|
||
* The ORM models no longer declare the dropped unique indexes.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from common.config import settings
|
||
from common.db.models import DataResources, Scripts, StorageObjects
|
||
|
||
|
||
def _index_names(table) -> set[str]:
|
||
return {idx.name for idx in table.indexes}
|
||
|
||
|
||
def test_scripts_model_dropped_unique_indexes() -> None:
|
||
"""The dropped unique indexes must not be declared on the Scripts model."""
|
||
names = _index_names(Scripts.__table__)
|
||
assert "uk_scripts_current_object" not in names
|
||
assert "uk_scripts_workspace_name" not in names
|
||
assert "uk_scripts_workspace_name_active" not in names
|
||
|
||
|
||
def test_data_resources_model_dropped_unique_index() -> None:
|
||
"""The dropped unique index must not be declared on DataResources."""
|
||
names = _index_names(DataResources.__table__)
|
||
assert "uk_data_resources_object" not in names
|
||
|
||
|
||
def test_storage_objects_unique_index_is_active_only() -> None:
|
||
"""After dropping uk_storage_bucket_key and adding
|
||
uk_storage_bucket_key_active (conditional UNIQUE via generated column),
|
||
the only physical unique index on storage_objects covers only the
|
||
`object_status='available'` subset (NULL-permissive slot for soft-deleted).
|
||
|
||
The workspace path lookup index is non-unique; path uniqueness for
|
||
active objects is enforced by the application-level conflict check
|
||
in ``scripts.py`` and ``services.storage._resolve_unique_object_key``.
|
||
"""
|
||
indexes_by_name = {idx.name: idx for idx in StorageObjects.__table__.indexes}
|
||
assert "uk_storage_bucket_key" not in indexes_by_name
|
||
assert "uk_storage_bucket_key_active" in indexes_by_name
|
||
assert indexes_by_name["uk_storage_bucket_key_active"].unique is True
|
||
# The active-column index must include the generated column.
|
||
assert any(
|
||
col.name == "object_key_hash_active"
|
||
for col in indexes_by_name["uk_storage_bucket_key_active"].columns
|
||
)
|
||
|
||
|
||
def _make_context() -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
request_id="01REQ0000000000000000000A",
|
||
user=SimpleNamespace(user_id="01USR0000000000000000000A"),
|
||
workspace=SimpleNamespace(workspace_id="01WS0000000000000000000A"),
|
||
is_admin=False,
|
||
)
|
||
|
||
|
||
def _make_request() -> MagicMock:
|
||
request = MagicMock()
|
||
request.app.state.runtime_client.delete_file = AsyncMock(return_value=None)
|
||
return request
|
||
|
||
|
||
class _AsyncSessionMock:
|
||
"""Minimal AsyncSession stand-in that records flush/add order."""
|
||
|
||
def __init__(self) -> None:
|
||
self.added: list[object] = []
|
||
self.flush_order: list[str] = []
|
||
self._refreshed: list[object] = []
|
||
self.scalar_results: list[Any] = []
|
||
|
||
def add(self, obj: object) -> None:
|
||
self.added.append(obj)
|
||
|
||
async def flush(self) -> None:
|
||
# Record the kind of object that triggered this flush.
|
||
self.flush_order.append(type(self.added[-1]).__name__)
|
||
|
||
async def refresh(self, obj: object, attribute_names: list[str] | None = None) -> None:
|
||
self._refreshed.append(obj)
|
||
|
||
async def scalar(self, *_args, **_kwargs) -> None:
|
||
if self.scalar_results:
|
||
return self.scalar_results.pop(0)
|
||
return None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_script_record_flushes_storage_object_before_script() -> None:
|
||
"""StorageObjects must flush first so path conflicts surface early."""
|
||
from backend.scripts import create_script_record
|
||
|
||
session = _AsyncSessionMock()
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
runtime_client = request.app.state.runtime_client
|
||
runtime_client.ensure_directory = AsyncMock(return_value=None)
|
||
runtime_client.create_notebook = AsyncMock(return_value={"name": "test.ipynb"})
|
||
|
||
script, storage_object = await create_script_record(
|
||
name="test.ipynb",
|
||
script_type="notebook",
|
||
content=b'{"cells": []}',
|
||
visibility="private",
|
||
parent_path=None,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
assert isinstance(storage_object, StorageObjects)
|
||
assert isinstance(script, Scripts)
|
||
assert session.flush_order == ["StorageObjects", "Scripts"]
|
||
assert session._refreshed == [storage_object, script]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_script_record_storage_object_flush_failure_does_not_add_script() -> None:
|
||
"""If the StorageObjects flush fails, the Scripts row must never be added."""
|
||
from backend.scripts import create_script_record
|
||
|
||
class FailingSession(_AsyncSessionMock):
|
||
async def flush(self) -> None:
|
||
self.flush_order.append(type(self.added[-1]).__name__)
|
||
raise RuntimeError("duplicate path")
|
||
|
||
session = FailingSession()
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
runtime_client = request.app.state.runtime_client
|
||
runtime_client.ensure_directory = AsyncMock(return_value=None)
|
||
runtime_client.create_notebook = AsyncMock(return_value={"name": "test.ipynb"})
|
||
|
||
with pytest.raises(RuntimeError, match="duplicate path"):
|
||
await create_script_record(
|
||
name="test.ipynb",
|
||
script_type="notebook",
|
||
content=b'{"cells": []}',
|
||
visibility="private",
|
||
parent_path=None,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
# Only the StorageObjects row was ever added.
|
||
assert len(session.added) == 1
|
||
assert isinstance(session.added[0], StorageObjects)
|
||
# The Jupyter cleanup was attempted.
|
||
request.app.state.runtime_client.delete_file.assert_awaited_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_script_record_allows_reupload_after_delete() -> None:
|
||
"""Without uk_scripts_workspace_name_active, re-uploading a script with
|
||
the same name after the previous one was soft-deleted succeeds.
|
||
"""
|
||
from backend.scripts import create_script_record
|
||
|
||
session = _AsyncSessionMock()
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
runtime_client = request.app.state.runtime_client
|
||
runtime_client.ensure_directory = AsyncMock(return_value=None)
|
||
runtime_client.create_notebook = AsyncMock(return_value={"name": "test.ipynb"})
|
||
|
||
script1, storage_object1 = await create_script_record(
|
||
name="test.ipynb",
|
||
script_type="notebook",
|
||
content=b'{"cells": []}',
|
||
visibility="private",
|
||
parent_path=None,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
# Simulate the first script and its storage object being soft-deleted.
|
||
script1.status = "deleted"
|
||
script1.is_deleted = 1
|
||
storage_object1.object_status = "deleted"
|
||
storage_object1.is_deleted = 1
|
||
|
||
# Re-upload with the same name: the Scripts layer must not reject it.
|
||
script2, storage_object2 = await create_script_record(
|
||
name="test.ipynb",
|
||
script_type="notebook",
|
||
content=b'{"cells": []}',
|
||
visibility="private",
|
||
parent_path=None,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
assert script2.script_name == script1.script_name
|
||
assert script2.script_id != script1.script_id
|
||
assert script2.status == "active"
|
||
assert storage_object2.storage_object_id != storage_object1.storage_object_id
|
||
# Each upload flushes its own StorageObject then Script.
|
||
assert session.flush_order == [
|
||
"StorageObjects",
|
||
"Scripts",
|
||
"StorageObjects",
|
||
"Scripts",
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_script_record_allows_same_name_different_parent() -> None:
|
||
"""Same filename in different parent directories of the same user
|
||
must coexist — they correspond to different Jupyter paths
|
||
(/user/foo.ipynb vs /user/test/foo.ipynb).
|
||
"""
|
||
from backend.scripts import create_script_record
|
||
|
||
session = _AsyncSessionMock()
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
runtime_client = request.app.state.runtime_client
|
||
runtime_client.ensure_directory = AsyncMock(return_value=None)
|
||
runtime_client.create_notebook = AsyncMock(return_value={"name": "x.ipynb"})
|
||
|
||
s1, so1 = await create_script_record(
|
||
name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
|
||
visibility="private", parent_path=None,
|
||
request=request, context=context, session=session,
|
||
)
|
||
|
||
# Prime the mock to return a directory row for the upcoming parent lookup.
|
||
user_prefix = f"workspace/{context.user.user_id}"
|
||
session.scalar_results.append(
|
||
StorageObjects(
|
||
storage_object_id="01OBJ0000000000000000AB",
|
||
workspace_id=context.workspace.workspace_id,
|
||
owner_user_id=context.user.user_id,
|
||
object_type="directory",
|
||
usage_type="working_copy",
|
||
storage_backend=settings.storage_backend,
|
||
storage_uri="s3://bucket/key",
|
||
file_name="test",
|
||
created_by=context.user.user_id,
|
||
relative_path=f"{user_prefix}/test",
|
||
)
|
||
)
|
||
|
||
# Second upload with the same name but in a subdirectory must NOT 409.
|
||
s2, so2 = await create_script_record(
|
||
name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
|
||
visibility="private", parent_path="test",
|
||
request=request, context=context, session=session,
|
||
)
|
||
|
||
assert s1.script_id != s2.script_id
|
||
assert so1.storage_object_id != so2.storage_object_id
|
||
# Different relative_paths because parent_path differs.
|
||
assert so1.relative_path != so2.relative_path
|
||
assert so1.relative_path.endswith("x.ipynb")
|
||
assert so2.relative_path.endswith("test/x.ipynb")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_script_after_soft_delete_does_not_conflict() -> None:
|
||
"""Resurrection regression: with uk_storage_bucket_key_active being a
|
||
conditional UNIQUE (NULL when object_status != 'available'), re-uploading
|
||
a script whose previous StorageObjects row is soft-deleted does NOT
|
||
raise IntegrityError — the generated column is NULL for the deleted row,
|
||
so it does not occupy the UNIQUE slot.
|
||
"""
|
||
from backend.scripts import create_script_record
|
||
|
||
session = _AsyncSessionMock()
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
runtime_client = request.app.state.runtime_client
|
||
runtime_client.ensure_directory = AsyncMock(return_value=None)
|
||
runtime_client.create_notebook = AsyncMock(return_value={"name": "x.ipynb"})
|
||
|
||
s1, so1 = await create_script_record(
|
||
name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
|
||
visibility="private", parent_path=None,
|
||
request=request, context=context, session=session,
|
||
)
|
||
s1.status = "deleted"
|
||
s1.is_deleted = 1
|
||
so1.object_status = "deleted"
|
||
so1.is_deleted = 1
|
||
|
||
s2, so2 = await create_script_record(
|
||
name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
|
||
visibility="private", parent_path=None,
|
||
request=request, context=context, session=session,
|
||
)
|
||
assert s2.script_id != s1.script_id
|
||
assert so2.storage_object_id != so1.storage_object_id
|
||
# Same object_key (no ULID suffix) — soft-deleted row excluded from
|
||
# _resolve_unique_object_key's "available" filter.
|
||
assert so2.object_key == so1.object_key
|
||
|
||
|
||
def test_scripts_model_allows_duplicate_active_name() -> None:
|
||
"""With uk_scripts_workspace_name_active dropped, no unique index covers
|
||
(workspace_id, script_name, script_type), so duplicate active names are
|
||
allowed at the ORM level.
|
||
"""
|
||
for idx in Scripts.__table__.indexes:
|
||
if idx.unique:
|
||
cols = {c.name for c in idx.columns}
|
||
assert not (
|
||
{"workspace_id", "script_name", "script_type"} <= cols
|
||
), f"unexpected unique index {idx.name} on script name"
|
||
|
||
|
||
def test_scripts_model_allows_duplicate_current_object_id() -> None:
|
||
"""With uk_scripts_current_object dropped, no unique index covers
|
||
current_object_id, so multiple scripts may point to the same storage object.
|
||
"""
|
||
names = _index_names(Scripts.__table__)
|
||
assert "uk_scripts_current_object" not in names
|
||
for idx in Scripts.__table__.indexes:
|
||
if idx.unique and len(idx.columns) == 1:
|
||
assert "current_object_id" not in {c.name for c in idx.columns}
|
||
|
||
|
||
def _script_row() -> Scripts:
|
||
return Scripts(
|
||
script_id="01SCR0000000000000000000A",
|
||
workspace_id="01WS0000000000000000000A",
|
||
current_object_id="01OBJ0000000000000000000A",
|
||
owner_user_id="01USR0000000000000000000A",
|
||
script_name="test.ipynb",
|
||
script_type="notebook",
|
||
visibility="private",
|
||
status="active",
|
||
)
|
||
|
||
|
||
def _storage_object_row() -> StorageObjects:
|
||
return StorageObjects(
|
||
storage_object_id="01OBJ0000000000000000000A",
|
||
workspace_id="01WS0000000000000000000A",
|
||
object_type="file",
|
||
usage_type="working_copy",
|
||
# 与生产构造器一致（common/db/models/storage.py 注释 "s3"）
|
||
storage_backend=settings.storage_backend,
|
||
storage_uri="s3://bucket/key",
|
||
file_name="test.ipynb",
|
||
created_by="01USR0000000000000000000A",
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_script_route_sets_is_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Soft-deleting a script via the route handler flips is_deleted=1."""
|
||
from backend.scripts import delete_script
|
||
|
||
script = _script_row()
|
||
storage_object = _storage_object_row()
|
||
|
||
request = _make_request()
|
||
context = _make_context()
|
||
session = AsyncMock()
|
||
|
||
async def _fake_get_script_row(
|
||
_script_id: str,
|
||
_context: SimpleNamespace,
|
||
_session: AsyncMock,
|
||
*,
|
||
for_update: bool = False,
|
||
allow_missing_storage_object: bool = False,
|
||
) -> tuple[Scripts, StorageObjects]:
|
||
return script, storage_object
|
||
|
||
monkeypatch.setattr("backend.scripts.get_script_row", _fake_get_script_row)
|
||
mock_soft_delete = AsyncMock(
|
||
return_value={
|
||
"data": {
|
||
"storage_object_id": storage_object.storage_object_id,
|
||
"object_status": "deleted",
|
||
"trash_key": "trash/bucket/key",
|
||
"trash_bucket": "trash",
|
||
}
|
||
}
|
||
)
|
||
monkeypatch.setattr("backend.scripts.soft_delete_object", mock_soft_delete)
|
||
|
||
result = await delete_script(
|
||
script_id=script.script_id,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
assert script.status == "deleted"
|
||
assert script.is_deleted == 1
|
||
assert script.deleted_at is not None
|
||
assert storage_object.object_status == "deleted"
|
||
assert storage_object.is_deleted == 1
|
||
assert storage_object.deleted_at is not None
|
||
assert result["data"]["status"] == "deleted"
|
||
mock_soft_delete.assert_awaited_once_with(
|
||
storage_object.storage_object_id, request, session
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_resource_sets_is_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""Soft-deleting a data resource must write is_deleted=1 on the row."""
|
||
from backend.resources import delete_resource
|
||
|
||
resource = DataResources(
|
||
resource_id="01RES0000000000000000000A",
|
||
workspace_id="01WS0000000000000000000A",
|
||
storage_object_id="01OBJ0000000000000000000A",
|
||
owner_user_id="01USR0000000000000000000A",
|
||
resource_name="data.csv",
|
||
visibility="workspace", # visible so can_view passes
|
||
status="active",
|
||
)
|
||
storage_object = _storage_object_row()
|
||
|
||
request = _make_request()
|
||
context = _make_context()
|
||
|
||
class _Result:
|
||
def one_or_none(self):
|
||
return (resource, storage_object)
|
||
|
||
session = AsyncMock()
|
||
session.execute = AsyncMock(return_value=_Result())
|
||
|
||
with monkeypatch.context() as mp:
|
||
mp.setattr(
|
||
"backend.resources.soft_delete_object",
|
||
AsyncMock(return_value={"data": {}}),
|
||
)
|
||
result = await delete_resource(
|
||
resource_id=resource.resource_id,
|
||
request=request,
|
||
context=context,
|
||
session=session,
|
||
)
|
||
|
||
assert resource.status == "deleted"
|
||
assert resource.is_deleted == 1
|
||
assert resource.deleted_at is not None
|
||
assert result["data"]["status"] == "deleted"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_soft_delete_object_sets_is_deleted() -> None:
|
||
"""The shared helper must flip is_deleted=1 on StorageObjects."""
|
||
from backend.services.storage import soft_delete_object
|
||
|
||
item = _storage_object_row()
|
||
item.storage_backend = "local" # 测的是"非默认后端短路 trash、只翻 DB"路径
|
||
item.object_status = "available"
|
||
|
||
request = MagicMock()
|
||
request.app.state.object_stores = {}
|
||
|
||
session = AsyncMock()
|
||
session.scalar = AsyncMock(return_value=item)
|
||
|
||
result = await soft_delete_object(
|
||
storage_object_id=item.storage_object_id,
|
||
request=request,
|
||
session=session,
|
||
)
|
||
|
||
assert item.object_status == "deleted"
|
||
assert item.is_deleted == 1
|
||
assert item.deleted_at is not None
|
||
assert result["data"]["object_status"] == "deleted"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_jupyter_check_notebook_lock_ignores_deleted_scripts() -> None:
|
||
"""``is_deleted == 0`` filter must hide deleted notebooks from Jupyter checks."""
|
||
from backend.jupyter import check_notebook_is_locked
|
||
|
||
session = AsyncMock()
|
||
session.execute = AsyncMock()
|
||
session.execute.return_value.one_or_none = MagicMock(return_value=None)
|
||
|
||
result = await check_notebook_is_locked(
|
||
workspace_id="01WS0000000000000000000A",
|
||
notebook_path="test.ipynb",
|
||
user_id="01USR0000000000000000000A",
|
||
session=session,
|
||
)
|
||
|
||
assert result is False
|
||
# Verify the query carries the is_deleted filter.
|
||
call = session.execute.await_args
|
||
statement = call[0][0]
|
||
compiled = str(statement.compile(compile_kwargs={"literal_binds": True}))
|
||
assert "is_deleted" in compiled
|