model-platform/backend/tests/test_resources.py

"""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 AsyncMock, MagicMock

import pytest
from fastapi import HTTPException
from sqlalchemy import Column, MetaData, String, Table, create_engine, select
from sqlalchemy.dialects import mysql as mysql_dialect
from backend.api.resources import (
    _build_list_resources_descendant_prefix,
    can_view,
    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.api.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.api.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.api.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.api.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.api.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.api.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")
    # Default None when no Users join row is provided (bind_resource path).
    assert payload["owner_display_name"] is None


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")
    # Explicit owner_display_name is passed through to the payload.
    payload = resource_payload(
        _make_resource(ws, user),
        _make_storage_object(f"{ws}/{user}/data.csv"),
        owner_display_name="张三",
    )
    assert payload["owner_display_name"] == "张三"


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")
    assert payload["owner_display_name"] is None


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"
            )


# ─── parent_path filtering (mirrors test_list_scripts_parent_path.py) ────────


def _resource_ctx(workspace_id: str = "W001") -> SimpleNamespace:
    return SimpleNamespace(
        request_id="test",
        user=SimpleNamespace(user_id="U001"),
        workspace=SimpleNamespace(workspace_id=workspace_id),
        is_admin=False,
    )


class TestCanViewVisibility:
    """同一 workspace 内：owner 永远可见自己的资源（含 private）；
    其他成员只见 visibility in {workspace, public} 的资源；
    admin 全部可见。与 list_resources 的 SQL 谓词一致。"""

    @staticmethod
    def _viewer() -> SimpleNamespace:
        return SimpleNamespace(
            request_id="test",
            user=SimpleNamespace(user_id="U002"),  # not the owner
            workspace=SimpleNamespace(workspace_id="W001"),
            is_admin=False,
        )

    @staticmethod
    def _owner() -> SimpleNamespace:
        ctx = TestCanViewVisibility._viewer()
        ctx.user = SimpleNamespace(user_id="U001")  # the owner
        return ctx

    @staticmethod
    def _admin() -> SimpleNamespace:
        ctx = TestCanViewVisibility._viewer()
        ctx.is_admin = True
        return ctx

    @staticmethod
    def _resource(visibility: str) -> SimpleNamespace:
        res = _make_resource("W001", "U001")
        res.visibility = visibility
        return res

    def test_owner_can_view_own_private_resource(self) -> None:
        assert (
            can_view(
                self._resource("private"),
                self._owner(),
            )
            is True
        )

    def test_workspace_member_cannot_view_others_private_resource(self) -> None:
        assert (
            can_view(
                self._resource("private"),
                self._viewer(),
            )
            is False
        )

    def test_workspace_member_can_view_others_workspace_resource(self) -> None:
        assert (
            can_view(
                self._resource("workspace"),
                self._viewer(),
            )
            is True
        )

    def test_workspace_member_can_view_others_public_resource(self) -> None:
        assert (
            can_view(
                self._resource("public"),
                self._viewer(),
            )
            is True
        )

    def test_admin_can_view_others_private_resource(self) -> None:
        assert (
            can_view(
                self._resource("private"),
                self._admin(),
            )
            is True
        )


class TestBuildListResourcesDescendantPrefix:
    """Pure-function contract for the escaped object_key prefix. Unlike
    the scripts helper there is NO user scope — data resources are
    workspace-wide, so the prefix is just the normalized+escaped path."""

    def test_empty_parent_path_returns_empty_prefix(self) -> None:
        assert _build_list_resources_descendant_prefix("") == ""

    def test_subdir_prefix_appends_trailing_slash(self) -> None:
        assert _build_list_resources_descendant_prefix("foo/bar") == "foo/bar/"

    def test_escapes_underscore(self) -> None:
        assert _build_list_resources_descendant_prefix("foo_bar") == r"foo\_bar/"

    def test_escapes_percent(self) -> None:
        assert _build_list_resources_descendant_prefix("100%match") == r"100\%match/"

    def test_normalizes_leading_trailing_slashes(self) -> None:
        assert _build_list_resources_descendant_prefix("/foo/bar/") == "foo/bar/"

    def test_rejects_traversal(self) -> None:
        with pytest.raises(HTTPException) as exc:
            _build_list_resources_descendant_prefix("foo/../bar")
        assert exc.value.status_code == 422


# ─── layer 2: SQL contract (mock session + mysql dialect compile) ────────────


def _compile_sql(stmt) -> str:
    return str(
        stmt.compile(
            dialect=mysql_dialect.dialect(),
            compile_kwargs={"literal_binds": True},
        )
    )


class _ListResourcesMockResult:
    def all(self):
        return []


def _list_resources_capturing_session(captured_sql: list[str]) -> MagicMock:
    mock_session = MagicMock()
    mock_session.execute = AsyncMock(
        side_effect=lambda stmt: (
            captured_sql.append(_compile_sql(stmt)) or _ListResourcesMockResult()
        )
    )
    return mock_session


async def test_list_resources_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
    from backend.api.resources import list_resources

    captured_sql: list[str] = []
    mock_session = _list_resources_capturing_session(captured_sql)

    await list_resources(
        parent_path="foo/bar",
        owner_user_id=None,
        context=_resource_ctx(),
        session=mock_session,
        visibility=None,
        keyword=None,
    )

    assert len(captured_sql) == 1
    sql = captured_sql[0].lower()
    # Default (no owner_user_id) scopes to the requester's own object_key
    # subtree: LIKE w001/u001/foo/bar/% (direct children), excluding deeper.
    assert "like 'w001/u001/foo/bar/%%'" in sql
    assert "not like 'w001/u001/foo/bar/%%/%%'" in sql


async def test_list_resources_where_clause_escapes_underscore() -> None:
    """Regression: parent_path containing ``_`` MUST be escaped in the
    compiled LIKE pattern, otherwise sibling-path leak (``fooXbar``) returns."""
    from backend.api.resources import list_resources

    captured_sql: list[str] = []
    mock_session = _list_resources_capturing_session(captured_sql)

    await list_resources(
        parent_path="foo_bar",
        owner_user_id=None,
        context=_resource_ctx(),
        session=mock_session,
        visibility=None,
        keyword=None,
    )
    sql = captured_sql[0]
    sql_lower = sql.lower()
    # SQLAlchemy doubles the escape char inside the SQL string literal, so
    # the helper's ``foo\_bar`` renders as ``foo\\_bar`` in the SQL text.
    assert r"like 'w001/u001/foo\\_bar/%%'" in sql_lower
    assert r"not like 'w001/u001/foo\\_bar/%%/%%'" in sql_lower
    # Both LIKE clauses declare ESCAPE '\\' (two in total).
    assert sql.count("ESCAPE '\\\\'") == 2, sql


async def test_list_resources_empty_parent_path_adds_root_like_clause() -> None:
    """Empty parent_path still applies the directory filter (symmetric with
    list_scripts): ``{ws_id}/{owner}/%`` AND NOT ``{ws_id}/{owner}/%/%`` so
    the owner-scoped root view returns only direct children of the
    requester's root, never nested descendants. Skipping the filter for
    empty input used to surface nested resources at the root and visually
    broke the directory tree.
    """
    from backend.api.resources import list_resources

    captured_sql: list[str] = []
    mock_session = _list_resources_capturing_session(captured_sql)

    await list_resources(
        parent_path="",
        owner_user_id=None,
        context=_resource_ctx(),
        session=mock_session,
        visibility=None,
        keyword=None,
    )
    sql = captured_sql[0].lower()
    assert " like 'w001/u001/%%'" in sql
    assert " not like 'w001/u001/%%/%%'" in sql


async def test_list_resources_owner_param_scopes_to_other_owner() -> None:
    """owner_user_id=<other> scopes object_key LIKE to that owner's subtree
    so the tree can lazily fetch another member's data resources on group
    expand. Non-admin visibility predicate is still applied, so the other
    owner's private resources are excluded (workspace/public only).
    """
    from backend.api.resources import list_resources

    captured_sql: list[str] = []
    mock_session = _list_resources_capturing_session(captured_sql)

    await list_resources(
        parent_path="",
        owner_user_id="U002",
        context=_resource_ctx(),
        session=mock_session,
        visibility=None,
        keyword=None,
    )
    sql = captured_sql[0].lower()
    assert " like 'w001/u002/%%'" in sql
    assert " not like 'w001/u002/%%/%%'" in sql
    # non-admin: visibility predicate still present (excludes U002 private)
    assert "data_resources.owner_user_id = 'u001'" in sql
    assert "data_resources.visibility in ('workspace', 'public')" in sql


async def test_list_resources_joins_users_for_display_name() -> None:
    """list_resources must OUTER JOIN users and SELECT users.display_name so
    every resource carries owner_display_name (frontend displayName chain)."""
    from backend.api.resources import list_resources

    captured_sql: list[str] = []
    mock_session = _list_resources_capturing_session(captured_sql)

    await list_resources(
        parent_path="",
        owner_user_id=None,
        context=_resource_ctx(),
        session=mock_session,
        visibility=None,
        keyword=None,
    )
    sql = captured_sql[0]
    sql_lower = sql.lower()
    assert "outer join users" in sql_lower
    assert "users.display_name" in sql_lower


# ─── layer 3: behavioral test on real LIKE execution (SQLite) ────────────────


@pytest.fixture
def sqlite_object_key_table():
    """SQLite in-memory stand-in for ``storage_objects.object_key`` rows.

    Three-layer structure: direct children under ``data`` owned by two
    different users (cross-owner), a deeper descendant, a sibling folder,
    a root file, plus ``data_x`` and its ``dataXx`` / ``data2x`` decoys.
    """
    engine = create_engine("sqlite:///:memory:")
    metadata = MetaData()
    table = Table(
        "object_keys",
        metadata,
        Column("object_key", String(1024), nullable=False),
    )
    metadata.create_all(engine)
    rows = [
        "W001/U001/data/alpha.csv",          # direct child (owner U001)
        "W001/U002/data/beta.csv",           # direct child (owner U002)
        "W001/U001/data/deep/nested.csv",    # deeper descendant
        "W001/U001/database/gamma.csv",      # sibling folder
        "W001/U001/root.csv",                # root file
        "W001/U001/data_x/delta.csv",        # target for parent_path=data_x
        "W001/U001/dataXx/decoy.csv",        # sibling decoy (unescaped match)
        "W001/U001/data2x/decoy2.csv",       # sibling decoy (unescaped match)
    ]
    with engine.begin() as conn:
        conn.execute(table.insert(), [{"object_key": key} for key in rows])
    yield engine, table
    engine.dispose()


def test_sqlite_direct_children_across_owners(sqlite_object_key_table) -> None:
    """parent_path='data' returns exactly the direct children under data,
    across every owner, excluding deeper/sibling/root paths."""
    engine, table = sqlite_object_key_table
    prefix = _build_list_resources_descendant_prefix("data")
    like = f"W001/%/{prefix}%"
    not_like = f"W001/%/{prefix}%/%"
    with engine.connect() as conn:
        rows = conn.execute(
            select(table.c.object_key).where(
                table.c.object_key.like(like, escape="\\"),
                ~table.c.object_key.like(not_like, escape="\\"),
            )
        ).fetchall()
    matched = sorted(r[0] for r in rows)
    assert matched == [
        "W001/U001/data/alpha.csv",
        "W001/U002/data/beta.csv",
    ], matched


def test_sqlite_escaped_underscore_does_not_match_sibling(
    sqlite_object_key_table,
) -> None:
    """parent_path='data_x' must match only the literal data_x folder, not
    the dataXx / data2x siblings an unescaped pattern would match."""
    engine, table = sqlite_object_key_table
    prefix = _build_list_resources_descendant_prefix("data_x")
    like = f"W001/%/{prefix}%"
    not_like = f"W001/%/{prefix}%/%"
    with engine.connect() as conn:
        rows = conn.execute(
            select(table.c.object_key).where(
                table.c.object_key.like(like, escape="\\"),
                ~table.c.object_key.like(not_like, escape="\\"),
            )
        ).fetchall()
    matched = sorted(r[0] for r in rows)
    assert matched == ["W001/U001/data_x/delta.csv"], matched


def test_sqlite_parent_path_with_slash_direct_children(sqlite_object_key_table) -> None:
    """parent_path='data/deep' returns only data/deep/* — never data/*."""
    engine, table = sqlite_object_key_table
    prefix = _build_list_resources_descendant_prefix("data/deep")
    like = f"W001/%/{prefix}%"
    not_like = f"W001/%/{prefix}%/%"
    with engine.connect() as conn:
        rows = conn.execute(
            select(table.c.object_key).where(
                table.c.object_key.like(like, escape="\\"),
                ~table.c.object_key.like(not_like, escape="\\"),
            )
        ).fetchall()
    matched = sorted(r[0] for r in rows)
    assert matched == ["W001/U001/data/deep/nested.csv"], matched


def test_sqlite_empty_parent_path_returns_root_level_across_owners(
    sqlite_object_key_table,
) -> None:
    """Empty parent_path now applies the root filter (symmetric with
    list_scripts): ``{ws_id}/%/%`` AND NOT ``{ws_id}/%/%/%`` returns only
    direct children of every owner's root, excluding nested descendants.
    Earlier 'no LIKE' behaviour used to surface every row in the
    workspace at the root, which is exactly what made scripts and data
    appear mutually visible and broke the tree.
    """
    engine, table = sqlite_object_key_table
    like = "W001/%/%"
    not_like = "W001/%/%/%"
    with engine.connect() as conn:
        rows = conn.execute(
            select(table.c.object_key).where(
                table.c.object_key.like(like, escape="\\"),
                ~table.c.object_key.like(not_like, escape="\\"),
            )
        ).fetchall()
    matched = sorted(r[0] for r in rows)
    # ``W001/U001/root.csv`` is the only 3-segment path (= direct child
    # of the owner root); all 4+ segment paths (data/*, database/*) are
    # excluded by the NOT LIKE clause. The other 4+ segment files would
    # be returned when the user expands the corresponding subdirectory.
    assert matched == ["W001/U001/root.csv"], matched