from __future__ import annotations

import builtins
import sys
from datetime import datetime
from importlib import util
from pathlib import Path
from types import ModuleType, SimpleNamespace
from unittest.mock import MagicMock

import pytest
from flask import Flask
from flask.views import MethodView
from pydantic import ValidationError
from werkzeug.datastructures import MultiDict

from configs import dify_config

# kombu references MethodView as a global when importing celery/kombu pools.
if not hasattr(builtins, "MethodView"):
    builtins.MethodView = MethodView  # type: ignore[attr-defined]


class _ConsoleModule(ModuleType):
    console_ns: object
    api: object | None
    bp: object | None
    app: ModuleType


def _unwrap(func):
    bound_self = getattr(func, "__self__", None)
    while hasattr(func, "__wrapped__"):
        func = func.__wrapped__
    if bound_self is not None:
        return func.__get__(bound_self, bound_self.__class__)
    return func


@pytest.fixture(scope="module")
def app_module():
    module_name = "controllers.console.app.app"
    root = Path(__file__).resolve().parents[5]
    module_path = root / "controllers" / "console" / "app" / "app.py"

    class _StubNamespace:
        def __init__(self):
            self.models: dict[str, object] = {}
            self.payload = None

        def schema_model(self, name, schema):
            self.models[name] = schema
            return schema

        def model(self, name, model_dict=None, **kwargs):
            """Register a model with the namespace (flask-restx compatibility)."""
            if model_dict is not None:
                self.models[name] = model_dict
            return model_dict

        def _decorator(self, obj):
            return obj

        def doc(self, *args, **kwargs):
            return self._decorator

        def expect(self, *args, **kwargs):
            return self._decorator

        def response(self, *args, **kwargs):
            return self._decorator

        def route(self, *args, **kwargs):
            def decorator(obj):
                return obj

            return decorator

    stub_namespace = _StubNamespace()

    original_modules: dict[str, ModuleType | None] = {
        "controllers.console": sys.modules.get("controllers.console"),
        "controllers.console.app": sys.modules.get("controllers.console.app"),
        "controllers.common.schema": sys.modules.get("controllers.common.schema"),
        module_name: sys.modules.get(module_name),
    }
    stubbed_modules: list[tuple[str, ModuleType | None]] = []

    console_module = _ConsoleModule("controllers.console")
    console_module.__path__ = [str(root / "controllers" / "console")]
    console_module.console_ns = stub_namespace
    console_module.api = None
    console_module.bp = None
    sys.modules["controllers.console"] = console_module

    app_package = ModuleType("controllers.console.app")
    app_package.__path__ = [str(root / "controllers" / "console" / "app")]
    sys.modules["controllers.console.app"] = app_package
    console_module.app = app_package

    def _stub_module(name: str, attrs: dict[str, object]) -> None:
        original = sys.modules.get(name)
        module = ModuleType(name)
        for key, value in attrs.items():
            setattr(module, key, value)
        sys.modules[name] = module
        stubbed_modules.append((name, original))

    class _OpsTraceManager:
        @staticmethod
        def get_app_tracing_config(app_id: str) -> dict[str, object]:
            return {}

        @staticmethod
        def update_app_tracing_config(app_id: str, **kwargs) -> None:
            return None

    _stub_module(
        "core.ops.ops_trace_manager",
        {
            "OpsTraceManager": _OpsTraceManager,
            "TraceQueueManager": object,
            "TraceTask": object,
        },
    )

    spec = util.spec_from_file_location(module_name, module_path)
    assert spec is not None
    module = util.module_from_spec(spec)
    sys.modules[module_name] = module

    assert spec.loader is not None
    spec.loader.exec_module(module)

    try:
        yield module
    finally:
        for name, original in reversed(stubbed_modules):
            if original is not None:
                sys.modules[name] = original
            else:
                sys.modules.pop(name, None)
        for name, original in original_modules.items():
            if original is not None:
                sys.modules[name] = original
            else:
                sys.modules.pop(name, None)


@pytest.fixture(scope="module")
def app_models(app_module):
    return SimpleNamespace(
        AppDetailWithSite=app_module.AppDetailWithSite,
        AppPagination=app_module.AppPagination,
        AppPartial=app_module.AppPartial,
    )


