from unittest.mock import AsyncMock, MagicMock, patch
from uuid import UUID, uuid4

import pytest
from fastapi import HTTPException
from langflow.helpers.flow import (
    get_flow_by_id_or_endpoint_name,
    get_flow_by_id_or_name,
    list_flows,
    list_flows_by_flow_folder,
    list_flows_by_folder_id,
)
from langflow.schema.data import Data
from langflow.services.database.models.flow.model import Flow


class TestListFlows:
    """Test list_flows function in backend."""

    @pytest.mark.asyncio
    async def test_list_flows_raises_error_without_user_id(self):
        """Test that list_flows raises ValueError without user_id."""
        with pytest.raises(ValueError, match="Session is invalid"):
            await list_flows(user_id=None)

    @pytest.mark.asyncio
    async def test_list_flows_queries_database(self):
        """Test that list_flows queries database correctly."""
        user_id = str(uuid4())

        mock_flow1 = MagicMock(spec=Flow)
        mock_flow1.to_data = MagicMock(return_value=Data(data={"name": "Flow 1"}))

        mock_flow2 = MagicMock(spec=Flow)
        mock_flow2.to_data = MagicMock(return_value=Data(data={"name": "Flow 2"}))

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.all = MagicMock(return_value=[mock_flow1, mock_flow2])
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await list_flows(user_id=user_id)

            assert len(result) == 2
            assert result[0].data["name"] == "Flow 1"
            assert result[1].data["name"] == "Flow 2"


class TestListFlowsByFlowFolder:
    """Test list_flows_by_flow_folder function in backend."""

    @pytest.mark.asyncio
    async def test_list_flows_by_flow_folder_raises_error_without_user_id(self):
        """Test that function raises ValueError without user_id."""
        with pytest.raises(ValueError, match="Session is invalid"):
            await list_flows_by_flow_folder(user_id=None, flow_id=str(uuid4()))

    @pytest.mark.asyncio
    async def test_list_flows_by_flow_folder_raises_error_without_flow_id(self):
        """Test that function raises ValueError without flow_id."""
        with pytest.raises(ValueError, match="Flow ID is required"):
            await list_flows_by_flow_folder(user_id=str(uuid4()), flow_id=None)

    @pytest.mark.asyncio
    async def test_list_flows_by_flow_folder_queries_same_folder(self):
        """Test that function queries flows in same folder."""
        user_id = str(uuid4())
        flow_id = str(uuid4())

        # Mock database results
        mock_row1 = MagicMock()
        mock_row1._mapping = {"id": str(uuid4()), "name": "Flow 1", "updated_at": "2024-01-01"}

        mock_row2 = MagicMock()
        mock_row2._mapping = {"id": str(uuid4()), "name": "Flow 2", "updated_at": "2024-01-02"}

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.all = MagicMock(return_value=[mock_row1, mock_row2])
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await list_flows_by_flow_folder(user_id=user_id, flow_id=flow_id)

            assert len(result) == 2
            assert isinstance(result[0], Data)
            assert result[0].data["name"] == "Flow 1"

    @pytest.mark.asyncio
    async def test_list_flows_by_flow_folder_respects_order_params(self):
        """Test that function respects ordering parameters."""
        user_id = str(uuid4())
        flow_id = str(uuid4())
        order_params = {"column": "name", "direction": "asc"}

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.all = MagicMock(return_value=[])
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await list_flows_by_flow_folder(user_id=user_id, flow_id=flow_id, order_params=order_params)

            # Verify query was executed (result should be empty list from mock)
            assert result == []


class TestListFlowsByFolderId:
    """Test list_flows_by_folder_id function in backend."""

    @pytest.mark.asyncio
    async def test_list_flows_by_folder_id_raises_error_without_user_id(self):
        """Test that function raises ValueError without user_id."""
        with pytest.raises(ValueError, match="Session is invalid"):
            await list_flows_by_folder_id(user_id=None, folder_id=str(uuid4()))

    @pytest.mark.asyncio
    async def test_list_flows_by_folder_id_raises_error_without_folder_id(self):
        """Test that function raises ValueError without folder_id."""
        with pytest.raises(ValueError, match="Folder ID is required"):
            await list_flows_by_folder_id(user_id=str(uuid4()), folder_id=None)

    @pytest.mark.asyncio
    async def test_list_flows_by_folder_id_queries_database(self):
        """Test that function queries database for flows in folder."""
        user_id = str(uuid4())
        folder_id = str(uuid4())

        mock_row1 = MagicMock()
        mock_row1._mapping = {"id": str(uuid4()), "name": "Flow 1"}

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.all = MagicMock(return_value=[mock_row1])
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await list_flows_by_folder_id(user_id=user_id, folder_id=folder_id)

            assert len(result) == 1
            assert isinstance(result[0], Data)


