"""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: from datetime import UTC, datetime 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", created_at=datetime.now(UTC).replace(tzinfo=None), updated_at=datetime.now(UTC).replace(tzinfo=None), ) 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_soft_delete_object_streams_via_get_stream() -> None: """P0-5 / B2: 必须走 ``get_stream()`` 流式迁移,不能调 ``get()`` 把整个对象加载到内存(10G 对象会 OOM)。""" from backend.services.storage import soft_delete_object item = _storage_object_row() # 让 storage_backend 与 settings 一致,确保走"移动到 trash"分支 item.storage_backend = settings.storage_backend item.object_status = "available" item.bucket_name = "workspace" item.object_key = "ws/u/file.bin" item.usage_type = "working_copy" source_store = MagicMock() # 模拟一个 async generator 作为 get_stream 的返回值 async def _fake_stream(_key, _chunk_size=65536): yield b"chunk-1" yield b"chunk-2" source_store.get_stream.side_effect = _fake_stream source_store.put = AsyncMock() source_store.delete = AsyncMock() trash_store = MagicMock() trash_store.put = AsyncMock() trash_store.delete = AsyncMock() request = MagicMock() request.app.state.object_stores = { item.bucket_name: source_store, "trash": trash_store, } session = AsyncMock() session.scalar = AsyncMock(return_value=item) # 捕获原 key —— helper 在移动后会把 item.object_key 重写成 trash_key, # 不捕获的话下面断言会拿不到原值。 original_object_key = item.object_key await soft_delete_object( storage_object_id=item.storage_object_id, request=request, session=session, ) # get_stream 必须被调用;get() 不应被调用 —— 否则仍是全量加载路径 source_store.get_stream.assert_called_once_with(original_object_key) source_store.get.assert_not_called() # 源对象删除 —— 同样用原始 key 断言(item.object_key 已被 helper 改写) source_store.delete.assert_awaited_once_with(original_object_key) # 目标桶写入走 put(接受 async iter),trash_key 尾部拼 storage_object_id trash_store.put.assert_awaited_once() put_args, _ = trash_store.put.call_args assert put_args[0] == f"workspace/{original_object_key}-{item.storage_object_id}" @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 @pytest.mark.asyncio async def test_update_script_writes_back_storage_object_metadata( monkeypatch: pytest.MonkeyPatch, ) -> None: """update_script must persist the new content_hash/size_bytes/relative_path onto the StorageObjects row (not just into the response dict). Pre-fix the DB row kept the create-time values forever, so cache/dedup/hash checks downstream saw stale data. Also locks in the workspace-prefixed relative_path convention so the frontend's slice(2) reducer produces a non-empty path even for root scripts. """ import hashlib from backend.schemas import UpdateScriptRequest from backend.scripts import update_script script = _script_row() storage_object = _storage_object_row() # Seed stale metadata as it would have looked after create_script_record. user_id = "01USR0000000000000000000A" storage_object.object_key = f"{script.workspace_id}/{user_id}/test.ipynb" storage_object.content_hash = hashlib.sha256(b"old").hexdigest() storage_object.size_bytes = 3 storage_object.relative_path = f"workspace/{user_id}/test.ipynb" request = _make_request() runtime_client = request.app.state.runtime_client runtime_client.upload_file = AsyncMock(return_value={"name": "test.ipynb"}) runtime_client.create_notebook = AsyncMock(return_value={"name": "test.ipynb"}) context = _make_context() session = AsyncMock() session.flush = AsyncMock() session.refresh = 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) new_content = '{"cells": [{"cell_type": "code", "source": ["print(1)"]}]}\n' payload = UpdateScriptRequest(content=new_content) result = await update_script( script_id=script.script_id, payload=payload, request=request, context=context, session=session, ) expected_hash = hashlib.sha256(new_content.encode("utf-8")).hexdigest() expected_size = len(new_content.encode("utf-8")) # DB row carries the new metadata. assert storage_object.content_hash == expected_hash assert storage_object.size_bytes == expected_size assert storage_object.relative_path == f"workspace/{user_id}/test.ipynb" # Response echoes the same values. data = result["data"] assert data["content_hash"] == expected_hash assert data["size_bytes"] == expected_size assert data["relative_path"] == f"workspace/{user_id}/test.ipynb" # session.flush + refresh were called to push the new metadata to DB. session.flush.assert_awaited_once() session.refresh.assert_awaited_once_with(storage_object) # The runtime client received the create_notebook for the Jupyter path # (script_type='notebook' goes through create_notebook, not upload_file). runtime_client.create_notebook.assert_awaited_once() @pytest.mark.asyncio async def test_update_script_jupyter_only_uses_dict_fallback( monkeypatch: pytest.MonkeyPatch, ) -> None: """When a script has no StorageObjects row (jupyter-only fallback path), update_script must still return the workspace-prefixed relative_path so the frontend's slice(2) reducer yields the basename — not an empty string. No DB write should be attempted when there is no row to update. """ import hashlib from backend.schemas import UpdateScriptRequest from backend.scripts import update_script script = _script_row() user_id = "01USR0000000000000000000A" request = _make_request() runtime_client = request.app.state.runtime_client runtime_client.upload_file = AsyncMock(return_value={"name": "test.ipynb"}) runtime_client.create_notebook = AsyncMock(return_value={"name": "test.ipynb"}) context = _make_context() session = AsyncMock() session.flush = AsyncMock() session.refresh = 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, None]: return script, None monkeypatch.setattr("backend.scripts.get_script_row", _fake_get_script_row) payload = UpdateScriptRequest(content='{"cells": []}\n') result = await update_script( script_id=script.script_id, payload=payload, request=request, context=context, session=session, ) expected_hash = hashlib.sha256(b'{"cells": []}\n').hexdigest() data = result["data"] # Jupyter-only fallback derives jupyter_path from the script_id via # _jupyter_path(), which appends ".ipynb". The contract we lock in is # "always workspace-prefixed, never bare user_id/basename" so the # frontend's slice(2) reducer is safe. assert data["relative_path"] == f"workspace/{script.script_id}.ipynb" assert data["relative_path"].startswith("workspace/") assert data["content_hash"] == expected_hash assert data["size_bytes"] == len(b'{"cells": []}\n') # No DB write should be attempted when there is no StorageObjects row. session.flush.assert_not_awaited() session.refresh.assert_not_awaited()