@pytest.fixture(autouse=True)
def patch_signed_url(monkeypatch: pytest.MonkeyPatch, app_module: ModuleType) -> None:
    """Ensure icon URL generation uses a deterministic helper for tests."""

    def _fake_build_icon_url(_icon_type, key: str | None) -> str | None:
        if key is None:
            return None
        icon_type = str(_icon_type).lower()
        if icon_type != "image":
            return None
        return f"signed:{key}"

    monkeypatch.setattr(app_module, "build_icon_url", _fake_build_icon_url)


def _ts(hour: int = 12) -> datetime:
    return datetime(2024, 1, 1, hour, 0, 0)


def _dummy_model_config():
    return SimpleNamespace(
        model_dict={"provider": "openai", "name": "gpt-4o"},
        pre_prompt="hello",
        created_by="config-author",
        created_at=_ts(9),
        updated_by="config-editor",
        updated_at=_ts(10),
    )


def _dummy_workflow():
    return SimpleNamespace(
        id="wf-1",
        created_by="workflow-author",
        created_at=_ts(8),
        updated_by="workflow-editor",
        updated_at=_ts(9),
    )


def test_app_list_query_reads_repeated_tag_ids(app_module):
    first_tag_id = "8c4ef3d1-58a1-4d94-8a1c-1c171d889e08"
    second_tag_id = "3c39395b-6d1f-4030-8b17-eaa7cc85221c"
    query_args = MultiDict(
        [
            ("page", "1"),
            ("limit", "30"),
            ("tag_ids", first_tag_id),
            ("tag_ids", second_tag_id),
        ]
    )

    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=query_args,
    )

    assert query.tag_ids == [first_tag_id, second_tag_id]


def test_app_list_query_reads_repeated_creator_ids(app_module):
    first_creator_id = "9e8959cf-a67b-4d34-9906-1d687517b248"
    second_creator_id = "1886f96a-5bf0-42bf-961d-8d2129049076"
    query_args = MultiDict(
        [
            ("page", "1"),
            ("limit", "30"),
            ("creator_ids", first_creator_id),
            ("creator_ids", second_creator_id),
        ]
    )

    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=query_args,
    )

    assert query.creator_ids == [first_creator_id, second_creator_id]


def test_app_list_query_preserves_regular_query_params(app_module):
    query_args = MultiDict(
        [
            ("page", "2"),
            ("limit", "50"),
            ("mode", "chat"),
            ("name", "Sales Copilot"),
            ("is_created_by_me", "true"),
        ]
    )

    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=query_args,
    )

    assert query.page == 2
    assert query.limit == 50
    assert query.mode == "chat"
    assert query.name == "Sales Copilot"
    assert query.is_created_by_me is True
    assert query.tag_ids is None


def test_app_list_query_normalizes_empty_repeated_tag_ids_to_none(app_module):
    query_args = MultiDict(
        [
            ("tag_ids", ""),
            ("tag_ids", "   "),
        ]
    )

    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=query_args,
    )

    assert query.tag_ids is None


def test_app_list_query_rejects_invalid_repeated_tag_id(app_module):
    with pytest.raises(ValidationError):
        app_module.query_params_from_request(
            app_module.AppListQuery,
            list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
            args=MultiDict([("tag_ids", "not-a-uuid")]),
        )


def test_app_list_query_rejects_invalid_repeated_creator_id(app_module):
    with pytest.raises(ValidationError):
        app_module.query_params_from_request(
            app_module.AppListQuery,
            list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
            args=MultiDict([("creator_ids", "not-a-uuid")]),
        )


def test_app_list_query_ignores_indexed_tag_ids(app_module):
    tag_id = "8c4ef3d1-58a1-4d94-8a1c-1c171d889e08"
    query_args = MultiDict(
        [
            ("tag_ids[0]", tag_id),
        ]
    )

    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=query_args,
    )

    assert query.tag_ids is None


def test_app_list_query_accepts_single_repeated_tag_id(app_module):
    tag_id = "8c4ef3d1-58a1-4d94-8a1c-1c171d889e08"
    query = app_module.query_params_from_request(
        app_module.AppListQuery,
        list_fields=app_module.APP_LIST_QUERY_ARRAY_FIELDS,
        args=MultiDict([("tag_ids", tag_id)]),
    )

    assert query.tag_ids == [tag_id]


