"""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] = []

    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:
        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_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