tao.chenandClaude
bca239ed4b
4-phase restructuring of the previously flat backend/ package. Each
phase lands as a single squash commit so future bisects stay readable
per phase if needed.
## Phase 1 — move + shim (location-only, zero behavior change)
* git mv 14 files into api/ schemas/ services/ clients/ subpackages
(history preserved via RM/R renames)
* New files: api/{admin,auth,dependencies,jupyter,platform,resources,
scripts,storage}.py + api/schedules/{schedules,runs}.py
* New files: schemas/{auth,common,jupyter,platform,resources,
schedules,scripts}.py
* New files: clients/{rclone,runtime,scheduler}.py
* Old paths kept as 1-line `from backend.<new> import *` shims so
tests/main.py/importers kept working untouched
* schemas/__init__.py now re-exports from backend.schemas.<domain>
## Phase 2 — APIRouter prefix consolidation
* Every APIRouter() now carries its prefix (e.g. prefix="/api/v1/auth")
and decorators are stripped of the redundant path prefix
* URL paths exposed to the frontend are byte-identical to before
* Affected: api/{auth,jupyter,admin,platform,resources,scripts,
storage}.py + api/schedules/{schedules,runs}.py
## Phase 3 — first service-layer extraction
* backend.services.schedules.validate_dag moved out of api/
(pure DAG validator, no Request/BackgroundTasks/DB)
* api/schedules/schedules.py now re-exports the symbol so existing
4 callsites keep working unchanged
* Added backend/tests/test_validate_dag.py: 8 unit tests covering
DAG_EMPTY, linear chain, diamond, cycle, self-edge, duplicate
edge, orphan edge, multi-root ordering
## Phase 4 — delete shims + unify test imports
* Removed 14 flat shim files + schemas/__init__.py
* Migrated 5 test files (32 import sites) to new paths:
backend.scripts.* → backend.api.scripts.*
backend.resources.* → backend.api.resources.*
backend.jupyter.* → backend.api.jupyter.*
backend.runtime_client.* → backend.clients.runtime.*
backend.schemas.UpdateScriptRequest → backend.schemas.scripts.*
* audit.py kept at backend.audit (main.py references it; not a
shim, real code)
## Final structure
backend/src/backend/
main.py, audit.py, __init__.py
api/ (10 files: routes + 2 subpackage)
schemas/ (7 files: Pydantic contracts)
services/ (storage + schedules)
clients/ (rclone, runtime, scheduler)
## Verification
* uv run python -m compileall backend/src backend/tests — clean
* uv run --package backend pytest backend/tests -q — 122 passed
(114 → 114 → 122 → 122 across phases)
* grep -r 'from backend\.\(scripts\|resources\|...\)' backend/ — 0 hits
* git blame --follow still traces file origins through the renames
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
724 lines
26 KiB
Python
724 lines
26 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.api.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.api.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.api.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.api.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.api.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.api.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.api.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.api.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.api.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.api.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.api.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.scripts import UpdateScriptRequest
|
||
from backend.api.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.api.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.scripts import UpdateScriptRequest
|
||
from backend.api.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.api.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()
|