def test_create_app_endpoint_rejects_agent_mode(app_module, monkeypatch: pytest.MonkeyPatch):
    payload = {"name": "Iris", "mode": "agent", "description": "Agent app"}
    app_service = MagicMock()
    monkeypatch.setattr(app_module, "AppService", lambda: app_service)

    app_module.console_ns.payload = payload
    try:
        with pytest.raises(ValidationError):
            _unwrap(app_module.AppListApi().post)(MagicMock(), "tenant-1", SimpleNamespace(id="account-1"))
    finally:
        app_module.console_ns.payload = None

    app_service.create_app.assert_not_called()


def test_app_partial_serialization_uses_aliases(app_models):
    AppPartial = app_models.AppPartial
    created_at = _ts()
    app_obj = SimpleNamespace(
        id="app-1",
        name="My App",
        desc_or_prompt="Prompt snippet",
        mode_compatible_with_agent="chat",
        icon_type="image",
        icon="icon-key",
        icon_background="#fff",
        app_model_config=_dummy_model_config(),
        workflow=_dummy_workflow(),
        created_by="creator",
        created_at=created_at,
        updated_by="editor",
        updated_at=created_at,
        tags=[SimpleNamespace(id="tag-1", name="Utilities", type="app")],
        access_mode="private",
        create_user_name="Creator",
        author_name="Author",
        has_draft_trigger=True,
        permission_keys=["app.acl.view_layout"],
        role="Should stay agent-only",
    )

    serialized = AppPartial.model_validate(app_obj, from_attributes=True).model_dump(mode="json")

    assert serialized["description"] == "Prompt snippet"
    assert serialized["mode"] == "chat"
    assert serialized["icon_url"] == "signed:icon-key"
    assert serialized["created_at"] == int(created_at.timestamp())
    assert serialized["updated_at"] == int(created_at.timestamp())
    assert serialized["model_config"]["model"] == {"provider": "openai", "name": "gpt-4o"}
    assert serialized["workflow"]["id"] == "wf-1"
    assert serialized["tags"][0]["name"] == "Utilities"
    assert serialized["permission_keys"] == ["app.acl.view_layout"]
    assert "role" not in serialized