class TestGetFlowByIdOrName:
    """Test get_flow_by_id_or_name function in backend."""

    @pytest.mark.asyncio
    async def test_get_flow_by_id_or_name_raises_error_without_user_id(self):
        """Test that function raises ValueError without user_id."""
        with pytest.raises(ValueError, match="Session is invalid"):
            await get_flow_by_id_or_name(user_id="", flow_id=str(uuid4()))

    @pytest.mark.asyncio
    async def test_get_flow_by_id_or_name_raises_error_without_id_or_name(self):
        """Test that function raises ValueError without flow_id or flow_name."""
        with pytest.raises(ValueError, match="Flow ID or Flow Name is required"):
            await get_flow_by_id_or_name(user_id=str(uuid4()))

    @pytest.mark.asyncio
    async def test_get_flow_by_id_or_name_queries_by_id(self):
        """Test that function queries database by flow ID."""
        user_id = str(uuid4())
        flow_id = str(uuid4())

        mock_flow = MagicMock(spec=Flow)
        mock_flow.to_data = MagicMock(return_value=Data(data={"name": "Test Flow"}))

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.first = MagicMock(return_value=mock_flow)
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await get_flow_by_id_or_name(user_id=user_id, flow_id=flow_id)

            assert isinstance(result, Data)
            assert result.data["name"] == "Test Flow"

    @pytest.mark.asyncio
    async def test_get_flow_by_id_or_name_queries_by_name(self):
        """Test that function queries database by flow name."""
        user_id = str(uuid4())
        flow_name = "Test Flow"

        mock_flow = MagicMock(spec=Flow)
        mock_flow.to_data = MagicMock(return_value=Data(data={"name": flow_name}))

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.first = MagicMock(return_value=mock_flow)
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await get_flow_by_id_or_name(user_id=user_id, flow_name=flow_name)

            assert isinstance(result, Data)
            assert result.data["name"] == flow_name

    @pytest.mark.asyncio
    async def test_get_flow_by_id_or_name_prefers_id_over_name(self):
        """Test that function prefers flow_id when both are provided."""
        user_id = str(uuid4())
        flow_id = str(uuid4())
        flow_name = "Test Flow"

        mock_flow = MagicMock(spec=Flow)
        mock_flow.to_data = MagicMock(return_value=Data(data={"id": flow_id, "name": flow_name}))

        with patch("langflow.helpers.flow.session_scope") as mock_session_scope:
            mock_session = MagicMock()
            mock_result = MagicMock()
            mock_result.first = MagicMock(return_value=mock_flow)
            mock_session.exec = AsyncMock(return_value=mock_result)
            mock_session_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
            mock_session_scope.return_value.__aexit__ = AsyncMock()

            result = await get_flow_by_id_or_name(user_id=user_id, flow_id=flow_id, flow_name=flow_name)

            assert isinstance(result, Data)
            # The query should have been made with flow_id (checking it was called)
            mock_session.exec.assert_called_once()


