model-platform/backend/tests/test_count_scripts.py

"""Unit tests for GET /api/v1/scripts/count endpoint.

Verifies the count endpoint matches the (workspace-wide) listing scope of
``list_scripts(parent_path="")``: workspace + active scripts across every
owner's ``StorageObjects.relative_path``, narrowed by visibility for
non-admin (admin short-circuits). This avoids under/over-reporting on the
dashboard — the count is the size of the set list_scripts would return if
it weren't lazy.
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest

from backend.scripts import count_scripts


def _ctx(
    user_id: str = "U001",
    workspace_id: str = "W001",
    *,
    is_admin: bool = False,
    is_system_admin: bool = False,
) -> SimpleNamespace:
    return SimpleNamespace(
        request_id="test",
        user=SimpleNamespace(user_id=user_id),
        workspace=SimpleNamespace(workspace_id=workspace_id),
        role=SimpleNamespace(role_code="admin" if is_admin else "developer"),
        is_system_admin=is_system_admin,
        # ``count_scripts`` now consults ``context.is_admin`` directly
        # (matching list_scripts / list_resources); SimpleNamespace needs
        # it as a plain attribute.
        is_admin=is_admin or is_system_admin,
    )


def _compile(stmt) -> str:
    from sqlalchemy.dialects import mysql as mysql_dialect

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


async def test_count_scripts_returns_scalar_int() -> None:
    captured = []

    mock_session = MagicMock()
    mock_session.scalar = AsyncMock(
        side_effect=lambda stmt: (captured.append(stmt), 7)[1]
    )

    result = await count_scripts(context=_ctx(), session=mock_session)
    assert result["data"] == {"total": 7}
    assert result["meta"] == {}
    assert result["request_id"] == "test"
    # Exactly one COUNT(*) query issued.
    assert len(captured) == 1
    stmt = captured[0]
    sql = _compile(stmt).lower()
    # JOIN to StorageObjects so orphaned scripts (no joinable row) are
    # excluded — matches list_scripts INNER JOIN behaviour.
    assert "inner join storage_objects" in sql
    # Scope: workspace_id + active status + workspace-wide prefix.
    assert "scripts.workspace_id" in sql
    assert "scripts.status" in sql
    assert "like 'workspace/%%'" in sql
    # Non-admin (default) narrows by visibility.
    assert "scripts.owner_user_id = 'u001'" in sql
    assert "scripts.visibility in ('workspace', 'public')" in sql


async def test_count_scripts_handles_null_result() -> None:
    """MySQL COUNT(*) on empty result returns 0, not NULL — but defensively
    coerce NULL to 0 to keep the response shape consistent."""
    mock_session = MagicMock()
    mock_session.scalar = AsyncMock(return_value=None)
    result = await count_scripts(context=_ctx(), session=mock_session)
    assert result["data"] == {"total": 0}


async def test_count_scripts_workspace_wide_not_user_scoped() -> None:
    """The prefix is workspace-wide (``workspace/%`` — no embedded user_id),
    so different users count the same physical tree; the only per-user
    difference is the non-admin visibility predicate (owner_user_id = me)."""
    captured = []

    mock_session = MagicMock()
    mock_session.scalar = AsyncMock(
        side_effect=lambda stmt: (captured.append(stmt), 3)[1]
    )

    await count_scripts(context=_ctx(user_id="alice"), session=mock_session)
    sql_alice = _compile(captured[-1]).lower()

    await count_scripts(context=_ctx(user_id="bob"), session=mock_session)
    sql_bob = _compile(captured[-1]).lower()

    # Both count the same workspace-wide subtree.
    assert "like 'workspace/%%'" in sql_alice
    assert "like 'workspace/%%'" in sql_bob
    # Neither embeds the user_id in the path prefix.
    assert "workspace/alice/%" not in sql_alice
    assert "workspace/bob/%" not in sql_bob
    # Per-user narrowing happens via the visibility predicate.
    assert "scripts.owner_user_id = 'alice'" in sql_alice
    assert "scripts.owner_user_id = 'bob'" in sql_bob


async def test_count_scripts_admin_skips_visibility_filter() -> None:
    """Admin short-circuits the visibility predicate and counts every
    active script in the workspace (dashboard '全部脚本' / '工作副本')."""
    captured = []

    mock_session = MagicMock()
    mock_session.scalar = AsyncMock(
        side_effect=lambda stmt: (captured.append(stmt), 42)[1]
    )

    result = await count_scripts(
        context=_ctx(user_id="alice", is_admin=True), session=mock_session
    )
    assert result["data"] == {"total": 42}
    sql = _compile(captured[0]).lower()
    assert "like 'workspace/%%'" in sql
    # visibility / owner_user_id still appear in the SELECT projection, but
    # the visibility WHERE predicate must be absent for admins.
    assert "scripts.visibility in ('workspace', 'public')" not in sql


async def test_count_scripts_route_declared_before_script_id_route() -> None:
    """Static check: the `/api/v1/scripts/count` route MUST be declared in
    scripts.py before `/api/v1/scripts/{script_id}/...`, otherwise FastAPI's
    declaration-order matching will interpret `count` as a script_id."""
    from backend.scripts import count_scripts, get_script

    assert callable(count_scripts)
    assert callable(get_script)