def test_app_detail_with_site_includes_nested_serialization(app_models):
    AppDetailWithSite = app_models.AppDetailWithSite
    timestamp = _ts(14)
    site = SimpleNamespace(
        code="site-code",
        title="Public Site",
        icon_type="image",
        icon="site-icon",
        input_placeholder="Ask anything",
        created_at=timestamp,
        updated_at=timestamp,
    )
    app_obj = SimpleNamespace(
        id="app-2",
        name="Detailed App",
        description="Desc",
        mode_compatible_with_agent="advanced-chat",
        icon_type="image",
        icon="detail-icon",
        icon_background="#123456",
        enable_site=True,
        enable_api=True,
        app_model_config={
            "opening_statement": "hi",
            "model": {"provider": "openai", "name": "gpt-4o"},
            "retriever_resource": {"enabled": True},
        },
        workflow=_dummy_workflow(),
        tracing={"enabled": True},
        use_icon_as_answer_icon=True,
        created_by="creator",
        created_at=timestamp,
        updated_by="editor",
        updated_at=timestamp,
        access_mode="public",
        tags=[SimpleNamespace(id="tag-2", name="Prod", type="app")],
        permission_keys=["app.acl.view_layout", "app.acl.edit"],
        api_base_url="https://api.example.com/v1",
        max_active_requests=5,
        deleted_tools=[{"type": "api", "tool_name": "search", "provider_id": "prov"}],
        site=site,
        bound_agent_id="agent-1",
        role="Should stay agent-only",
    )

    serialized = AppDetailWithSite.model_validate(app_obj, from_attributes=True).model_dump(mode="json")

    assert serialized["icon_url"] == "signed:detail-icon"
    assert serialized["model_config"]["retriever_resource"] == {"enabled": True}
    assert serialized["deleted_tools"][0]["tool_name"] == "search"
    assert serialized["site"]["icon_url"] == "signed:site-icon"
    assert serialized["site"]["input_placeholder"] == "Ask anything"
    assert serialized["site"]["created_at"] == int(timestamp.timestamp())
    assert serialized["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
    assert serialized["bound_agent_id"] == "agent-1"
    assert "role" not in serialized


def test_app_response_view_uses_the_caller_session_for_query_backed_fields(app_module, monkeypatch):
    session = MagicMock()
    app_obj = MagicMock()
    app_model_config = SimpleNamespace(app_id="app-1")
    app_obj.desc_or_prompt_with_session.return_value = "Description"
    app_obj.site_with_session.return_value = SimpleNamespace(id="site-1")
    app_obj.app_model_config_with_session.return_value = app_model_config
    app_obj.workflow_with_session.return_value = SimpleNamespace(id="workflow-1")
    app_obj.bound_agent_id_with_session.return_value = "agent-1"
    app_obj.mode_compatible_with_agent_with_session.return_value = "agent"
    app_obj.deleted_tools_with_session.return_value = []
    app_obj.tags_with_session.return_value = []
    app_obj.author_name_with_session.return_value = "Author"
    load_annotation_reply = MagicMock(return_value={"enabled": False})
    monkeypatch.setattr("services.app_service.load_annotation_reply_config", load_annotation_reply)

    view = app_module.AppResponseView(app_obj, session=session)
    site = view.site
    workflow = view.workflow
    model_config = view.app_model_config

    assert view.desc_or_prompt == "Description"
    assert site is not None
    assert site.id == "site-1"
    assert workflow is not None
    assert workflow.id == "workflow-1"
    assert view.bound_agent_id == "agent-1"
    assert view.mode_compatible_with_agent == "agent"
    assert view.deleted_tools == []
    assert view.tags == []
    assert view.author_name == "Author"
    assert model_config is not None
    assert model_config.annotation_reply_dict == {"enabled": False}
    for method in (
        app_obj.desc_or_prompt_with_session,
        app_obj.site_with_session,
        app_obj.app_model_config_with_session,
        app_obj.workflow_with_session,
        app_obj.bound_agent_id_with_session,
        app_obj.mode_compatible_with_agent_with_session,
        app_obj.deleted_tools_with_session,
        app_obj.tags_with_session,
        app_obj.author_name_with_session,
    ):
        method.assert_called_once_with(session=session)
    load_annotation_reply.assert_called_once_with(session, "app-1")


def test_app_pagination_aliases_per_page_and_has_next(app_models):
    AppPagination = app_models.AppPagination
    item_one = SimpleNamespace(
        id="app-10",
        name="Paginated One",
        desc_or_prompt="Summary",
        mode_compatible_with_agent="chat",
        icon_type="image",
        icon="first-icon",
        created_at=_ts(15),
        updated_at=_ts(15),
        permission_keys=["app.acl.edit"],
    )
    item_two = SimpleNamespace(
        id="app-11",
        name="Paginated Two",
        desc_or_prompt="Summary",
        mode_compatible_with_agent="agent-chat",
        icon_type="emoji",
        icon="🙂",
        created_at=_ts(16),
        updated_at=_ts(16),
    )
    pagination = SimpleNamespace(
        page=2,
        per_page=10,
        total=50,
        has_next=True,
        items=[item_one, item_two],
    )

    serialized = AppPagination.model_validate(pagination, from_attributes=True).model_dump(mode="json")

    assert serialized["page"] == 2
    assert serialized["limit"] == 10
    assert serialized["has_more"] is True
    assert len(serialized["data"]) == 2
    assert serialized["data"][0]["icon_url"] == "signed:first-icon"
    assert serialized["data"][1]["icon_url"] is None


def test_app_list_uses_injected_session_for_draft_workflows(
    app: Flask, app_module: ModuleType, monkeypatch: pytest.MonkeyPatch
) -> None:
    api = app_module.AppListApi()
    method = _unwrap(api.get)
    app_item = SimpleNamespace(
        id="app-1",
        name="Workflow App",
        desc_or_prompt="Summary",
        mode="workflow",
        mode_compatible_with_agent="workflow",
    )
    app_pagination = SimpleNamespace(page=1, per_page=20, total=1, has_next=False, items=[app_item])
    workflow = SimpleNamespace(
        id="workflow-1",
        app_id="app-1",
        walk_nodes=lambda: iter([("trigger-1", {"type": "trigger-webhook"})]),
    )
    session = MagicMock()
    session.execute.return_value.scalars.return_value.all.return_value = [workflow]
    scoped_session = MagicMock()
    scoped_session.execute.side_effect = AssertionError("db.session should not be used")

    monkeypatch.setattr(
        app_module,
        "AppService",
        lambda: SimpleNamespace(get_paginate_apps=lambda *_args, **_kwargs: app_pagination),
    )
    monkeypatch.setattr(
        app_module,
        "FeatureService",
        SimpleNamespace(get_system_features=lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))),
    )
    get_permissions = MagicMock(
        return_value=app_module.enterprise_rbac_service.MyPermissionsResponse(
            app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
                overrides=[
                    app_module.enterprise_rbac_service.ResourcePermissionKeys(
                        resource_id="app-1",
                        permission_keys=["app.acl.edit"],
                    )
                ]
            )
        )
    )
    monkeypatch.setattr(
        app_module.enterprise_rbac_service.RBACService.MyPermissions,
        "get",
        get_permissions,
    )
    monkeypatch.setattr(app_module, "db", SimpleNamespace(session=scoped_session))

    with app.test_request_context("/console/api/apps?page=1&limit=20", method="GET"):
        response, status = method("tenant-1", "user-1", session)

    assert status == 200
    assert response["data"][0]["has_draft_trigger"] is True
    session.execute.assert_called_once()
    scoped_session.execute.assert_not_called()
    get_permissions.assert_called_once_with("tenant-1", "user-1", session=session)
    assert response["data"][0]["permission_keys"] == ["app.acl.edit"]