class TestGetFlowByIdOrEndpointName:
    """Regression tests for the IDOR fix in get_flow_by_id_or_endpoint_name.

    The UUID branch previously called ``session.get(Flow, flow_id)`` without
    applying any ownership check, so any authenticated caller could resolve
    another user's flow by UUID.  The endpoint_name branch had the matching
    issue when ``user_id`` was not supplied (the FastAPI ``Depends`` pattern
    provided no user_id by default).  These tests lock in the fix.
    """

    @staticmethod
    def _patch_session(mock_session):
        """Patch session_scope to yield the provided mock session."""
        patcher = patch("langflow.helpers.flow.session_scope")
        mock_scope = patcher.start()
        mock_scope.return_value.__aenter__ = AsyncMock(return_value=mock_session)
        mock_scope.return_value.__aexit__ = AsyncMock(return_value=False)
        return patcher

    @pytest.mark.asyncio
    async def test_uuid_branch_returns_flow_for_owner(self):
        """Same-user UUID lookup returns the flow (happy path)."""
        owner_id = uuid4()
        flow_id = uuid4()
        flow = MagicMock(spec=Flow)
        flow.id = flow_id
        flow.user_id = owner_id
        flow.name = "test_flow"
        flow.endpoint_name = None
        flow.data = {}
        flow.description = None
        flow.is_component = False
        flow.updated_at = None
        flow.folder_id = None
        flow.user_id = owner_id

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=flow)

        patcher = self._patch_session(mock_session)
        try:
            with patch("langflow.helpers.flow.FlowRead") as mock_flow_read:
                mock_flow_read.model_validate = MagicMock(return_value="validated_flow")
                result = await get_flow_by_id_or_endpoint_name(str(flow_id), str(owner_id))
                assert result == "validated_flow"
                mock_session.get.assert_awaited_once_with(Flow, flow_id)
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_uuid_branch_rejects_cross_user_access(self):
        """Cross-user UUID lookup raises 404 (the core IDOR fix)."""
        attacker_id = uuid4()
        victim_id = uuid4()
        flow_id = uuid4()
        victim_flow = MagicMock(spec=Flow)
        victim_flow.id = flow_id
        victim_flow.user_id = victim_id

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=victim_flow)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException) as exc_info:
                await get_flow_by_id_or_endpoint_name(str(flow_id), str(attacker_id))
            assert exc_info.value.status_code == 404
            # Must NOT 403 -- returning 403 would confirm the flow exists to
            # an attacker enumerating UUIDs.  404 hides existence.
            assert "not found" in exc_info.value.detail.lower()
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_uuid_branch_accepts_uuid_instance_for_user_id(self):
        """Callers passing a UUID object (not str) are handled correctly."""
        attacker_id = uuid4()
        victim_id = uuid4()
        flow_id = uuid4()
        victim_flow = MagicMock(spec=Flow)
        victim_flow.id = flow_id
        victim_flow.user_id = victim_id

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=victim_flow)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException) as exc_info:
                # Pass UUID, not str -- matches workflow.py:151 call style.
                await get_flow_by_id_or_endpoint_name(str(flow_id), attacker_id)
            assert exc_info.value.status_code == 404
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_uuid_branch_without_user_id_returns_flow(self):
        """No user_id = no scoping (preserves webhook/internal caller behavior)."""
        flow_id = uuid4()
        flow = MagicMock(spec=Flow)
        flow.id = flow_id
        flow.user_id = uuid4()

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=flow)

        patcher = self._patch_session(mock_session)
        try:
            with patch("langflow.helpers.flow.FlowRead") as mock_flow_read:
                mock_flow_read.model_validate = MagicMock(return_value="validated_flow")
                # user_id=None: any flow is returned.
                result = await get_flow_by_id_or_endpoint_name(str(flow_id), user_id=None)
                assert result == "validated_flow"
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_uuid_branch_missing_flow_raises_404(self):
        """Non-existent UUID returns 404 regardless of user_id."""
        owner_id = uuid4()
        flow_id = uuid4()

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=None)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException) as exc_info:
                await get_flow_by_id_or_endpoint_name(str(flow_id), str(owner_id))
            assert exc_info.value.status_code == 404
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_endpoint_name_branch_scopes_by_user(self):
        """endpoint_name lookup filters by user_id when supplied."""
        owner_id = uuid4()

        mock_session = MagicMock()
        mock_result = MagicMock()
        mock_result.first = MagicMock(return_value=None)
        mock_session.exec = AsyncMock(return_value=mock_result)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException) as exc_info:
                await get_flow_by_id_or_endpoint_name("my-webhook", str(owner_id))
            assert exc_info.value.status_code == 404
            # The select statement should have been filtered by user_id --
            # verified indirectly by the exec call count (1 call, scoped).
            mock_session.exec.assert_awaited_once()
            stmt = mock_session.exec.await_args.args[0]
            # Compile the where clauses and confirm the user_id filter is present.
            where_sql = str(stmt.whereclause)
            assert "user_id" in where_sql

        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_endpoint_name_branch_without_user_id_not_scoped(self):
        """No user_id on endpoint_name branch = no user scoping (webhook behavior)."""
        flow = MagicMock(spec=Flow)
        flow.id = uuid4()
        flow.user_id = uuid4()

        mock_session = MagicMock()
        mock_result = MagicMock()
        mock_result.first = MagicMock(return_value=flow)
        mock_session.exec = AsyncMock(return_value=mock_result)

        patcher = self._patch_session(mock_session)
        try:
            with patch("langflow.helpers.flow.FlowRead") as mock_flow_read:
                mock_flow_read.model_validate = MagicMock(return_value="validated_flow")
                result = await get_flow_by_id_or_endpoint_name("webhook-ep", user_id=None)
                assert result == "validated_flow"
                stmt = mock_session.exec.await_args.args[0]
                where_sql = str(stmt.whereclause) if stmt.whereclause is not None else ""
                # With no user_id, the only filter should be endpoint_name.
                assert "user_id" not in where_sql
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_accepts_uuid_instance_for_user_id_parameter(self):
        """user_id can be a UUID object (matches workflow.py call sites)."""
        owner_id = UUID(str(uuid4()))

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=None)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException):
                await get_flow_by_id_or_endpoint_name(str(uuid4()), owner_id)
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_widening_is_opt_in_even_when_authz_active(self):
        """Cross-user UUID lookup stays 404 unless caller explicitly opts in.

        Regression: the share-aware widening must never fire from the helper's
        default arguments — agentic MCP tools call the helper without
        ``widen_for_shares=True`` and then read ``flow.data`` directly, so an
        always-on widen would leak another user's graph metadata before any
        policy decision runs.
        """
        attacker_id = uuid4()
        victim_id = uuid4()
        flow_id = uuid4()
        victim_flow = MagicMock(spec=Flow)
        victim_flow.id = flow_id
        victim_flow.user_id = victim_id

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=victim_flow)

        # Stub authz so cross-user fetch + AUTHZ_ENABLED are both ON, which
        # under the old contract would have widened the lookup. With the new
        # opt-in flag default-false, widening must NOT happen.
        stub_authz = MagicMock()
        stub_authz.supports_cross_user_fetch = AsyncMock(return_value=True)
        stub_authz.is_enabled = AsyncMock(return_value=True)

        patcher = self._patch_session(mock_session)
        try:
            with patch("langflow.services.deps.get_authorization_service", return_value=stub_authz):  # noqa: SIM117
                with pytest.raises(HTTPException) as exc_info:
                    await get_flow_by_id_or_endpoint_name(str(flow_id), str(attacker_id))
            assert exc_info.value.status_code == 404
        finally:
            patcher.stop()

    @pytest.mark.asyncio
    async def test_widening_allows_cross_user_when_opt_in_and_authz_active(self):
        """When the caller opts in AND authz is active, share-aware lookup widens.

        This is the path used by API routes that immediately call
        ``ensure_flow_permission`` afterward — widening is safe there because
        a permission check follows the fetch.
        """
        attacker_id = uuid4()
        victim_id = uuid4()
        flow_id = uuid4()
        victim_flow = MagicMock(spec=Flow)
        victim_flow.id = flow_id
        victim_flow.user_id = victim_id

        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=victim_flow)

        stub_authz = MagicMock()
        stub_authz.supports_cross_user_fetch = AsyncMock(return_value=True)
        stub_authz.is_enabled = AsyncMock(return_value=True)

        patcher = self._patch_session(mock_session)
        try:
            with patch("langflow.services.deps.get_authorization_service", return_value=stub_authz):  # noqa: SIM117
                with patch("langflow.helpers.flow.FlowRead") as mock_flow_read:
                    mock_flow_read.model_validate = MagicMock(return_value="validated_flow")
                    result = await get_flow_by_id_or_endpoint_name(
                        str(flow_id),
                        str(attacker_id),
                        widen_for_shares=True,
                    )
                    assert result == "validated_flow"
        finally:
            patcher.stop()

    @pytest.mark.parametrize("bad_user_id", ["not-a-uuid", "", "12345-not-a-real-uuid"])
    @pytest.mark.asyncio
    async def test_malformed_user_id_raises_404_not_500(self, bad_user_id):
        """Malformed user_id (e.g. ``?user_id=foo``) must fail closed with 404.

        Previously the eager ``UUID(user_id)`` normalization raised a raw
        ``ValueError`` that surfaced as a 500 to the client.  The helper now
        converts that into the same 404 we'd return for "flow not found" so
        that a malformed caller identity can never be used to enumerate or
        exfiltrate flows, and doesn't give attackers a new error signal.
        """
        mock_session = MagicMock()
        mock_session.get = AsyncMock(return_value=None)

        patcher = self._patch_session(mock_session)
        try:
            with pytest.raises(HTTPException) as exc_info:
                await get_flow_by_id_or_endpoint_name(str(uuid4()), bad_user_id)
            assert exc_info.value.status_code == 404
            # Session lookup should never happen if user_id couldn't be parsed.
            mock_session.get.assert_not_awaited()
        finally:
            patcher.stop()
