invoke-ai--invokeai/tests/app/routers/test_multiuser_authorization.py

"""Tests for API-level authorization on board-image mutations, image mutations,
workflow thumbnail access, and admin email leak prevention.

These tests verify the security fixes for:
1. Shared-board write protection bypass via direct API calls
2. Image mutation endpoints lacking ownership checks
3. Private workflow thumbnail exposure
4. Admin email leak on unauthenticated status endpoint
"""

import logging
from typing import Any
from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import status
from fastapi.testclient import TestClient

from invokeai.app.api.dependencies import ApiDependencies
from invokeai.app.api_app import app
from invokeai.app.services.config.config_default import InvokeAIAppConfig
from invokeai.app.services.invocation_services import InvocationServices
from invokeai.app.services.invoker import Invoker
from invokeai.app.services.session_queue.session_queue_common import SessionQueueItem
from invokeai.app.services.users.users_common import UserCreateRequest
from invokeai.app.services.workflow_records.workflow_records_sqlite import SqliteWorkflowRecordsStorage
from invokeai.backend.util.logging import InvokeAILogger
from tests.fixtures.sqlite_database import create_mock_sqlite_database


class MockApiDependencies(ApiDependencies):
    invoker: Invoker

    def __init__(self, invoker: Invoker) -> None:
        self.invoker = invoker


WORKFLOW_BODY = {
    "name": "Test Workflow",
    "author": "",
    "description": "",
    "version": "1.0.0",
    "contact": "",
    "tags": "",
    "notes": "",
    "nodes": [],
    "edges": [],
    "exposedFields": [],
    "meta": {"version": "3.0.0", "category": "user"},
    "id": None,
    "form_fields": [],
}


@pytest.fixture
def setup_jwt_secret():
    from invokeai.app.services.auth.token_service import set_jwt_secret

    set_jwt_secret("test-secret-key-for-unit-tests-only-do-not-use-in-production")


@pytest.fixture
def client():
    return TestClient(app)


@pytest.fixture
def mock_services() -> InvocationServices:
    from invokeai.app.services.board_image_records.board_image_records_sqlite import SqliteBoardImageRecordStorage
    from invokeai.app.services.board_records.board_records_sqlite import SqliteBoardRecordStorage
    from invokeai.app.services.boards.boards_default import BoardService
    from invokeai.app.services.bulk_download.bulk_download_default import BulkDownloadService
    from invokeai.app.services.client_state_persistence.client_state_persistence_sqlite import (
        ClientStatePersistenceSqlite,
    )
    from invokeai.app.services.image_records.image_records_sqlite import SqliteImageRecordStorage
    from invokeai.app.services.images.images_default import ImageService
    from invokeai.app.services.invocation_cache.invocation_cache_memory import MemoryInvocationCache
    from invokeai.app.services.invocation_stats.invocation_stats_default import InvocationStatsService
    from invokeai.app.services.users.users_default import UserService
    from tests.test_nodes import TestEventService

    configuration = InvokeAIAppConfig(use_memory_db=True, node_cache_size=0)
    logger = InvokeAILogger.get_logger()
    db = create_mock_sqlite_database(configuration, logger)

    return InvocationServices(
        board_image_records=SqliteBoardImageRecordStorage(db=db),
        board_images=None,  # type: ignore
        board_records=SqliteBoardRecordStorage(db=db),
        boards=BoardService(),
        bulk_download=BulkDownloadService(),
        configuration=configuration,
        events=TestEventService(),
        image_files=None,  # type: ignore
        image_records=SqliteImageRecordStorage(db=db),
        images=ImageService(),
        invocation_cache=MemoryInvocationCache(max_cache_size=0),
        logger=logging,  # type: ignore
        model_images=None,  # type: ignore
        model_manager=None,  # type: ignore
        download_queue=None,  # type: ignore
        names=None,  # type: ignore
        performance_statistics=InvocationStatsService(),
        session_processor=None,  # type: ignore
        session_queue=None,  # type: ignore
        urls=None,  # type: ignore
        workflow_records=SqliteWorkflowRecordsStorage(db=db),
        tensors=None,  # type: ignore
        conditioning=None,  # type: ignore
        style_preset_records=None,  # type: ignore
        style_preset_image_files=None,  # type: ignore
        workflow_thumbnails=None,  # type: ignore
        model_relationship_records=None,  # type: ignore
        model_relationships=None,  # type: ignore
        client_state_persistence=ClientStatePersistenceSqlite(db=db),
        users=UserService(db),
        external_generation=None,  # type: ignore
    )


@pytest.fixture()
def mock_invoker(mock_services: InvocationServices) -> Invoker:
    return Invoker(services=mock_services)


def _save_image(mock_invoker: Invoker, image_name: str, user_id: str) -> None:
    """Helper to insert an image record owned by a specific user."""
    from invokeai.app.services.image_records.image_records_common import ImageCategory, ResourceOrigin

    mock_invoker.services.image_records.save(
        image_name=image_name,
        image_origin=ResourceOrigin.INTERNAL,
        image_category=ImageCategory.GENERAL,
        width=100,
        height=100,
        has_workflow=False,
        user_id=user_id,
    )


def _create_user(mock_invoker: Invoker, email: str, display_name: str, is_admin: bool = False) -> str:
    user = mock_invoker.services.users.create(
        UserCreateRequest(email=email, display_name=display_name, password="TestPass123", is_admin=is_admin)
    )
    return user.user_id


def _login(client: TestClient, email: str) -> str:
    r = client.post("/api/v1/auth/login", json={"email": email, "password": "TestPass123", "remember_me": False})
    assert r.status_code == 200
    return r.json()["token"]


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def enable_multiuser(monkeypatch: Any, mock_invoker: Invoker):
    mock_invoker.services.configuration.multiuser = True

    mock_board_images = MagicMock()
    mock_board_images.get_all_board_image_names_for_board.return_value = []
    mock_invoker.services.board_images = mock_board_images

    mock_workflow_thumbnails = MagicMock()
    mock_workflow_thumbnails.get_url.return_value = None
    mock_invoker.services.workflow_thumbnails = mock_workflow_thumbnails

    mock_deps = MockApiDependencies(mock_invoker)
    monkeypatch.setattr("invokeai.app.api.routers.auth.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.auth_dependencies.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.boards.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.board_images.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.images.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers._access.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.workflows.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.session_queue.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.recall_parameters.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.model_manager.ApiDependencies", mock_deps)
    monkeypatch.setattr("invokeai.app.api.routers.custom_nodes.ApiDependencies", mock_deps)
    yield


@pytest.fixture
def admin_token(setup_jwt_secret: None, enable_multiuser: Any, mock_invoker: Invoker, client: TestClient):
    _create_user(mock_invoker, "admin@test.com", "Admin", is_admin=True)
    return _login(client, "admin@test.com")


@pytest.fixture
def user1_token(enable_multiuser: Any, mock_invoker: Invoker, client: TestClient, admin_token: str):
    _create_user(mock_invoker, "user1@test.com", "User One")
    return _login(client, "user1@test.com")


@pytest.fixture
def user2_token(enable_multiuser: Any, mock_invoker: Invoker, client: TestClient, admin_token: str):
    _create_user(mock_invoker, "user2@test.com", "User Two")
    return _login(client, "user2@test.com")


def _create_board(client: TestClient, token: str, name: str = "Test Board") -> str:
    r = client.post(f"/api/v1/boards/?board_name={name.replace(' ', '+')}", headers=_auth(token))
    assert r.status_code == status.HTTP_201_CREATED
    return r.json()["board_id"]


def _share_board(client: TestClient, token: str, board_id: str) -> None:
    r = client.patch(f"/api/v1/boards/{board_id}", json={"board_visibility": "shared"}, headers=_auth(token))
    assert r.status_code == status.HTTP_201_CREATED


def _set_board_visibility(client: TestClient, token: str, board_id: str, visibility: str) -> None:
    r = client.patch(f"/api/v1/boards/{board_id}", json={"board_visibility": visibility}, headers=_auth(token))
    assert r.status_code == status.HTTP_201_CREATED


def _create_workflow(client: TestClient, token: str) -> str:
    r = client.post("/api/v1/workflows/", json={"workflow": WORKFLOW_BODY}, headers=_auth(token))
    assert r.status_code == 200
    return r.json()["workflow_id"]