def test_app_create_api_attaches_permission_keys(app, app_module):
    method = app_module.AppListApi.post
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    app_obj = SimpleNamespace(
        id="app-new",
        name="Created App",
        description="Summary",
        mode_compatible_with_agent="advanced-chat",
        enable_site=True,
        enable_api=True,
        permission_keys=[],
    )

    with app.test_request_context("/apps", method="POST", json={}):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
            app_module.console_ns.payload = {
                "name": "Created App",
                "description": "Summary",
                "mode": "advanced-chat",
            }
            monkeypatch.setattr(
                app_module,
                "AppService",
                lambda: SimpleNamespace(create_app=lambda tenant_id, params, user, session: app_obj),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppPermissions,
                "batch_get",
                lambda tenant_id, account_id, app_ids, session: {"app-new": ["app.acl.view_layout", "app.acl.edit"]},
            )
            initialize_rbac_task = MagicMock()
            monkeypatch.setattr(
                app_module,
                "initialize_created_app_rbac_access_task",
                initialize_rbac_task,
            )
            replace_whitelist = MagicMock()
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "replace_whitelist",
                replace_whitelist,
            )

            resp, status = method(app_module.AppListApi(), MagicMock(), "tenant-1", SimpleNamespace(id="acct-1"))

    assert status == 201
    assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
    assert replace_whitelist.call_args.kwargs["payload"].scope is app_module.RBACResourceWhitelistScope.ALL
    initialize_rbac_task.delay.assert_called_once_with("tenant-1", "acct-1", app_id="app-new")


def test_app_list_api_attaches_permission_keys(app, app_module):
    method = app_module.AppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    app_obj = SimpleNamespace(
        id="app-1",
        name="List App",
        desc_or_prompt="Summary",
        mode_compatible_with_agent="chat",
        mode="chat",
        created_at=_ts(15),
        updated_at=_ts(15),
        permission_keys=[],
    )
    pagination = SimpleNamespace(page=1, per_page=20, total=1, has_next=False, items=[app_obj])
    get_paginate_apps = MagicMock(return_value=pagination)

    with app.test_request_context("/apps"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(
                app_module.AppService,
                "get_paginate_apps",
                get_paginate_apps,
            )
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
                    app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
                        default_permission_keys=["app.preview", "app.acl.view_layout"],
                        overrides=[
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-1",
                                permission_keys=["app.acl.view_layout", "app.acl.edit"],
                            )
                        ],
                    )
                ),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "whitelist_resources",
                lambda tenant_id, account_id: SimpleNamespace(unrestricted=True, resource_ids=[]),
            )

            session = MagicMock()
            session.execute.return_value.scalars.return_value.all.return_value = []
            resp, status = method(app_module.AppListApi(), "tenant-1", "acct-1", session)

    assert status == 200
    params = get_paginate_apps.call_args.args[2]
    assert params.accessible_app_ids is None
    assert params.is_created_by_me is None
    assert resp["data"][0]["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]


def test_recent_app_list_api_returns_only_home_card_fields(app, app_module):
    method = app_module.RecentAppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    recent_app = SimpleNamespace(
        id="app-1",
        name="Recent App",
        icon_type="emoji",
        icon="🚀",
        icon_background="#FFFFFF",
        mode="chat",
        author_name="Recent Author",
        updated_at=_ts(15),
        maintainer="acct-1",
    )
    get_recent_apps = MagicMock(return_value=[recent_app])

    with app.test_request_context("/apps/recent?limit=8"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(dify_config, "RBAC_ENABLED", False)
            monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
                    app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
                        overrides=[
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-1",
                                permission_keys=["app.acl.monitor"],
                            )
                        ]
                    )
                ),
            )

            resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())

    assert status == 200
    assert resp == {
        "data": [
            {
                "id": "app-1",
                "name": "Recent App",
                "icon_type": "emoji",
                "icon": "🚀",
                "icon_background": "#FFFFFF",
                "mode": "chat",
                "author_name": "Recent Author",
                "updated_at": int(_ts(15).timestamp()),
                "permission_keys": ["app.acl.monitor"],
                "maintainer": "acct-1",
                "icon_url": None,
            }
        ]
    }
    params = get_recent_apps.call_args.args[2]
    assert params.limit == 8
    assert "total" not in resp
    assert "description" not in resp["data"][0]
    assert "tags" not in resp["data"][0]
    assert "workflow" not in resp["data"][0]


