189 lines
6.7 KiB
Python
189 lines
6.7 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_payload
|
|
from backend.services.storage import _safe_file_name, _safe_path_segment
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_resource_rejects_duplicate_active_name() -> None:
|
|
"""Two binds with the same resource_name but different upload_ids raise 409."""
|
|
from backend.resources import bind_resource
|
|
|
|
class _BindSessionMock:
|
|
def __init__(self, existing_active=None):
|
|
self._existing_active = existing_active
|
|
self._calls = 0
|
|
|
|
async def scalar(self, _stmt):
|
|
self._calls += 1
|
|
# 1st scalar: UploadSessions lookup; 2nd: active name clash check.
|
|
if self._calls == 1:
|
|
return SimpleNamespace(
|
|
upload_id="01UPL0000000000000000000B",
|
|
storage_object_id="01OBJ0000000000000000000B",
|
|
workspace_id="01WS0000000000000000000A",
|
|
user_id="01USR0000000000000000000A",
|
|
)
|
|
return self._existing_active
|
|
|
|
async def get(self, _model, _pk):
|
|
return _make_storage_object("01WS0000000000000000000A/01USR0000000000000000000A/data.csv")
|
|
|
|
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)
|
|
|
|
context = SimpleNamespace(
|
|
request_id="01REQ0000000000000000000A",
|
|
user=SimpleNamespace(user_id="01USR0000000000000000000A"),
|
|
workspace=SimpleNamespace(workspace_id="01WS0000000000000000000A"),
|
|
)
|
|
payload = SimpleNamespace(
|
|
resource_name="data.csv", description=None, visibility="private"
|
|
)
|
|
request = MagicMock()
|
|
|
|
# First bind succeeds: no active resource with this name yet.
|
|
session1 = _BindSessionMock(existing_active=None)
|
|
await bind_resource(
|
|
upload_id="01UPL0000000000000000000A",
|
|
payload=payload,
|
|
request=request,
|
|
context=context,
|
|
session=session1,
|
|
)
|
|
|
|
# Second bind with same resource_name fails because an active row exists.
|
|
session2 = _BindSessionMock(
|
|
existing_active=SimpleNamespace(resource_id="01RES0000000000000000000A")
|
|
)
|
|
with pytest.raises(Exception) as exc_info:
|
|
await bind_resource(
|
|
upload_id="01UPL0000000000000000000B",
|
|
payload=payload,
|
|
request=request,
|
|
context=context,
|
|
session=session2,
|
|
)
|
|
assert exc_info.value.status_code == 409
|
|
assert "already exists" 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_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"
|
|
)
|