def _insert_pending_queue_item(session_queue: Any, user_id: str, queue_id: str = "default") -> int:
    """Insert a pending queue item owned by ``user_id`` directly into the queue's database."""
    import uuid

    from invokeai.app.services.shared.graph import Graph, GraphExecutionState
    from tests.test_nodes import PromptTestInvocation

    graph = Graph()
    graph.add_node(PromptTestInvocation(id="prompt", prompt="test"))
    session = GraphExecutionState(graph=graph)
    session_json = session.model_dump_json(warnings=False, exclude_none=True)
    with session_queue._db.transaction() as cursor:
        cursor.execute(
            """--sql
            INSERT INTO session_queue (
                queue_id, session, session_id, batch_id, field_values, priority,
                workflow, origin, destination, retried_from_item_id, user_id
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (queue_id, session_json, session.id, str(uuid.uuid4()), None, 0, None, None, None, None, user_id),
        )
        return cursor.lastrowid


# ===========================================================================
# 1. Board-image mutation authorization
# ===========================================================================


class TestBoardImageMutationAuth:
    """Tests that board_images mutation endpoints enforce ownership."""

    def test_add_image_to_board_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/board_images/", json={"board_id": "x", "image_name": "y"})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_add_image_to_board_batch_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/board_images/batch", json={"board_id": "x", "image_names": ["y"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_remove_image_from_board_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.request("DELETE", "/api/v1/board_images/", json={"image_name": "y"})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_remove_images_from_board_batch_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/board_images/batch/delete", json={"image_names": ["y"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_owner_cannot_add_image_to_shared_board(self, client: TestClient, user1_token: str, user2_token: str):
        board_id = _create_board(client, user1_token, "User1 Shared Board")
        _share_board(client, user1_token, board_id)

        r = client.post(
            "/api/v1/board_images/",
            json={"board_id": board_id, "image_name": "some-image"},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_add_images_batch_to_shared_board(
        self, client: TestClient, user1_token: str, user2_token: str
    ):
        board_id = _create_board(client, user1_token, "User1 Shared Board Batch")
        _share_board(client, user1_token, board_id)

        r = client.post(
            "/api/v1/board_images/batch",
            json={"board_id": board_id, "image_names": ["img1", "img2"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_admin_can_add_image_to_any_board(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-admin-board-img", user1.user_id)
        board_id = _create_board(client, user1_token, "User1 Board For Admin")

        # Admin can add any image to any board — should not be 403
        r = client.post(
            "/api/v1/board_images/",
            json={"board_id": board_id, "image_name": "user1-admin-board-img"},
            headers=_auth(admin_token),
        )
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_non_owner_can_add_own_image_to_public_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """Public boards are documented as writable by other authenticated users."""
        public_board_id = _create_board(client, user1_token, "User1 Public Board")
        _set_board_visibility(client, user1_token, public_board_id, "public")

        user2 = mock_invoker.services.users.get_by_email("user2@test.com")
        assert user2 is not None
        _save_image(mock_invoker, "user2-public-board-img", user2.user_id)

        r = client.post(
            "/api/v1/board_images/",
            json={"board_id": public_board_id, "image_name": "user2-public-board-img"},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_201_CREATED

    def test_owner_can_add_image_to_own_board(self, client: TestClient, mock_invoker: Invoker, user1_token: str):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-own-board-img", user1.user_id)
        board_id = _create_board(client, user1_token, "User1 Own Board")

        r = client.post(
            "/api/v1/board_images/",
            json={"board_id": board_id, "image_name": "user1-own-board-img"},
            headers=_auth(user1_token),
        )
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_add_other_users_image_to_own_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """Attacker creates their own board, then tries to add victim's image to it.
        This must be rejected — otherwise the attacker gains mutation rights via
        the board-ownership fallback in _assert_image_owner."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "victim-image", user1.user_id)

        attacker_board = _create_board(client, user2_token, "Attacker Board")

        r = client.post(
            "/api/v1/board_images/",
            json={"board_id": attacker_board, "image_name": "victim-image"},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_batch_add_other_users_images_to_own_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """Same attack via the batch endpoint."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "victim-batch-img", user1.user_id)

        attacker_board = _create_board(client, user2_token, "Attacker Batch Board")

        r = client.post(
            "/api/v1/board_images/batch",
            json={"board_id": attacker_board, "image_names": ["victim-batch-img"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN


# ===========================================================================
# 2a. Image read-access authorization
# ===========================================================================


class TestImageReadAuth:
    """Tests that image GET endpoints enforce visibility."""

    def test_get_image_dto_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/images/i/some-image")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_image_metadata_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/images/i/some-image/metadata")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_image_full_is_unauthenticated(self, enable_multiuser: Any, client: TestClient):
        # Binary image endpoints are intentionally unauthenticated because
        # browsers load them via <img src> which cannot send Bearer tokens.
        r = client.get("/api/v1/images/i/some-image/full")
        assert r.status_code != status.HTTP_401_UNAUTHORIZED

    def test_get_image_thumbnail_is_unauthenticated(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/images/i/some-image/thumbnail")
        assert r.status_code != status.HTTP_401_UNAUTHORIZED

    def test_get_image_urls_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/images/i/some-image/urls")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_owner_cannot_read_private_image(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 should not be able to read user1's image that is not on a shared board."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-private-img", user1.user_id)

        r = client.get("/api/v1/images/i/user1-private-img", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_owner_can_read_own_image(self, client: TestClient, mock_invoker: Invoker, user1_token: str):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-readable", user1.user_id)

        r = client.get("/api/v1/images/i/user1-readable", headers=_auth(user1_token))
        # Should not be 403 (may be 404/500 due to missing board_image_records mock)
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_admin_can_read_any_image(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-admin-read", user1.user_id)

        r = client.get("/api/v1/images/i/user1-admin-read", headers=_auth(admin_token))
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_shared_board_image_readable_by_other_user(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """An image on a shared board should be readable by any authenticated user."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "shared-board-img", user1.user_id)

        # Create a shared board and add the image to it
        board_id = _create_board(client, user1_token, "Shared Read Board")
        _share_board(client, user1_token, board_id)
        mock_invoker.services.board_image_records.add_image_to_board(board_id=board_id, image_name="shared-board-img")

        r = client.get("/api/v1/images/i/shared-board-img", headers=_auth(user2_token))
        # Should not be 403 — image is on a shared board
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_read_image_metadata(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-meta-blocked", user1.user_id)

        r = client.get("/api/v1/images/i/user1-meta-blocked/metadata", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_list_images_private_board_rejected_for_non_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to enumerate images on user1's private board
        via GET /api/v1/images?board_id=..."""
        board_id = _create_board(client, user1_token, "Private Enum Board")

        r = client.get(f"/api/v1/images/?board_id={board_id}", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_list_images_shared_board_allowed_for_non_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 should be able to list images on user1's shared board."""
        board_id = _create_board(client, user1_token, "Shared Enum Board")
        _share_board(client, user1_token, board_id)

        r = client.get(f"/api/v1/images/?board_id={board_id}", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_200_OK

    def test_get_image_names_private_board_rejected_for_non_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to enumerate image names on user1's private board
        via GET /api/v1/images/names?board_id=..."""
        board_id = _create_board(client, user1_token, "Private Names Board")

        r = client.get(f"/api/v1/images/names?board_id={board_id}", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_get_image_names_shared_board_allowed_for_non_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 should be able to list image names on user1's shared board."""
        board_id = _create_board(client, user1_token, "Shared Names Board")
        _share_board(client, user1_token, board_id)

        r = client.get(f"/api/v1/images/names?board_id={board_id}", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_200_OK

    def test_list_images_own_private_board_allowed(self, client: TestClient, mock_invoker: Invoker, user1_token: str):
        """Owner should be able to list images on their own private board."""
        board_id = _create_board(client, user1_token, "Own Private Board")

        r = client.get(f"/api/v1/images/?board_id={board_id}", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_200_OK

    def test_admin_can_list_images_on_any_board(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        """Admin should be able to list images on any board."""
        board_id = _create_board(client, user1_token, "Admin Enum Board")

        r = client.get(f"/api/v1/images/?board_id={board_id}", headers=_auth(admin_token))
        assert r.status_code == status.HTTP_200_OK


# ===========================================================================
# 2b. Image mutation authorization
# ===========================================================================


class TestImageUploadAuth:
    """Tests that image upload enforces board ownership."""

    def test_upload_to_other_users_shared_board_forbidden(self, client: TestClient, user1_token: str, user2_token: str):
        """A user should not be able to upload an image into another user's shared board."""
        board_id = _create_board(client, user1_token, "User1 Shared Upload Board")
        _share_board(client, user1_token, board_id)

        # user2 tries to upload into user1's shared board
        import io

        fake_image = io.BytesIO(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
        r = client.post(
            f"/api/v1/images/upload?image_category=general&is_intermediate=false&board_id={board_id}",
            files={"file": ("test.png", fake_image, "image/png")},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_owner_can_upload_to_own_shared_board(self, client: TestClient, user1_token: str):
        board_id = _create_board(client, user1_token, "User1 Own Upload Board")
        _share_board(client, user1_token, board_id)

        import io

        fake_image = io.BytesIO(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
        r = client.post(
            f"/api/v1/images/upload?image_category=general&is_intermediate=false&board_id={board_id}",
            files={"file": ("test.png", fake_image, "image/png")},
            headers=_auth(user1_token),
        )
        # Should not be 403 (may fail for other reasons in test env)
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_non_owner_can_upload_to_public_board(self, client: TestClient, user1_token: str, user2_token: str):
        """Public boards allow any authenticated user to upload images."""
        board_id = _create_board(client, user1_token, "User1 Public Upload Board")
        _set_board_visibility(client, user1_token, board_id, "public")

        import io

        fake_image = io.BytesIO(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
        r = client.post(
            f"/api/v1/images/upload?image_category=general&is_intermediate=false&board_id={board_id}",
            files={"file": ("test.png", fake_image, "image/png")},
            headers=_auth(user2_token),
        )
        # Should not be 403 (may fail downstream for other reasons in test env)
        assert r.status_code != status.HTTP_403_FORBIDDEN


class TestImageMutationAuth:
    """Tests that image mutation endpoints enforce ownership."""

    def test_delete_image_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.delete("/api/v1/images/i/some-image")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_update_image_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.patch("/api/v1/images/i/some-image", json={"starred": True})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_batch_delete_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/images/delete", json={"image_names": ["x"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_star_images_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/images/star", json={"image_names": ["x"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_unstar_images_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/images/unstar", json={"image_names": ["x"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_clear_intermediates_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.delete("/api/v1/images/intermediates")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_delete_uncategorized_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.delete("/api/v1/images/uncategorized")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_owner_cannot_delete_image(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 should not be able to delete user1's image."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-image", user1.user_id)

        r = client.delete("/api/v1/images/i/user1-image", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_owner_can_delete_own_image(self, client: TestClient, mock_invoker: Invoker, user1_token: str):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-delete-me", user1.user_id)

        r = client.delete("/api/v1/images/i/user1-delete-me", headers=_auth(user1_token))
        # Should not be 403 (may be 200 or 500 depending on file system)
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_admin_can_delete_any_image(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-admin-delete", user1.user_id)

        r = client.delete("/api/v1/images/i/user1-admin-delete", headers=_auth(admin_token))
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_board_owner_can_delete_image_on_own_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str
    ):
        """Board owner should be able to delete images on their board even if
        the image's user_id is 'system' (e.g. generated images)."""
        # Create image owned by "system" (simulates queue-generated image)
        _save_image(mock_invoker, "system-img-on-board", "system")

        # Create a board owned by user1 and add the image to it
        board_id = _create_board(client, user1_token, "User1 Board With System Img")
        mock_invoker.services.board_image_records.add_image_to_board(
            board_id=board_id, image_name="system-img-on-board"
        )

        r = client.delete("/api/v1/images/i/system-img-on-board", headers=_auth(user1_token))
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_update_image(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-no-star", user1.user_id)

        r = client.patch(
            "/api/v1/images/i/user1-no-star",
            json={"starred": True},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_star_image(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-star-blocked", user1.user_id)

        r = client.post(
            "/api/v1/images/star",
            json={"image_names": ["user1-star-blocked"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_owner_cannot_batch_delete_image(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-batch-del", user1.user_id)

        r = client.post(
            "/api/v1/images/delete",
            json={"image_names": ["user1-batch-del"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_owner_can_delete_image_from_public_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """Public-board semantics promise delete access to images contained in the board."""
        public_board_id = _create_board(client, user1_token, "User1 Public Delete Board")
        _set_board_visibility(client, user1_token, public_board_id, "public")

        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-public-delete", user1.user_id)
        mock_invoker.services.board_image_records.add_image_to_board(public_board_id, "user1-public-delete")

        r = client.delete(
            "/api/v1/images/i/user1-public-delete",
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_200_OK

    def test_clear_intermediates_non_admin_forbidden(self, client: TestClient, user1_token: str):
        r = client.delete("/api/v1/images/intermediates", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_get_intermediates_count_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/images/intermediates")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_download_images_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/images/download", json={"image_names": ["x"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_owner_cannot_fetch_existing_bulk_download_item(
        self,
        client: TestClient,
        mock_invoker: Invoker,
        monkeypatch: Any,
        tmp_path: Any,
        user1_token: str,
        user2_token: str,
    ):
        """A bulk download zip should be fetchable only by its owner."""
        from fastapi import BackgroundTasks

        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None

        mock_file = tmp_path / "owned-download.zip"
        mock_file.write_text("contents")

        monkeypatch.setattr(mock_invoker.services.bulk_download, "get_path", lambda _: str(mock_file))
        monkeypatch.setattr(mock_invoker.services.bulk_download, "get_owner", lambda _: user1.user_id)
        monkeypatch.setattr(BackgroundTasks, "add_task", lambda *args, **kwargs: None)

        r = client.get("/api/v1/images/download/owned-download.zip", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_images_by_names_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/images/images_by_names", json={"image_names": ["x"]})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_images_by_names_filters_unauthorized(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """images_by_names should silently skip images the caller cannot access."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-by-name", user1.user_id)

        r = client.post(
            "/api/v1/images/images_by_names",
            json={"image_names": ["user1-by-name"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == 200
        # user2 should get an empty list — the image belongs to user1
        assert r.json() == []

    def test_none_board_image_names_only_return_callers_uncategorized_images(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """The uncategorized-images sentinel must not expose other users' image names."""
        mock_invoker.services.board_images.get_all_board_image_names_for_board.side_effect = (
            mock_invoker.services.board_image_records.get_all_board_image_names_for_board
        )

        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        user2 = mock_invoker.services.users.get_by_email("user2@test.com")
        assert user1 is not None
        assert user2 is not None

        _save_image(mock_invoker, "user1-uncategorized-private", user1.user_id)
        _save_image(mock_invoker, "user2-uncategorized-private", user2.user_id)

        r = client.get("/api/v1/boards/none/image_names", headers=_auth(user2_token))
        assert r.status_code == status.HTTP_200_OK
        assert "user2-uncategorized-private" in r.json()
        assert "user1-uncategorized-private" not in r.json()


# ===========================================================================
# 3. Workflow mutation authorization (additional)
# ===========================================================================


class TestWorkflowListScoping:
    """Tests that listing workflows in multiuser mode does not filter out default workflows."""

    def test_default_workflows_visible_when_listing_user_and_default(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str
    ):
        """When categories=['user','default'], default workflows must still appear even
        though user_id_filter is set to the current user (default workflows belong to 'system')."""
        from invokeai.app.services.workflow_records.workflow_records_common import (
            Workflow,
            WorkflowCategory,
            WorkflowMeta,
            WorkflowWithoutID,
        )
        from invokeai.app.util.misc import uuid_string

        default_wf = WorkflowWithoutID(
            name="Test Default Workflow",
            description="A built-in workflow",
            meta=WorkflowMeta(version="3.0.0", category=WorkflowCategory.Default),
            nodes=[],
            edges=[],
            tags="",
            author="",
            contact="",
            version="1.0.0",
            notes="",
            exposedFields=[],
            form_fields=[],
        )
        wf_with_id = Workflow(**default_wf.model_dump(), id=uuid_string())
        # Insert directly via DB since the create API rejects default workflows
        with mock_invoker.services.workflow_records._db.transaction() as cursor:
            cursor.execute(
                "INSERT INTO workflow_library (workflow_id, workflow, user_id) VALUES (?, ?, ?)",
                (wf_with_id.id, wf_with_id.model_dump_json(), "system"),
            )

        # Also create a user workflow via the API
        _create_workflow(client, user1_token)

        # List with categories=user&categories=default
        r = client.get(
            "/api/v1/workflows/?categories=user&categories=default",
            headers=_auth(user1_token),
        )
        assert r.status_code == 200
        data = r.json()
        categories_found = {item["category"] for item in data["items"]}
        assert "default" in categories_found, (
            f"Default workflows were filtered out. Categories found: {categories_found}"
        )
        assert "user" in categories_found

    def test_default_workflows_visible_when_no_category_filter(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str
    ):
        """When no categories filter is given, default workflows should still appear."""
        from invokeai.app.services.workflow_records.workflow_records_common import (
            Workflow,
            WorkflowCategory,
            WorkflowMeta,
            WorkflowWithoutID,
        )
        from invokeai.app.util.misc import uuid_string

        default_wf = WorkflowWithoutID(
            name="Another Default Workflow",
            description="Built-in",
            meta=WorkflowMeta(version="3.0.0", category=WorkflowCategory.Default),
            nodes=[],
            edges=[],
            tags="",
            author="",
            contact="",
            version="1.0.0",
            notes="",
            exposedFields=[],
            form_fields=[],
        )
        wf_with_id = Workflow(**default_wf.model_dump(), id=uuid_string())
        with mock_invoker.services.workflow_records._db.transaction() as cursor:
            cursor.execute(
                "INSERT INTO workflow_library (workflow_id, workflow, user_id) VALUES (?, ?, ?)",
                (wf_with_id.id, wf_with_id.model_dump_json(), "system"),
            )

        _create_workflow(client, user1_token)

        r = client.get("/api/v1/workflows/", headers=_auth(user1_token))
        assert r.status_code == 200
        data = r.json()
        categories_found = {item["category"] for item in data["items"]}
        assert "default" in categories_found, (
            f"Default workflows were filtered out. Categories found: {categories_found}"
        )


class TestWorkflowMutationAuth:
    """Tests for additional workflow mutation endpoints."""

    def test_update_opened_at_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.put("/api/v1/workflows/i/some-id/opened_at")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_owner_cannot_update_opened_at(self, client: TestClient, user1_token: str, user2_token: str):
        workflow_id = _create_workflow(client, user1_token)
        r = client.put(
            f"/api/v1/workflows/i/{workflow_id}/opened_at",
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_owner_can_update_opened_at(self, client: TestClient, user1_token: str):
        workflow_id = _create_workflow(client, user1_token)
        r = client.put(
            f"/api/v1/workflows/i/{workflow_id}/opened_at",
            headers=_auth(user1_token),
        )
        assert r.status_code == 200


# ===========================================================================
# 4. Workflow thumbnail authorization
# ===========================================================================


class TestWorkflowThumbnailAuth:
    """Tests for the workflow thumbnail GET endpoint.

    Workflow and image thumbnail endpoints are intentionally unauthenticated
    because browsers load them via <img src> tags which cannot send Bearer
    tokens. IDs are UUIDs, providing security through unguessability.
    """

    def test_thumbnail_is_unauthenticated(self, enable_multiuser: Any, client: TestClient):
        # Binary image endpoints don't require auth — loaded via <img src>
        r = client.get("/api/v1/workflows/i/some-workflow/thumbnail")
        assert r.status_code != status.HTTP_401_UNAUTHORIZED


# ===========================================================================
# 4. Admin email leak prevention
# ===========================================================================


class TestAdminEmailLeak:
    """Tests that the auth status endpoint does not leak admin email."""

    def test_status_does_not_leak_admin_email_when_setup_complete(self, client: TestClient, admin_token: str):
        """After setup is complete, admin_email must be null."""
        r = client.get("/api/v1/auth/status")
        assert r.status_code == 200
        data = r.json()
        assert data["multiuser_enabled"] is True
        assert data["setup_required"] is False
        assert data["admin_email"] is None

    def test_status_returns_admin_email_during_setup(
        self, setup_jwt_secret: None, enable_multiuser: Any, mock_invoker: Invoker, client: TestClient
    ):
        """Before any admin exists, setup_required=True and admin_email may be returned."""
        # Don't create any users -- setup_required should be True
        r = client.get("/api/v1/auth/status")
        assert r.status_code == 200
        data = r.json()
        assert data["setup_required"] is True
        # admin_email is null here because no admin exists yet, which is correct

    def test_status_no_leak_in_single_user_mode(
        self, setup_jwt_secret: None, monkeypatch: Any, mock_invoker: Invoker, client: TestClient
    ):
        """In single-user mode, admin_email should always be null."""
        mock_invoker.services.configuration.multiuser = False
        mock_deps = MockApiDependencies(mock_invoker)
        monkeypatch.setattr("invokeai.app.api.routers.auth.ApiDependencies", mock_deps)

        r = client.get("/api/v1/auth/status")
        assert r.status_code == 200
        data = r.json()
        assert data["admin_email"] is None
        assert data["multiuser_enabled"] is False


# ===========================================================================
# 6. Session queue authorization
# ===========================================================================


class TestSessionQueueAuth:
    """Tests that session queue endpoints enforce authentication."""

    def test_get_queue_item_ids_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/queue/default/item_ids")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_current_queue_item_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/queue/default/current")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_next_queue_item_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/queue/default/next")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_batch_status_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/queue/default/b/some-batch/status")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_counts_by_destination_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/queue/default/counts_by_destination?destination=canvas")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED


# ===========================================================================
# 6b. Session queue sanitization (cross-user isolation)
# ===========================================================================


class TestSessionQueueSanitization:
    """Tests that sanitize_queue_item_for_user strips all sensitive fields
    from queue items viewed by non-owner, non-admin users."""

    @pytest.fixture
    def _sample_queue_item(self):
        from invokeai.app.services.shared.graph import Graph, GraphExecutionState

        return SessionQueueItem(
            item_id=42,
            status="pending",
            priority=10,
            batch_id="batch-abc",
            origin="workflows",
            destination="canvas",
            session_id="sess-123",
            session=GraphExecutionState(id="sess-123", graph=Graph()),
            error_type="RuntimeError",
            error_message="something broke",
            error_traceback="Traceback ...",
            created_at="2026-01-01T00:00:00",
            updated_at="2026-01-01T01:00:00",
            started_at="2026-01-01T00:30:00",
            completed_at=None,
            queue_id="default",
            user_id="owner-user",
            user_display_name="Owner Display",
            user_email="owner@test.com",
            field_values=None,
            workflow=None,
        )

    def test_owner_sees_all_fields(self, _sample_queue_item: SessionQueueItem):
        from invokeai.app.api.routers.session_queue import sanitize_queue_item_for_user

        result = sanitize_queue_item_for_user(_sample_queue_item, "owner-user", is_admin=False)
        assert result.user_id == "owner-user"
        assert result.user_display_name == "Owner Display"
        assert result.user_email == "owner@test.com"
        assert result.batch_id == "batch-abc"
        assert result.origin == "workflows"
        assert result.destination == "canvas"
        assert result.session_id == "sess-123"
        assert result.priority == 10

    def test_admin_sees_all_fields(self, _sample_queue_item: SessionQueueItem):
        from invokeai.app.api.routers.session_queue import sanitize_queue_item_for_user

        result = sanitize_queue_item_for_user(_sample_queue_item, "admin-user", is_admin=True)
        assert result.user_id == "owner-user"
        assert result.user_display_name == "Owner Display"
        assert result.user_email == "owner@test.com"
        assert result.batch_id == "batch-abc"

    def test_non_owner_sees_only_status_timestamps_errors(self, _sample_queue_item: SessionQueueItem):
        from invokeai.app.api.routers.session_queue import sanitize_queue_item_for_user

        result = sanitize_queue_item_for_user(_sample_queue_item, "other-user", is_admin=False)

        # Preserved: item_id, queue_id, status, timestamps
        assert result.item_id == 42
        assert result.queue_id == "default"
        assert result.status == "pending"
        assert result.created_at == "2026-01-01T00:00:00"
        assert result.updated_at == "2026-01-01T01:00:00"
        assert result.started_at == "2026-01-01T00:30:00"
        assert result.completed_at is None

        # Stripped: errors (may leak file paths, prompts, model names)
        assert result.error_type is None
        assert result.error_message is None
        assert result.error_traceback is None

        # Stripped: user identity
        assert result.user_id == "redacted"
        assert result.user_display_name is None
        assert result.user_email is None

        # Stripped: generation metadata
        assert result.batch_id == "redacted"
        assert result.session_id == "redacted"
        assert result.origin is None
        assert result.destination is None
        assert result.priority == 0
        assert result.field_values is None
        assert result.retried_from_item_id is None
        assert result.workflow is None
        assert result.session.id == "redacted"
        assert len(result.session.graph.nodes) == 0

    def test_sanitization_does_not_mutate_original(self, _sample_queue_item: SessionQueueItem):
        from invokeai.app.api.routers.session_queue import sanitize_queue_item_for_user

        sanitize_queue_item_for_user(_sample_queue_item, "other-user", is_admin=False)
        # Original should be unchanged
        assert _sample_queue_item.user_id == "owner-user"
        assert _sample_queue_item.user_email == "owner@test.com"
        assert _sample_queue_item.batch_id == "batch-abc"


# ===========================================================================
# 7. Recall parameters authorization
# ===========================================================================


class TestRecallParametersAuth:
    """Tests that recall parameter endpoints enforce authentication."""

    def test_get_recall_parameters_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v1/recall/default")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_update_recall_parameters_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v1/recall/default", json={"positive_prompt": "test"})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED


# ===========================================================================
# 7a2. Recall parameters image access control
# ===========================================================================


class TestRecallImageAccess:
    """Tests that recall parameter image references are validated for read access."""

    def test_recall_controlnet_with_other_users_image_rejected(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to reference user1's private image in a control layer."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "victim-ctrl-img", user1.user_id)

        r = client.post(
            "/api/v1/recall/default",
            json={"control_layers": [{"model_name": "some-controlnet", "image_name": "victim-ctrl-img"}]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_recall_ip_adapter_with_other_users_image_rejected(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to reference user1's private image in an IP adapter."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "victim-ip-img", user1.user_id)

        r = client.post(
            "/api/v1/recall/default",
            json={"ip_adapters": [{"model_name": "some-ip-adapter", "image_name": "victim-ip-img"}]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_recall_own_image_allowed(self, client: TestClient, mock_invoker: Invoker, user1_token: str):
        """Owner should be able to reference their own image in recall parameters."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "own-ctrl-img", user1.user_id)

        r = client.post(
            "/api/v1/recall/default",
            json={"control_layers": [{"model_name": "some-controlnet", "image_name": "own-ctrl-img"}]},
            headers=_auth(user1_token),
        )
        # Should not be 403 (may fail downstream for other reasons, e.g. model not found)
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_recall_shared_board_image_allowed(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """An image on a shared board should be usable in recall by any user."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "shared-recall-img", user1.user_id)

        board_id = _create_board(client, user1_token, "Shared Recall Board")
        _share_board(client, user1_token, board_id)
        mock_invoker.services.board_image_records.add_image_to_board(board_id=board_id, image_name="shared-recall-img")

        r = client.post(
            "/api/v1/recall/default",
            json={"ip_adapters": [{"model_name": "some-ip-adapter", "image_name": "shared-recall-img"}]},
            headers=_auth(user2_token),
        )
        assert r.status_code != status.HTTP_403_FORBIDDEN

    def test_recall_admin_can_reference_any_image(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        """Admin should be able to reference any user's image."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "admin-recall-img", user1.user_id)

        r = client.post(
            "/api/v1/recall/default",
            json={"control_layers": [{"model_name": "some-controlnet", "image_name": "admin-recall-img"}]},
            headers=_auth(admin_token),
        )
        assert r.status_code != status.HTTP_403_FORBIDDEN


# ===========================================================================
# 7b. Recall parameters cross-user isolation
# ===========================================================================


class TestRecallParametersIsolation:
    """Tests that recall parameters are scoped per-user, not globally by queue_id."""

    def test_user1_write_does_not_leak_to_user2(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User1 sets a recall parameter; user2 should not see it in client state."""
        # user1 writes a recall parameter
        r = client.post(
            "/api/v1/recall/default",
            json={"positive_prompt": "user1 secret prompt"},
            headers=_auth(user1_token),
        )
        assert r.status_code == 200

        # Verify that user1's data is stored under user1's user_id, not the queue_id
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        user2 = mock_invoker.services.users.get_by_email("user2@test.com")
        assert user1 is not None
        assert user2 is not None

        # user1 should have the value
        val = mock_invoker.services.client_state_persistence.get_by_key(user1.user_id, "recall_positive_prompt")
        assert val is not None
        assert "user1 secret prompt" in val

        # user2 should NOT have the value
        val2 = mock_invoker.services.client_state_persistence.get_by_key(user2.user_id, "recall_positive_prompt")
        assert val2 is None

    def test_two_users_independent_state(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """Both users can write recall params independently without overwriting each other."""
        r1 = client.post(
            "/api/v1/recall/default",
            json={"positive_prompt": "prompt from user1"},
            headers=_auth(user1_token),
        )
        assert r1.status_code == 200

        r2 = client.post(
            "/api/v1/recall/default",
            json={"positive_prompt": "prompt from user2"},
            headers=_auth(user2_token),
        )
        assert r2.status_code == 200

        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        user2 = mock_invoker.services.users.get_by_email("user2@test.com")
        assert user1 is not None
        assert user2 is not None

        val1 = mock_invoker.services.client_state_persistence.get_by_key(user1.user_id, "recall_positive_prompt")
        val2 = mock_invoker.services.client_state_persistence.get_by_key(user2.user_id, "recall_positive_prompt")
        assert val1 is not None and "prompt from user1" in val1
        assert val2 is not None and "prompt from user2" in val2


# ===========================================================================
# 9. Recall parameters event user scoping
# ===========================================================================


class TestRecallParametersEventScoping:
    """Tests that RecallParametersUpdatedEvent carries user_id for targeted delivery."""

    def test_event_includes_user_id(self):
        """RecallParametersUpdatedEvent.build() must set user_id so the socket handler
        can route the event to the correct user room instead of broadcasting."""
        from invokeai.app.services.events.events_common import RecallParametersUpdatedEvent

        event = RecallParametersUpdatedEvent.build(
            queue_id="default",
            user_id="user-abc",
            parameters={"positive_prompt": "test"},
        )
        assert event.queue_id == "default"
        assert event.user_id == "user-abc"
        assert event.parameters == {"positive_prompt": "test"}

    def test_event_not_broadcast_to_all_queue_subscribers(self):
        """RecallParametersUpdatedEvent must have a user_id field so _handle_queue_event
        in sockets.py can route it to the owner room + admin room, not the queue room."""
        from invokeai.app.services.events.events_common import RecallParametersUpdatedEvent

        event = RecallParametersUpdatedEvent.build(
            queue_id="default",
            user_id="owner-123",
            parameters={"seed": 42},
        )
        # The event must carry user_id; without it the socket handler would
        # fall through to the generic else branch and broadcast to all subscribers
        assert hasattr(event, "user_id")
        assert event.user_id == "owner-123"


# ===========================================================================
# 10. Queue status endpoint scoping
# ===========================================================================


class TestQueueStatusScoping:
    """Tests that queue status, batch status, and counts_by_destination
    endpoints scope data to the current user for non-admin callers."""

    def test_get_queue_status_hides_current_item_for_non_owner(self):
        """get_queue_status() must not expose current item details to non-owner, non-admin users."""
        from invokeai.app.services.session_queue.session_queue_common import SessionQueueStatus

        # Simulate a status where the current item belongs to another user
        # When user_id is provided and doesn't match, item details should be None
        status_obj = SessionQueueStatus(
            queue_id="default",
            item_id=None,  # hidden because user doesn't own current item
            session_id=None,
            batch_id=None,
            pending=2,
            in_progress=0,
            waiting=0,
            completed=1,
            failed=0,
            canceled=0,
            total=3,
        )
        # Verify the model accepts None for item details
        assert status_obj.item_id is None
        assert status_obj.session_id is None
        assert status_obj.batch_id is None

    def test_session_queue_status_has_user_fields(self):
        """SessionQueueStatus carries per-user counts (user_pending/user_in_progress) alongside
        the global aggregate counts. The frontend badge needs both to render "own / total" for
        non-admin users in multiuser mode. The per-user fields are optional (None for admins
        and single-user/global callers)."""
        from invokeai.app.services.session_queue.session_queue_common import SessionQueueStatus

        fields = set(SessionQueueStatus.model_fields.keys())
        assert "user_pending" in fields
        assert "user_in_progress" in fields

        # Per-user fields default to None (global/admin caller) and aggregate counts stay global.
        status_obj = SessionQueueStatus(
            queue_id="default",
            item_id=None,
            session_id=None,
            batch_id=None,
            pending=5,
            in_progress=1,
            waiting=0,
            completed=0,
            failed=0,
            canceled=0,
            total=6,
        )
        assert status_obj.user_pending is None
        assert status_obj.user_in_progress is None

        # A non-admin caller's status carries their own subset of the global counts.
        scoped = status_obj.model_copy(update={"user_pending": 2, "user_in_progress": 1})
        assert scoped.pending == 5  # global, unchanged
        assert scoped.user_pending == 2  # this user's share

    def _setup_queue_router(self, mock_invoker: Invoker):
        """Wire a real session queue and a stub processor into the invoker the router uses,
        so GET /queue/{queue_id}/status exercises the real service contract."""
        from unittest.mock import MagicMock

        from invokeai.app.services.session_processor.session_processor_common import SessionProcessorStatus
        from invokeai.app.services.session_queue.session_queue_sqlite import SqliteSessionQueue

        db = mock_invoker.services.board_records._db
        queue = SqliteSessionQueue(db=db)
        queue.start(mock_invoker)
        mock_invoker.services.session_queue = queue

        processor = MagicMock()
        processor.get_status.return_value = SessionProcessorStatus(is_started=True, is_processing=False)
        mock_invoker.services.session_processor = processor
        return queue

    def test_get_queue_status_route_returns_global_and_user_counts(
        self, setup_jwt_secret: None, enable_multiuser: Any, mock_invoker: Invoker, client: TestClient
    ):
        """Regression test: GET /api/v1/queue/{queue_id}/status must return 200 (not 500) and the
        expected global and per-user counts for both non-admin and admin callers. Previously the
        router called get_queue_status() with a keyword the service did not accept, raising a
        TypeError that the broad except turned into a 500 for every status request."""
        queue = self._setup_queue_router(mock_invoker)

        user1_id = _create_user(mock_invoker, "user1@test.com", "User One")
        user2_id = _create_user(mock_invoker, "user2@test.com", "User Two")
        _create_user(mock_invoker, "admin@test.com", "Admin", is_admin=True)
        user1_tok = _login(client, "user1@test.com")
        admin_tok = _login(client, "admin@test.com")

        # Three pending jobs globally: two owned by user1, one by user2; none by the admin.
        _insert_pending_queue_item(queue, user_id=user1_id)
        _insert_pending_queue_item(queue, user_id=user1_id)
        _insert_pending_queue_item(queue, user_id=user2_id)

        # Non-admin caller sees the global total but only their own pending count.
        r = client.get("/api/v1/queue/default/status", headers=_auth(user1_tok))
        assert r.status_code == 200
        queue_status = r.json()["queue"]
        assert queue_status["pending"] == 3
        assert queue_status["total"] == 3
        assert queue_status["user_pending"] == 2
        assert queue_status["user_in_progress"] == 0

        # Admin caller sees the same global total. Admins query with user_id=None, so no
        # per-user counts are computed (the badge falls back to the global total for admins).
        r = client.get("/api/v1/queue/default/status", headers=_auth(admin_tok))
        assert r.status_code == 200
        queue_status = r.json()["queue"]
        assert queue_status["pending"] == 3
        assert queue_status["total"] == 3
        assert queue_status["user_pending"] is None


# ===========================================================================
# 10b. Model install job authorization
# ===========================================================================


class TestModelInstallAuth:
    """Tests that model install job endpoints require admin authentication."""

    def test_list_model_installs_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v2/models/install")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_get_model_install_job_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.get("/api/v2/models/install/1")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_pause_model_install_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v2/models/install/1/pause")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_resume_model_install_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v2/models/install/1/resume")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_restart_failed_model_install_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v2/models/install/1/restart_failed")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_restart_model_install_file_requires_auth(self, enable_multiuser: Any, client: TestClient):
        r = client.post("/api/v2/models/install/1/restart_file", json="https://example.com/model.safetensors")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_non_admin_cannot_list_model_installs(self, enable_multiuser: Any, client: TestClient, user1_token: str):
        r = client.get("/api/v2/models/install", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_non_admin_cannot_pause_model_install(self, enable_multiuser: Any, client: TestClient, user1_token: str):
        r = client.post("/api/v2/models/install/1/pause", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN


# ===========================================================================
# 11. Bulk download access control
# ===========================================================================


class TestBulkDownloadAccessControl:
    """Tests that bulk download endpoints enforce image/board read access and
    that the fetch endpoint verifies ownership of the zip file."""

    @pytest.fixture(autouse=True)
    def _mock_background_tasks(self, monkeypatch: Any):
        """Prevent BackgroundTasks.add_task from actually running the handler,
        which would fail because image_files is None in the test fixture."""
        from fastapi import BackgroundTasks

        monkeypatch.setattr(BackgroundTasks, "add_task", lambda *args, **kwargs: None)

    def test_bulk_download_by_image_names_rejected_for_non_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to bulk-download images owned by user1."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-private-dl", user1.user_id)

        r = client.post(
            "/api/v1/images/download",
            json={"image_names": ["user1-private-dl"]},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_bulk_download_by_image_names_allowed_for_owner(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str
    ):
        """Owner should be able to bulk-download their own images."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-own-dl", user1.user_id)

        r = client.post(
            "/api/v1/images/download",
            json={"image_names": ["user1-own-dl"]},
            headers=_auth(user1_token),
        )
        assert r.status_code == status.HTTP_202_ACCEPTED

    def test_bulk_download_by_board_rejected_for_private_board(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 must not be able to bulk-download from user1's private board."""
        board_id = _create_board(client, user1_token, "Private DL Board")

        r = client.post(
            "/api/v1/images/download",
            json={"board_id": board_id},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_bulk_download_by_shared_board_allowed(
        self, client: TestClient, mock_invoker: Invoker, user1_token: str, user2_token: str
    ):
        """User2 should be able to bulk-download from user1's shared board."""
        board_id = _create_board(client, user1_token, "Shared DL Board")
        _share_board(client, user1_token, board_id)

        r = client.post(
            "/api/v1/images/download",
            json={"board_id": board_id},
            headers=_auth(user2_token),
        )
        assert r.status_code == status.HTTP_202_ACCEPTED

    def test_admin_can_bulk_download_any_images(
        self, client: TestClient, mock_invoker: Invoker, admin_token: str, user1_token: str
    ):
        """Admin should be able to bulk-download any user's images."""
        user1 = mock_invoker.services.users.get_by_email("user1@test.com")
        assert user1 is not None
        _save_image(mock_invoker, "user1-admin-dl", user1.user_id)

        r = client.post(
            "/api/v1/images/download",
            json={"image_names": ["user1-admin-dl"]},
            headers=_auth(admin_token),
        )
        assert r.status_code == status.HTTP_202_ACCEPTED

    def test_bulk_download_events_carry_user_id(self):
        """BulkDownloadEventBase must carry user_id so events can be routed privately."""
        from invokeai.app.services.events.events_common import (
            BulkDownloadCompleteEvent,
            BulkDownloadErrorEvent,
            BulkDownloadEventBase,
            BulkDownloadStartedEvent,
        )

        assert "user_id" in BulkDownloadEventBase.model_fields

        started = BulkDownloadStartedEvent.build("default", "item-1", "item-1.zip", user_id="owner-abc")
        assert started.user_id == "owner-abc"

        complete = BulkDownloadCompleteEvent.build("default", "item-2", "item-2.zip", user_id="owner-abc")
        assert complete.user_id == "owner-abc"

        error = BulkDownloadErrorEvent.build("default", "item-3", "item-3.zip", "oops", user_id="owner-abc")
        assert error.user_id == "owner-abc"

    def test_bulk_download_event_not_emitted_to_shared_default_room(self, mock_invoker: Invoker, monkeypatch: Any):
        """Bulk download capability tokens must not be broadcast to the shared default room."""
        import asyncio
        from unittest.mock import AsyncMock

        from fastapi import FastAPI

        from invokeai.app.api.sockets import SocketIO
        from invokeai.app.services.events.events_common import BulkDownloadCompleteEvent

        mock_deps = MockApiDependencies(mock_invoker)
        monkeypatch.setattr("invokeai.app.api.dependencies.ApiDependencies", mock_deps)

        fastapi_app = FastAPI()
        socketio = SocketIO(fastapi_app)

        event = BulkDownloadCompleteEvent.build("default", "item-x", "item-x.zip", user_id="owner-xyz")

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_bulk_image_download_event(("bulk_download_complete", event)))

        rooms_emitted_to = [call.kwargs.get("room") for call in mock_emit.call_args_list]
        assert "default" not in rooms_emitted_to
        assert "user:owner-xyz" in rooms_emitted_to


# ===========================================================================
# 12. WebSocket authentication and event scoping
# ===========================================================================


class TestWebSocketAuth:
    """Tests that anonymous WebSocket clients cannot subscribe to queue rooms
    in multiuser mode, and that queue item events are scoped to the owner +
    admin rooms instead of being broadcast to the full queue room."""

    @pytest.fixture
    def socketio(self, mock_invoker: Invoker, monkeypatch: Any):
        """Create a SocketIO instance wired to the mock invoker's configuration."""
        from fastapi import FastAPI

        from invokeai.app.api.sockets import SocketIO

        # The SocketIO connect/sub handlers look up ApiDependencies.invoker.services.configuration.multiuser
        # at request time. Patch it to point at the mock invoker.
        mock_deps = MockApiDependencies(mock_invoker)
        monkeypatch.setattr("invokeai.app.api.dependencies.ApiDependencies", mock_deps)

        fastapi_app = FastAPI()
        return SocketIO(fastapi_app)

    def test_connect_rejected_without_token_in_multiuser_mode(self, socketio: Any, mock_invoker: Invoker) -> None:
        """In multiuser mode, _handle_connect must return False when no valid token is provided."""
        import asyncio

        mock_invoker.services.configuration.multiuser = True

        result = asyncio.run(socketio._handle_connect("sid-anon-1", environ={}, auth=None))
        assert result is False
        # The socket must not be recorded in the users dict
        assert "sid-anon-1" not in socketio._socket_users

    def test_connect_rejected_with_invalid_token_in_multiuser_mode(
        self, socketio: Any, mock_invoker: Invoker, setup_jwt_secret: None
    ) -> None:
        """An invalid/garbage token in multiuser mode must still be rejected."""
        import asyncio

        mock_invoker.services.configuration.multiuser = True

        result = asyncio.run(socketio._handle_connect("sid-bad-1", environ={}, auth={"token": "not-a-real-token"}))
        assert result is False
        assert "sid-bad-1" not in socketio._socket_users

    def test_connect_accepted_without_token_in_single_user_mode(self, socketio: Any, mock_invoker: Invoker) -> None:
        """In single-user mode, the socket handler should accept unauthenticated connections
        as the system admin user (matching how the REST API's get_current_user_or_default behaves)."""
        import asyncio

        mock_invoker.services.configuration.multiuser = False
        socketio._sio.enter_room = AsyncMock()

        result = asyncio.run(socketio._handle_connect("sid-single-1", environ={}, auth=None))
        assert result is True
        assert socketio._socket_users["sid-single-1"]["user_id"] == "system"
        assert socketio._socket_users["sid-single-1"]["is_admin"] is True
        socketio._sio.enter_room.assert_any_call("sid-single-1", "user:system")
        socketio._sio.enter_room.assert_any_call("sid-single-1", "workflows:shared")
        socketio._sio.enter_room.assert_any_call("sid-single-1", "admin")

    def test_connect_accepted_with_valid_token_in_multiuser_mode(
        self,
        socketio: Any,
        mock_invoker: Invoker,
        setup_jwt_secret: None,
    ) -> None:
        """A valid token in multiuser mode should be accepted with the correct user identity."""
        import asyncio

        from invokeai.app.services.auth.token_service import TokenData, create_access_token
        from invokeai.app.services.users.users_common import UserCreateRequest

        mock_invoker.services.configuration.multiuser = True
        socketio._sio.enter_room = AsyncMock()

        # Create the user in the database so the active-user check passes
        user = mock_invoker.services.users.create(
            UserCreateRequest(email="real@test.com", display_name="Real User", password="Test1234!@#$")
        )
        token = create_access_token(TokenData(user_id=user.user_id, email=user.email, is_admin=False))

        result = asyncio.run(socketio._handle_connect("sid-good-1", environ={}, auth={"token": token}))
        assert result is True
        assert socketio._socket_users["sid-good-1"]["user_id"] == user.user_id
        assert socketio._socket_users["sid-good-1"]["is_admin"] is False

    def test_connect_rejected_for_deleted_user_in_multiuser_mode(
        self, socketio: Any, mock_invoker: Invoker, setup_jwt_secret: None
    ) -> None:
        """A structurally valid JWT for a user that no longer exists in the database
        must be rejected.  This mirrors the REST auth check in auth_dependencies.py:53-58."""
        import asyncio

        from invokeai.app.services.auth.token_service import TokenData, create_access_token

        mock_invoker.services.configuration.multiuser = True
        # Create a token for a user_id that was never created in the user service
        token = create_access_token(TokenData(user_id="deleted-user-999", email="gone@test.com", is_admin=False))

        result = asyncio.run(socketio._handle_connect("sid-deleted-1", environ={}, auth={"token": token}))
        assert result is False
        assert "sid-deleted-1" not in socketio._socket_users

    def test_connect_rejected_for_inactive_user_in_multiuser_mode(
        self, socketio: Any, mock_invoker: Invoker, setup_jwt_secret: None
    ) -> None:
        """A structurally valid JWT for a deactivated user must be rejected even though
        the token itself has not expired."""
        import asyncio

        from invokeai.app.services.auth.token_service import TokenData, create_access_token
        from invokeai.app.services.users.users_common import UserCreateRequest

        mock_invoker.services.configuration.multiuser = True

        # Create a real user, then deactivate them
        user = mock_invoker.services.users.create(
            UserCreateRequest(email="inactive@test.com", display_name="Inactive", password="Test1234!@#$")
        )
        token = create_access_token(TokenData(user_id=user.user_id, email=user.email, is_admin=False))

        # Deactivate the user
        from invokeai.app.services.users.users_common import UserUpdateRequest

        mock_invoker.services.users.update(user.user_id, UserUpdateRequest(is_active=False))

        result = asyncio.run(socketio._handle_connect("sid-inactive-1", environ={}, auth={"token": token}))
        assert result is False
        assert "sid-inactive-1" not in socketio._socket_users

    def test_sub_queue_refuses_unknown_socket_in_multiuser_mode(self, socketio: Any, mock_invoker: Invoker) -> None:
        """If a socket somehow reaches _handle_sub_queue without a recorded identity
        in multiuser mode (e.g. bug, race), it must be refused rather than falling back
        to an anonymous system user who could then observe queue item events."""
        import asyncio

        mock_invoker.services.configuration.multiuser = True

        # Call sub_queue without a corresponding connect — the sid is unknown.
        asyncio.run(socketio._handle_sub_queue("sid-ghost-1", {"queue_id": "default"}))

        # The ghost socket must not have been added to the internal users dict
        assert "sid-ghost-1" not in socketio._socket_users

    def test_queue_item_status_changed_has_user_id(self) -> None:
        """QueueItemStatusChangedEvent must carry user_id so _handle_queue_event can
        route it to the owner + admin rooms instead of the public queue room. Without
        this field the event falls through to the generic broadcast branch and any
        subscriber to the queue can observe cross-user queue activity."""
        from invokeai.app.services.events.events_common import (
            InvocationEventBase,
            QueueItemEventBase,
            QueueItemStatusChangedEvent,
        )

        # The event base carries a user_id field
        assert "user_id" in QueueItemEventBase.model_fields
        # QueueItemStatusChangedEvent inherits it
        assert "user_id" in QueueItemStatusChangedEvent.model_fields
        # It is NOT an InvocationEventBase (so the generic QueueItemEventBase branch
        # in _handle_queue_event must also handle it privately)
        assert not issubclass(QueueItemStatusChangedEvent, InvocationEventBase)

    def test_batch_enqueued_event_carries_user_id(self) -> None:
        """BatchEnqueuedEvent must carry user_id so it can be routed privately to the
        owner and admin rooms. Otherwise a subscriber on the same queue_id would see
        every other user's batch_id, origin and enqueued counts."""
        from invokeai.app.services.events.events_common import BatchEnqueuedEvent
        from invokeai.app.services.session_queue.session_queue_common import (
            Batch,
            EnqueueBatchResult,
        )
        from invokeai.app.services.shared.graph import Graph

        enqueue_result = EnqueueBatchResult(
            queue_id="default",
            enqueued=3,
            requested=3,
            batch=Batch(batch_id="batch-xyz", origin="workflows", graph=Graph()),
            priority=0,
            item_ids=[1, 2, 3],
        )
        event = BatchEnqueuedEvent.build(enqueue_result, user_id="owner-123")
        assert event.user_id == "owner-123"
        assert event.batch_id == "batch-xyz"
        assert event.queue_id == "default"

    def test_queue_item_status_changed_routed_privately(self, socketio: Any) -> None:
        """_handle_queue_event must emit the FULL QueueItemStatusChangedEvent only to the
        owner's user room and the admin room. A sanitized companion (user_id="redacted",
        identifiers stripped) is also emitted to the queue_id room so other users' UIs can
        refresh, with the owner's and admins' sids in skip_sid so they don't get a duplicate
        that would clobber their cache."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import QueueItemStatusChangedEvent
        from invokeai.app.services.session_queue.session_queue_common import (
            BatchStatus,
            SessionQueueStatus,
        )

        event = QueueItemStatusChangedEvent(
            queue_id="default",
            item_id=1,
            batch_id="batch-private",
            origin="workflows",
            destination="canvas",
            user_id="owner-xyz",
            session_id="sess-private",
            status="in_progress",
            created_at="2026-01-01T00:00:00",
            updated_at="2026-01-01T00:01:00",
            started_at="2026-01-01T00:00:30",
            completed_at=None,
            batch_status=BatchStatus(
                queue_id="default",
                batch_id="batch-private",
                origin="workflows",
                destination="canvas",
                pending=0,
                in_progress=1,
                waiting=0,
                completed=0,
                failed=0,
                canceled=0,
                total=1,
            ),
            queue_status=SessionQueueStatus(
                queue_id="default",
                item_id=1,
                session_id="sess-private",
                batch_id="batch-private",
                pending=0,
                in_progress=1,
                waiting=0,
                completed=0,
                failed=0,
                canceled=0,
                total=1,
            ),
        )

        # Track owner sid so we can verify skip_sid is honored
        socketio._socket_users["sid-owner"] = {"user_id": "owner-xyz", "is_admin": False}
        socketio._socket_users["sid-admin"] = {"user_id": "admin-1", "is_admin": True}
        socketio._socket_users["sid-other"] = {"user_id": "other-user", "is_admin": False}

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("queue_item_status_changed", event)))

        # Collect (room, payload, skip_sid) for each emit call
        emits = [
            (c.kwargs.get("room"), c.kwargs.get("data"), c.kwargs.get("skip_sid")) for c in mock_emit.call_args_list
        ]

        # Full event must go to owner room and admin room with original sensitive fields
        owner_emits = [(p, s) for r, p, s in emits if r == "user:owner-xyz"]
        admin_emits = [(p, s) for r, p, s in emits if r == "admin"]
        assert len(owner_emits) == 1 and len(admin_emits) == 1
        for payload, _ in owner_emits + admin_emits:
            assert payload["user_id"] == "owner-xyz"
            assert payload["batch_id"] == "batch-private"
            assert payload["session_id"] == "sess-private"
            assert payload["destination"] == "canvas"

        # A sanitized companion event must go to the queue_id room with sensitive fields cleared
        queue_emits = [(p, s) for r, p, s in emits if r == "default"]
        assert len(queue_emits) == 1, "expected exactly one sanitized emit to queue room"
        sanitized_payload, skip_sid = queue_emits[0]
        assert sanitized_payload["user_id"] == "redacted"
        assert sanitized_payload["batch_id"] == "redacted"
        assert sanitized_payload["session_id"] == "redacted"
        assert sanitized_payload["origin"] is None
        assert sanitized_payload["destination"] is None
        assert sanitized_payload["error_type"] is None
        assert sanitized_payload["batch_status"]["batch_id"] == "redacted"
        assert sanitized_payload["batch_status"]["destination"] is None
        assert sanitized_payload["queue_status"]["item_id"] is None
        assert sanitized_payload["queue_status"]["batch_id"] is None
        assert sanitized_payload["queue_status"]["user_pending"] is None
        # Owner and admin sids must be skipped so they don't receive the duplicate
        assert "sid-owner" in skip_sid
        assert "sid-admin" in skip_sid
        # Third-party user must NOT be skipped — they need the sanitized event
        assert "sid-other" not in skip_sid
        # Status (non-sensitive) is preserved so the non-owner UI knows what changed
        assert sanitized_payload["status"] == "in_progress"
        assert sanitized_payload["item_id"] == 1

    def test_batch_enqueued_routed_privately(self, socketio: Any) -> None:
        """_handle_queue_event must emit the FULL BatchEnqueuedEvent only to the owner's
        user room and the admin room. A sanitized companion (user_id="redacted", batch_id
        and origin stripped) is also emitted to the queue_id room so other users' badge
        totals refresh, with owner/admin sids in skip_sid."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import BatchEnqueuedEvent
        from invokeai.app.services.session_queue.session_queue_common import (
            Batch,
            EnqueueBatchResult,
        )
        from invokeai.app.services.shared.graph import Graph

        enqueue_result = EnqueueBatchResult(
            queue_id="default",
            enqueued=5,
            requested=5,
            batch=Batch(batch_id="batch-pvt", origin="workflows", graph=Graph()),
            priority=0,
            item_ids=[10, 11, 12, 13, 14],
        )
        event = BatchEnqueuedEvent.build(enqueue_result, user_id="owner-zzz")

        socketio._socket_users["sid-owner"] = {"user_id": "owner-zzz", "is_admin": False}
        socketio._socket_users["sid-admin"] = {"user_id": "admin-1", "is_admin": True}
        socketio._socket_users["sid-other"] = {"user_id": "other-user", "is_admin": False}

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("batch_enqueued", event)))

        emits = [
            (c.kwargs.get("room"), c.kwargs.get("data"), c.kwargs.get("skip_sid")) for c in mock_emit.call_args_list
        ]

        # Full event to owner + admin contains the real batch_id and origin
        owner_emits = [(p, s) for r, p, s in emits if r == "user:owner-zzz"]
        admin_emits = [(p, s) for r, p, s in emits if r == "admin"]
        assert len(owner_emits) == 1 and len(admin_emits) == 1
        for payload, _ in owner_emits + admin_emits:
            assert payload["user_id"] == "owner-zzz"
            assert payload["batch_id"] == "batch-pvt"
            assert payload["origin"] == "workflows"

        # Sanitized event to queue room: user/batch/origin redacted, owner+admin skipped
        queue_emits = [(p, s) for r, p, s in emits if r == "default"]
        assert len(queue_emits) == 1
        sanitized_payload, skip_sid = queue_emits[0]
        assert sanitized_payload["user_id"] == "redacted"
        assert sanitized_payload["batch_id"] == "redacted"
        assert sanitized_payload["origin"] is None
        assert sanitized_payload["enqueued"] == 5  # count is non-sensitive
        assert "sid-owner" in skip_sid
        assert "sid-admin" in skip_sid
        assert "sid-other" not in skip_sid

    def test_queue_items_retried_event_carries_user_ids(self) -> None:
        """QueueItemsRetriedEvent must carry owner identity so retry notifications are
        routed privately instead of broadcast to all queue subscribers."""
        from invokeai.app.services.events.events_common import QueueItemsRetriedEvent
        from invokeai.app.services.session_queue.session_queue_common import RetryItemsResult

        retry_result = RetryItemsResult(queue_id="default", retried_item_ids=[10])
        event = QueueItemsRetriedEvent.build(
            retry_result, user_ids=["owner-123"], retried_item_ids_by_user={"owner-123": [10]}
        )

        assert event.user_ids == ["owner-123"]
        assert event.retried_item_ids == [10]
        assert event.retried_item_ids_by_user == {"owner-123": [10]}
        assert event.queue_id == "default"

    def test_queue_items_retried_routed_privately(self, socketio: Any) -> None:
        """Verify that queue_items_retried is emitted only to owner/admin rooms."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import QueueItemsRetriedEvent
        from invokeai.app.services.session_queue.session_queue_common import RetryItemsResult

        event = QueueItemsRetriedEvent.build(
            RetryItemsResult(queue_id="default", retried_item_ids=[10]),
            user_ids=["owner-123", "owner-456"],
            retried_item_ids_by_user={"owner-123": [10], "owner-456": []},
        )

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("queue_items_retried", event)))

        rooms_emitted_to = [call.kwargs.get("room") for call in mock_emit.call_args_list]
        assert "user:owner-123" in rooms_emitted_to
        assert "user:owner-456" in rooms_emitted_to
        assert "admin" in rooms_emitted_to
        assert "default" not in rooms_emitted_to

    def test_queue_items_retried_payload_is_filtered_per_owner_room(self, socketio: Any) -> None:
        """Each owner room should only receive retry payload for that owner."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import QueueItemsRetriedEvent
        from invokeai.app.services.session_queue.session_queue_common import RetryItemsResult

        event = QueueItemsRetriedEvent.build(
            RetryItemsResult(queue_id="default", retried_item_ids=[10, 20]),
            user_ids=["owner-123", "owner-456"],
            retried_item_ids_by_user={"owner-123": [10], "owner-456": [20]},
        )

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("queue_items_retried", event)))

        owner_123_calls = [call for call in mock_emit.call_args_list if call.kwargs.get("room") == "user:owner-123"]
        owner_456_calls = [call for call in mock_emit.call_args_list if call.kwargs.get("room") == "user:owner-456"]
        admin_calls = [call for call in mock_emit.call_args_list if call.kwargs.get("room") == "admin"]

        assert len(owner_123_calls) == 1
        assert len(owner_456_calls) == 1
        assert len(admin_calls) == 1
        assert owner_123_calls[0].kwargs["data"]["retried_item_ids"] == [10]
        assert owner_123_calls[0].kwargs["data"]["user_ids"] == ["owner-123"]
        assert owner_123_calls[0].kwargs["data"]["retried_item_ids_by_user"] == {"owner-123": [10]}
        assert owner_456_calls[0].kwargs["data"]["retried_item_ids"] == [20]
        assert owner_456_calls[0].kwargs["data"]["user_ids"] == ["owner-456"]
        assert owner_456_calls[0].kwargs["data"]["retried_item_ids_by_user"] == {"owner-456": [20]}
        assert admin_calls[0].kwargs["data"]["retried_item_ids"] == [10, 20]
        assert admin_calls[0].kwargs["data"]["user_ids"] == ["owner-123", "owner-456"]
        assert admin_calls[0].kwargs["data"]["retried_item_ids_by_user"] == {"owner-123": [10], "owner-456": [20]}

    def test_unscoped_queue_cleared_still_broadcast(self, socketio: Any) -> None:
        """An unscoped QueueClearedEvent (user_id=None — an admin or single-user clear that
        deleted every user's items) should still be broadcast to all queue subscribers."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import QueueClearedEvent

        event = QueueClearedEvent.build(queue_id="default")

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("queue_cleared", event)))

        assert len(mock_emit.call_args_list) == 1
        assert mock_emit.call_args_list[0].kwargs.get("room") == "default"
        assert mock_emit.call_args_list[0].kwargs.get("data")["user_id"] is None

    def test_user_scoped_queue_cleared_routed_privately(self, socketio: Any) -> None:
        """A user-scoped QueueClearedEvent only deleted that user's rows. The full event must
        go to the owner + admin rooms; the rest of the queue room gets a sanitized companion
        (user_id="redacted") so their queue lists refresh without treating the clear as their
        own — otherwise another user's clear would abort their in-flight reconciliations and
        mark their tracked items canceled."""
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import QueueClearedEvent

        event = QueueClearedEvent.build(queue_id="default", user_id="owner-xyz")

        socketio._socket_users["sid-owner"] = {"user_id": "owner-xyz", "is_admin": False}
        socketio._socket_users["sid-admin"] = {"user_id": "admin-1", "is_admin": True}
        socketio._socket_users["sid-other"] = {"user_id": "other-user", "is_admin": False}

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("queue_cleared", event)))

        emits = [
            (c.kwargs.get("room"), c.kwargs.get("data"), c.kwargs.get("skip_sid")) for c in mock_emit.call_args_list
        ]

        # Full event goes to owner + admin rooms in a single emit (room list deduplicates a
        # socket that is in both rooms)
        private_emits = [(p, s) for r, p, s in emits if r == ["user:owner-xyz", "admin"]]
        assert len(private_emits) == 1
        assert private_emits[0][0]["user_id"] == "owner-xyz"

        # Sanitized companion goes to the queue room, skipping the owner's and admins' sids
        queue_emits = [(p, s) for r, p, s in emits if r == "default"]
        assert len(queue_emits) == 1, "expected exactly one sanitized emit to queue room"
        sanitized_payload, skip_sid = queue_emits[0]
        assert sanitized_payload["user_id"] == "redacted"
        assert "sid-owner" in skip_sid
        assert "sid-admin" in skip_sid
        assert "sid-other" not in skip_sid

    def test_recall_parameters_emitted_once_to_owner_and_admin_rooms(self, socketio: Any) -> None:
        """RecallParametersUpdatedEvent must be delivered to the owner + admin rooms
        in a SINGLE emit call (room list), not two separate emits.

        A socket that is in both rooms — e.g. the system user in single-user mode,
        who is also an admin — would otherwise receive the event twice. That is
        harmless for the idempotent scalar recall fields but doubles every entry
        for the append-mode reference-image recall, which pushes rather than
        replaces. python-socketio deduplicates recipients across a room list, so
        a single emit to [user_room, "admin"] delivers exactly once per socket.
        """
        import asyncio
        from unittest.mock import AsyncMock

        from invokeai.app.services.events.events_common import RecallParametersUpdatedEvent

        event = RecallParametersUpdatedEvent.build(
            queue_id="default",
            user_id="owner-recall",
            parameters={"reference_images": [{"image": {"image_name": "cat.png"}}], "append": True},
        )

        mock_emit = AsyncMock()
        socketio._sio.emit = mock_emit

        asyncio.run(socketio._handle_queue_event(("recall_parameters_updated", event)))

        # Exactly one emit, targeting the union of the owner and admin rooms.
        assert mock_emit.call_count == 1, (
            "recall event must be emitted once to a room list, not once per room — "
            "two emits double-deliver to a socket in both rooms"
        )
        room = mock_emit.call_args.kwargs.get("room")
        assert isinstance(room, list)
        assert set(room) == {"user:owner-recall", "admin"}
        # And never to the shared queue room, which would leak to other users.
        assert "default" not in room


class TestCustomNodesAuthorization:
    """Tests that custom_nodes endpoints enforce AdminUserOrDefault.

    All four routes (list, install, uninstall, reload) should reject
    unauthenticated callers and non-admin users in multiuser mode,
    and succeed for admin callers.
    """

    # -- unauthenticated -------------------------------------------------------

    def test_list_rejects_unauthenticated(self, client: TestClient, enable_multiuser: Any) -> None:
        r = client.get("/api/v2/custom_nodes/")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_install_rejects_unauthenticated(self, client: TestClient, enable_multiuser: Any) -> None:
        r = client.post("/api/v2/custom_nodes/install", json={"source": "https://example.com/repo.git"})
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_uninstall_rejects_unauthenticated(self, client: TestClient, enable_multiuser: Any) -> None:
        r = client.delete("/api/v2/custom_nodes/some_pack")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    def test_reload_rejects_unauthenticated(self, client: TestClient, enable_multiuser: Any) -> None:
        r = client.post("/api/v2/custom_nodes/reload")
        assert r.status_code == status.HTTP_401_UNAUTHORIZED

    # -- non-admin user --------------------------------------------------------

    def test_list_rejects_non_admin(self, client: TestClient, user1_token: str) -> None:
        r = client.get("/api/v2/custom_nodes/", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_install_rejects_non_admin(self, client: TestClient, user1_token: str) -> None:
        r = client.post(
            "/api/v2/custom_nodes/install",
            json={"source": "https://example.com/repo.git"},
            headers=_auth(user1_token),
        )
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_uninstall_rejects_non_admin(self, client: TestClient, user1_token: str) -> None:
        r = client.delete("/api/v2/custom_nodes/some_pack", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    def test_reload_rejects_non_admin(self, client: TestClient, user1_token: str) -> None:
        r = client.post("/api/v2/custom_nodes/reload", headers=_auth(user1_token))
        assert r.status_code == status.HTTP_403_FORBIDDEN

    # -- admin caller succeeds -------------------------------------------------

    def test_list_allows_admin(self, client: TestClient, admin_token: str) -> None:
        r = client.get("/api/v2/custom_nodes/", headers=_auth(admin_token))
        assert r.status_code == status.HTTP_200_OK

    def test_reload_allows_admin(self, client: TestClient, admin_token: str, monkeypatch: Any) -> None:
        # Stub load_custom_nodes so it doesn't actually scan the filesystem
        monkeypatch.setattr(
            "invokeai.app.api.routers.custom_nodes.load_custom_nodes", lambda *a, **kw: None, raising=False
        )
        r = client.post("/api/v2/custom_nodes/reload", headers=_auth(admin_token))
        assert r.status_code == status.HTTP_200_OK

    def test_install_allows_admin(self, client: TestClient, admin_token: str, monkeypatch: Any, tmp_path: Any) -> None:
        """Admin caller can successfully install a node pack (filesystem/subprocess mocked)."""
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._get_custom_nodes_path", lambda: tmp_path)

        # Simulate a successful git clone by creating the target dir with __init__.py
        def fake_git_clone(cmd: list[str], **kwargs: Any) -> MagicMock:
            target_dir = tmp_path / "test-pack"
            target_dir.mkdir(parents=True, exist_ok=True)
            (target_dir / "__init__.py").touch()
            result = MagicMock()
            result.returncode = 0
            return result

        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes.subprocess.run", fake_git_clone)
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._load_node_pack", lambda *a, **kw: None)
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._import_workflows_from_pack", lambda *a, **kw: [])
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._write_pack_manifest", lambda *a, **kw: None)

        r = client.post(
            "/api/v2/custom_nodes/install",
            json={"source": "https://example.com/test-pack.git"},
            headers=_auth(admin_token),
        )
        assert r.status_code == status.HTTP_200_OK
        data = r.json()
        assert data["success"] is True
        assert data["name"] == "test-pack"

    def test_uninstall_allows_admin(
        self, client: TestClient, admin_token: str, monkeypatch: Any, tmp_path: Any
    ) -> None:
        """Admin caller can successfully uninstall a node pack (filesystem mocked)."""
        # Create a fake installed pack directory
        pack_dir = tmp_path / "test-pack"
        pack_dir.mkdir()
        (pack_dir / "__init__.py").touch()

        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._get_custom_nodes_path", lambda: tmp_path)
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._read_pack_manifest", lambda *a, **kw: [])
        monkeypatch.setattr(
            "invokeai.app.api.routers.custom_nodes.InvocationRegistry.unregister_pack",
            lambda *a, **kw: [],
        )
        monkeypatch.setattr("invokeai.app.api.routers.custom_nodes._remove_workflows_by_ids", lambda *a, **kw: 0)

        r = client.delete("/api/v2/custom_nodes/test-pack", headers=_auth(admin_token))
        assert r.status_code == status.HTTP_200_OK
        data = r.json()
        assert data["success"] is True
        assert data["name"] == "test-pack"