@pytest.mark.parametrize("mode", ["channel", "rag-pipeline", "agent"])
def test_recent_app_response_rejects_non_home_app_modes(app_module, mode: str) -> None:
    with pytest.raises(ValidationError):
        app_module.RecentAppResponse.model_validate(
            {
                "id": "app-1",
                "name": "Recent App",
                "mode": mode,
                "updated_at": _ts(),
            }
        )


def test_recent_app_list_api_applies_rbac_visibility_filter(app, app_module):
    method = app_module.RecentAppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    get_recent_apps = MagicMock(return_value=[])
    with app.test_request_context("/apps/recent"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
                    workspace=app_module.enterprise_rbac_service.WorkspacePermissionSnapshot(
                        permission_keys=["app.create_and_management"]
                    )
                ),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "whitelist_resources",
                lambda tenant_id, account_id: SimpleNamespace(
                    unrestricted=False,
                    resource_ids=["app-shared"],
                ),
            )

            resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())

    assert status == 200
    assert resp == {"data": []}
    params = get_recent_apps.call_args.args[2]
    assert params.accessible_app_ids == ["app-shared"]
    assert params.include_own_apps is True


def test_app_list_api_limits_to_apps_created_by_current_user_without_view_permission(app, app_module):
    method = app_module.AppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    pagination = SimpleNamespace(page=1, per_page=20, total=0, has_next=False, items=[])
    get_paginate_apps = MagicMock(return_value=pagination)

    with app.test_request_context("/apps"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(app_module.AppService, "get_paginate_apps", get_paginate_apps)
            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
                    workspace=app_module.enterprise_rbac_service.WorkspacePermissionSnapshot(
                        permission_keys=["app.create_and_management"]
                    )
                ),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "whitelist_resources",
                lambda tenant_id, account_id: SimpleNamespace(resource_ids=["app-shared", "app-not-permitted"]),
            )
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )

            session = MagicMock()
            resp, status = method(app_module.AppListApi(), "tenant-1", "acct-1", session)

    assert status == 200
    assert resp["data"] == []
    params = get_paginate_apps.call_args.args[2]
    assert params.accessible_app_ids == ["app-not-permitted", "app-shared"]
    assert params.include_own_apps is True
    assert params.is_created_by_me is None


def test_app_list_api_limits_to_preview_overrides_without_manage_own_permission(app, app_module):
    method = app_module.AppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    pagination = SimpleNamespace(page=1, per_page=20, total=0, has_next=False, items=[])
    get_paginate_apps = MagicMock(return_value=pagination)

    with app.test_request_context("/apps"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(app_module.AppService, "get_paginate_apps", get_paginate_apps)
            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
                    app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
                        overrides=[
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-acl-shared",
                                permission_keys=["app.acl.preview"],
                            ),
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-full",
                                permission_keys=["app.full_access"],
                            ),
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-shared",
                                permission_keys=["app.preview"],
                            ),
                        ]
                    )
                ),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "whitelist_resources",
                lambda tenant_id, account_id: SimpleNamespace(
                    resource_ids=["app-shared", "app-acl-shared", "app-full", "app-whitelist-only"]
                ),
            )
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )

            session = MagicMock()
            method(app_module.AppListApi(), "tenant-1", "acct-1", session)

    params = get_paginate_apps.call_args.args[2]
    assert params.accessible_app_ids == ["app-acl-shared", "app-full", "app-shared", "app-whitelist-only"]
    assert params.include_own_apps is False
    assert params.is_created_by_me is None


def test_app_list_api_returns_no_apps_without_workspace_or_resource_view_permission(app, app_module):
    method = app_module.AppListApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    pagination = SimpleNamespace(page=1, per_page=20, total=0, has_next=False, items=[])
    get_paginate_apps = MagicMock(return_value=pagination)

    with app.test_request_context("/apps"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(app_module.AppService, "get_paginate_apps", get_paginate_apps)
            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(),
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppAccess,
                "whitelist_resources",
                lambda tenant_id, account_id: SimpleNamespace(resource_ids=["app-not-permitted"]),
            )
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )

            session = MagicMock()
            method(app_module.AppListApi(), "tenant-1", "acct-1", session)

    params = get_paginate_apps.call_args.args[2]
    assert params.accessible_app_ids == ["app-not-permitted"]
    assert params.include_own_apps is False
    assert params.is_created_by_me is None


def test_app_detail_api_attaches_current_user_permission_keys(app, app_module):
    method = app_module.AppApi.get
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    app_obj = SimpleNamespace(
        id="app-1",
        name="Detail App",
        description="Summary",
        mode_compatible_with_agent="chat",
        enable_site=True,
        enable_api=True,
        permission_keys=[],
    )

    with app.test_request_context("/apps/app-1"):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
            get_app = MagicMock(return_value=app_obj)
            monkeypatch.setattr(app_module, "AppService", lambda: SimpleNamespace(get_app=get_app))
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )
            get_permissions = MagicMock(
                return_value=app_module.enterprise_rbac_service.MyPermissionsResponse(
                    app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
                        overrides=[
                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
                                resource_id="app-1",
                                permission_keys=["app.acl.view_layout", "app.acl.edit", "app.acl.monitor"],
                            )
                        ]
                    )
                )
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.MyPermissions,
                "get",
                get_permissions,
            )

            session = MagicMock()
            resp = method(
                app_module.AppApi(),
                session,
                "tenant-1",
                SimpleNamespace(id="acct-1"),
                app_model=app_obj,
            )

    get_app.assert_called_once_with(app_obj, session=session)
    get_permissions.assert_called_once_with("tenant-1", "acct-1", app_id="app-1", session=session)
    assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit", "app.acl.monitor"]


def test_app_copy_api_attaches_permission_keys(app, app_module):
    method = app_module.AppCopyApi.post
    while hasattr(method, "__wrapped__"):
        method = method.__wrapped__

    app_obj = SimpleNamespace(
        id="app-new",
        name="Copied App",
        description="Summary",
        mode_compatible_with_agent="workflow",
        enable_site=True,
        enable_api=True,
        permission_keys=[],
    )

    import_result = SimpleNamespace(status=app_module.ImportStatus.COMPLETED, app_id="app-new")
    fake_session = MagicMock()
    fake_session.__enter__.return_value = fake_session
    fake_session.__exit__.return_value = None
    fake_session.scalar.return_value = app_obj

    with app.test_request_context("/apps/app-original/copy", method="POST", json={}):
        with pytest.MonkeyPatch.context() as monkeypatch:
            monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
            monkeypatch.setattr(
                app_module,
                "AppDslService",
                lambda *_args, **_kwargs: SimpleNamespace(
                    export_dsl=lambda **_kwargs: "dsl",
                    import_app=lambda **_kwargs: import_result,
                ),
            )
            monkeypatch.setattr(
                app_module.FeatureService,
                "get_system_features",
                lambda: SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False)),
            )
            monkeypatch.setattr(app_module, "db", SimpleNamespace(engine=object(), session=lambda: MagicMock()))
            monkeypatch.setattr(
                app_module,
                "Session",
                lambda *_args, **_kwargs: fake_session,
            )
            monkeypatch.setattr(
                app_module.enterprise_rbac_service.RBACService.AppPermissions,
                "batch_get",
                lambda tenant_id, account_id, app_ids, session: {"app-new": ["app.acl.view_layout", "app.acl.edit"]},
            )

            resp, status = method(
                app_module.AppCopyApi(),
                "tenant-1",
                SimpleNamespace(id="acct-1"),
                app_model=SimpleNamespace(id="app-original"),
            )

    assert status == 201
    assert fake_session.scalar.called
    assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
