"""Route-handler-level tests for deployment endpoints.

Covers the integration behaviour of the FastAPI route handlers:
- Rollback on commit failure (create / update)
- GET single-deployment synchronization (deployment-level and snapshot-level)
- Project-scoped flow-version validation (create / update)
- resolve_deployment_adapter edge cases
"""

from __future__ import annotations

from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import ANY, AsyncMock, MagicMock, patch
from uuid import uuid4

import pytest
from fastapi import HTTPException
from langflow.api.v1.deployments import DeploymentTelemetryCtx
from langflow.api.v1.mappers.deployments.contracts import ProviderSnapshotBinding
from langflow.api.v1.schemas.deployments import (
    DeploymentLlmListResponse,
    DeploymentProviderAccountCreateRequest,
    DeploymentProviderAccountUpdateRequest,
    DeploymentUpdateRequest,
    SnapshotUpdateRequest,
)
from langflow.services.database.models.deployment.exceptions import DeploymentGuardError
from langflow.services.database.models.deployment_provider_account.schemas import DeploymentProviderKey
from lfx.services.adapters.deployment.exceptions import (
    AuthenticationError,
    DeploymentNotFoundError,
    ResourceConflictError,
    ServiceUnavailableError,
)
from lfx.services.adapters.deployment.schema import (
    ConfigListItem,
    ConfigListResult,
    DeploymentCreateResult,
    DeploymentListParams,
    DeploymentListResult,
    DeploymentUpdateResult,
    ItemResult,
    SnapshotItem,
    SnapshotListResult,
)

ROUTES_MODULE = "langflow.api.v1.deployments"
HELPERS_MODULE = "langflow.api.v1.mappers.deployments.helpers"


# ---------------------------------------------------------------------------
# Shared factories
# ---------------------------------------------------------------------------


def _fake_provider_account(
    *,
    provider_key: str = DeploymentProviderKey.WATSONX_ORCHESTRATE,
    provider_url: str = "https://api.us-south.wxo.cloud.ibm.com/instances/tenant-1",
    api_key: str = "encrypted-key",
    provider_tenant_id: str | None = "tenant-1",
) -> SimpleNamespace:
    return SimpleNamespace(
        id=uuid4(),
        provider_key=provider_key,
        provider_url=provider_url,
        api_key=api_key,
        provider_tenant_id=provider_tenant_id,
    )


def _fake_deployment_row(**overrides) -> SimpleNamespace:
    return SimpleNamespace(
        id=overrides.get("id", uuid4()),
        resource_key=overrides.get("resource_key", "rk-1"),
        name=overrides.get("name", "test-deployment"),
        description=overrides.get("description"),
        deployment_type=overrides.get("deployment_type", "agent"),
        deployment_provider_account_id=overrides.get("deployment_provider_account_id", uuid4()),
        project_id=overrides.get("project_id", uuid4()),
        created_at=overrides.get("created_at"),
        updated_at=overrides.get("updated_at"),
    )


def _fake_user() -> SimpleNamespace:
    return SimpleNamespace(id=uuid4())


def _fake_telemetry() -> DeploymentTelemetryCtx:
    return DeploymentTelemetryCtx()


def _fake_attachment(*, provider_snapshot_id: str | None = None) -> SimpleNamespace:
    return SimpleNamespace(
        flow_version_id=uuid4(),
        provider_snapshot_id=provider_snapshot_id,
        deployment_id=uuid4(),
    )


class _AsyncNoopSavepoint:
    async def __aenter__(self):
        return None

    async def __aexit__(self, exc_type, exc, tb):
        return False


# ---------------------------------------------------------------------------
# query parameter validators
# ---------------------------------------------------------------------------


class TestQueryParameterValidators:
    def test_dedupe_snapshot_names_preserves_order(self):
        from langflow.api.v1.deployments import _dedupe_snapshot_names

        assert _dedupe_snapshot_names(["my_tool", "other_tool", "my_tool", "third_tool"]) == [
            "my_tool",
            "other_tool",
            "third_tool",
        ]

    def test_dedupe_snapshot_names_keeps_none(self):
        from langflow.api.v1.deployments import _dedupe_snapshot_names

        assert _dedupe_snapshot_names(None) is None


# ---------------------------------------------------------------------------
# create_deployment: rollback on commit failure
# ---------------------------------------------------------------------------


class TestCreateDeploymentRollback:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.rollback_provider_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_snapshot_map_for_create", return_value={})
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_rollback_called_on_commit_failure(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,  # noqa: ARG002
        mock_create_db,
        mock_resolve_snap,  # noqa: ARG002
        mock_attach,  # noqa: ARG002
        mock_rollback,
    ):
        """When session.commit() fails, rollback_provider_create is called."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        create_result = DeploymentCreateResult(
            id="provider-dep-1",
            provider_result={"app_ids": ["cfg-1"], "tools_with_refs": [{"tool_id": "tool-1", "source_ref": "fv-1"}]},
        )
        adapter.create.return_value = create_result
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_create_flow_version_ids.return_value = []
        mapper.util_existing_deployment_resource_key_for_create.return_value = None
        mapper.resolve_deployment_create = AsyncMock(return_value=MagicMock())
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        mock_create_db.return_value = _fake_deployment_row()

        session = AsyncMock()
        session.commit.side_effect = RuntimeError("DB commit failed")

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "test"
        payload.type = "agent"
        payload.description = None

        with pytest.raises(RuntimeError, match="DB commit failed"):
            await create_deployment(
                session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        mock_rollback.assert_awaited_once()
        assert mock_rollback.call_args.kwargs["resource_id"] == create_result.id
        assert mock_rollback.call_args.kwargs["provider_result"] == create_result.provider_result

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.rollback_provider_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_snapshot_map_for_create", return_value={})
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_no_rollback_on_success(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,  # noqa: ARG002
        mock_create_db,
        mock_resolve_snap,  # noqa: ARG002
        mock_attach,  # noqa: ARG002
        mock_rollback,
    ):
        """On successful commit, rollback is NOT called."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        create_result = DeploymentCreateResult(id="provider-dep-1")
        adapter.create.return_value = create_result
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_create_flow_version_ids.return_value = []
        mapper.util_existing_deployment_resource_key_for_create.return_value = None
        mapper.resolve_deployment_create = AsyncMock(return_value=MagicMock())
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        dep_row = _fake_deployment_row()
        mock_create_db.return_value = dep_row

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "test"
        payload.type = "agent"
        payload.description = None

        mapper.shape_deployment_create_result.return_value = MagicMock()
        await create_deployment(
            session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
        )

        mock_rollback.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.rollback_provider_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_snapshot_map_for_create", return_value={})
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_deployment_by_resource_key", new_callable=AsyncMock, return_value=None)
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_existing_agent_mutation_commit_failure_uses_non_delete_rollback(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_by_resource_key,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,  # noqa: ARG002
        mock_create_db,
        mock_resolve_snap,  # noqa: ARG002
        mock_attach,  # noqa: ARG002
        mock_rollback,
    ):
        """Commit failure after existing-agent mutation triggers rollback without delete fallback."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.update.return_value = DeploymentUpdateResult(id="existing-agent-1", provider_result={"ok": True})
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_existing_deployment_resource_key_for_create.return_value = "existing-agent-1"
        mapper.util_should_mutate_provider_for_existing_deployment_create.return_value = True
        mapper.util_create_flow_version_ids.return_value = []
        mapper.resolve_deployment_update_for_existing_create = AsyncMock(return_value=MagicMock())
        mapper.util_create_result_from_existing_update.return_value = DeploymentCreateResult(
            id="existing-agent-1",
            provider_result={"app_ids": ["app-1"]},
        )
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        mock_create_db.return_value = _fake_deployment_row(resource_key="existing-agent-1")

        session = AsyncMock()
        session.commit.side_effect = RuntimeError("DB commit failed")

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "existing"
        payload.type = "agent"
        payload.description = "desc"

        with pytest.raises(RuntimeError, match="DB commit failed"):
            await create_deployment(
                session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        mock_rollback.assert_awaited_once()
        assert mock_rollback.call_args.kwargs["resource_id"] == "existing-agent-1"
        assert mock_rollback.call_args.kwargs["allow_delete_fallback"] is False


# ---------------------------------------------------------------------------
# create_deployment: existing_agent_id behavior
# ---------------------------------------------------------------------------


class TestCreateDeploymentExistingAgent:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_deployment_by_resource_key", new_callable=AsyncMock, return_value=None)
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_existing_agent_without_mutation_skips_provider_calls(
        self,
        mock_get_pa,
        mock_name_exists,
        mock_get_by_resource_key,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
        mock_create_db,
        mock_attach,
    ):
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_existing_deployment_resource_key_for_create.return_value = "existing-agent-1"
        mapper.util_should_mutate_provider_for_existing_deployment_create.return_value = False
        mapper.util_create_flow_version_ids.return_value = []
        mapper.util_create_result_from_existing_resource.return_value = DeploymentCreateResult(id="existing-agent-1")
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        dep_row = _fake_deployment_row(resource_key="existing-agent-1")
        mock_create_db.return_value = dep_row

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "existing"
        payload.type = "agent"
        payload.description = None

        mapper.shape_deployment_create_result.return_value = MagicMock()
        await create_deployment(
            session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
        )

        _ = (mock_name_exists, mock_get_by_resource_key, mock_validate_fv, mock_attach)
        adapter.create.assert_not_awaited()
        adapter.update.assert_not_awaited()
        mapper.util_create_result_from_existing_resource.assert_called_once_with(
            existing_resource_key="existing-agent-1"
        )
        assert mock_create_db.call_args.kwargs["resource_key"] == "existing-agent-1"

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_deployment_by_resource_key", new_callable=AsyncMock, return_value=None)
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_existing_agent_with_mutation_uses_provider_update(
        self,
        mock_get_pa,
        mock_name_exists,
        mock_get_by_resource_key,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
        mock_create_db,
        mock_attach,
    ):
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.update.return_value = DeploymentUpdateResult(id="existing-agent-1", provider_result={"ok": True})
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_existing_deployment_resource_key_for_create.return_value = "existing-agent-1"
        mapper.util_should_mutate_provider_for_existing_deployment_create.return_value = True
        mapper.util_create_flow_version_ids.return_value = []
        mapper.resolve_deployment_update_for_existing_create = AsyncMock(
            return_value=MagicMock(provider_data={"upsert_flows": []})
        )
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        dep_row = _fake_deployment_row(resource_key="existing-agent-1")
        mock_create_db.return_value = dep_row

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "existing"
        payload.type = "agent"
        payload.description = "desc"

        mapper.shape_deployment_create_result.return_value = MagicMock()
        await create_deployment(
            session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
        )

        _ = (mock_name_exists, mock_get_by_resource_key, mock_validate_fv, mock_attach)
        adapter.create.assert_not_awaited()
        adapter.update.assert_awaited_once()
        mapper.resolve_deployment_update_for_existing_create.assert_awaited_once()
        assert adapter.update.call_args.kwargs["deployment_id"] == "existing-agent-1"

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_deployment_by_resource_key", new_callable=AsyncMock, return_value=None)
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_existing_agent_with_mutation_uses_mapper_create_result_normalizer(
        self,
        mock_get_pa,
        mock_name_exists,
        mock_get_by_resource_key,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
        mock_create_db,
        mock_attach,
    ):
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.update.return_value = DeploymentUpdateResult(
            id="existing-agent-1",
            provider_result={"ok": True},
        )
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_existing_deployment_resource_key_for_create.return_value = "existing-agent-1"
        mapper.util_should_mutate_provider_for_existing_deployment_create.return_value = True
        mapper.util_create_flow_version_ids.return_value = []
        mapper.resolve_deployment_update_for_existing_create = AsyncMock(
            return_value=MagicMock(provider_data={"upsert_flows": []})
        )
        mapped_create_result = DeploymentCreateResult(id="existing-agent-1", provider_result={"ok": True})
        mapper.util_create_result_from_existing_update.return_value = mapped_create_result
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()
        dep_row = _fake_deployment_row(resource_key="existing-agent-1")
        mock_create_db.return_value = dep_row

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "existing"
        payload.type = "agent"
        payload.description = "desc"

        mapper.shape_deployment_create_result.return_value = MagicMock()
        await create_deployment(
            session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
        )

        _ = (mock_name_exists, mock_get_by_resource_key, mock_validate_fv, mock_attach)
        adapter.create.assert_not_awaited()
        adapter.update.assert_awaited_once()
        mapper.resolve_deployment_update_for_existing_create.assert_awaited_once()
        assert adapter.update.call_args.kwargs["deployment_id"] == "existing-agent-1"
        mapper.util_create_result_from_existing_update.assert_called_once_with(
            existing_resource_key="existing-agent-1",
            result=adapter.update.return_value,
        )
        mapper.shape_deployment_create_result.assert_called_once_with(
            mapped_create_result,
            dep_row,
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
        )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_deployment_by_resource_key", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_existing_agent_conflict_when_resource_key_already_persisted(
        self,
        mock_get_pa,
        mock_name_exists,
        mock_get_by_resource_key,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_create_db,
    ):
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        mock_get_by_resource_key.return_value = _fake_deployment_row(resource_key="existing-agent-1")
        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_existing_deployment_resource_key_for_create.return_value = "existing-agent-1"
        mock_get_mapper.return_value = mapper

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "existing"
        payload.type = "agent"
        payload.description = None

        with pytest.raises(HTTPException) as exc_info:
            await create_deployment(
                session=AsyncMock(), payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 409
        _ = mock_name_exists
        mock_create_db.assert_not_awaited()
        adapter.create.assert_not_awaited()
        adapter.update.assert_not_awaited()


# ---------------------------------------------------------------------------
# list_deployments: provider passthrough mode
# ---------------------------------------------------------------------------


class TestListDeploymentsLoadFromProvider:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_load_from_provider_bypasses_db_view(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_list_synced,
    ):
        from langflow.api.v1.deployments import list_deployments

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.list.return_value = DeploymentListResult(
            deployments=[ItemResult(id="agent-1", name="Agent 1", type="agent")]
        )
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        expected = MagicMock()
        mapper.shape_deployment_list_result.return_value = expected
        mapper.resolve_load_from_provider_deployment_list_params.return_value = {"environment": "draft"}
        mock_get_mapper.return_value = mapper

        result = await list_deployments(
            provider_id=pa.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            params=SimpleNamespace(page=2, size=50),
            deployment_type=None,
            load_from_provider=True,
            flow_version_ids=None,
        )

        assert result is expected
        mock_list_synced.assert_not_awaited()
        adapter.list.assert_awaited_once()
        list_call_kwargs = adapter.list.await_args.kwargs
        assert isinstance(list_call_kwargs["params"], DeploymentListParams)
        assert list_call_kwargs["params"].provider_params == {"environment": "draft"}
        mapper.shape_deployment_list_result.assert_called_once()

    @pytest.mark.asyncio
    async def test_load_from_provider_rejects_flow_version_filters(self):
        from langflow.api.v1.deployments import list_deployments

        with pytest.raises(HTTPException) as exc_info:
            await list_deployments(
                provider_id=uuid4(),
                session=AsyncMock(),
                current_user=_fake_user(),
                params=SimpleNamespace(page=1, size=20),
                deployment_type=None,
                load_from_provider=True,
                flow_version_ids=[str(uuid4())],
            )
        assert exc_info.value.status_code == 422

    @pytest.mark.asyncio
    async def test_load_from_provider_rejects_flow_ids_filter(self):
        from langflow.api.v1.deployments import list_deployments

        with pytest.raises(HTTPException) as exc_info:
            await list_deployments(
                provider_id=uuid4(),
                session=AsyncMock(),
                current_user=_fake_user(),
                params=SimpleNamespace(page=1, size=20),
                deployment_type=None,
                load_from_provider=True,
                flow_ids=[str(uuid4())],
            )
        assert exc_info.value.status_code == 422


# ---------------------------------------------------------------------------
# list_deployments: flow_ids filter
# ---------------------------------------------------------------------------


class TestListDeploymentsFlowIdsFilter:
    @pytest.mark.asyncio
    async def test_flow_ids_and_flow_version_ids_mutually_exclusive(self):
        from langflow.api.v1.deployments import list_deployments

        with pytest.raises(HTTPException) as exc_info:
            await list_deployments(
                provider_id=uuid4(),
                session=AsyncMock(),
                current_user=_fake_user(),
                params=SimpleNamespace(page=1, size=20),
                deployment_type=None,
                load_from_provider=False,
                flow_version_ids=[str(uuid4())],
                flow_ids=[str(uuid4())],
            )
        assert exc_info.value.status_code == 422
        assert "mutually exclusive" in exc_info.value.detail

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.flow_version_ids_for_flows", new_callable=AsyncMock, return_value=[])
    async def test_flow_ids_no_versions_returns_empty(self, mock_fv_for_flows):
        from langflow.api.v1.deployments import list_deployments

        result = await list_deployments(
            provider_id=uuid4(),
            session=AsyncMock(),
            current_user=_fake_user(),
            params=SimpleNamespace(page=1, size=20),
            deployment_type=None,
            load_from_provider=False,
            flow_ids=[str(uuid4())],
        )

        assert result.deployments == []
        assert result.total == 0
        mock_fv_for_flows.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.flow_version_ids_for_flows", new_callable=AsyncMock)
    async def test_flow_ids_resolves_to_flow_version_ids(
        self,
        mock_fv_for_flows,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_list_synced,
    ):
        from langflow.api.v1.deployments import list_deployments

        fv_id = uuid4()
        mock_fv_for_flows.return_value = [fv_id]

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        mock_resolve_adapter.return_value = AsyncMock()

        mapper = MagicMock()
        mapper.shape_deployment_list_items.return_value = []
        mock_get_mapper.return_value = mapper

        mock_list_synced.return_value = ([], 0)

        await list_deployments(
            provider_id=pa.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            params=SimpleNamespace(page=1, size=20),
            deployment_type=None,
            load_from_provider=False,
            flow_ids=[str(uuid4())],
        )

        mock_list_synced.assert_awaited_once()
        call_kwargs = mock_list_synced.call_args.kwargs
        assert call_kwargs["flow_version_ids"] == [fv_id]


# ---------------------------------------------------------------------------
# list_deployments: project_id filter
# ---------------------------------------------------------------------------


class TestListDeploymentsProjectIdFilter:
    @pytest.mark.asyncio
    async def test_load_from_provider_rejects_project_id(self):
        """project_id filtering is not supported when load_from_provider=true."""
        from langflow.api.v1.deployments import list_deployments

        with pytest.raises(HTTPException) as exc_info:
            await list_deployments(
                provider_id=uuid4(),
                session=AsyncMock(),
                current_user=_fake_user(),
                params=SimpleNamespace(page=1, size=20),
                deployment_type=None,
                load_from_provider=True,
                project_id=uuid4(),
            )
        assert exc_info.value.status_code == 422
        assert "project_id" in exc_info.value.detail

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_project_id_threaded_to_synced(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_list_synced,
    ):
        """When project_id is supplied it is forwarded to list_deployments_synced."""
        from langflow.api.v1.deployments import list_deployments

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        mock_resolve_adapter.return_value = MagicMock()
        mapper = MagicMock()
        mapper.shape_deployment_list_items.return_value = []
        mock_get_mapper.return_value = mapper
        mock_list_synced.return_value = ([], 0)

        pid = uuid4()
        await list_deployments(
            provider_id=pa.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            params=SimpleNamespace(page=1, size=20),
            deployment_type=None,
            load_from_provider=False,
            project_id=pid,
        )

        mock_list_synced.assert_awaited_once()
        assert mock_list_synced.call_args.kwargs["project_id"] == pid

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_project_id_none_when_omitted(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_list_synced,
    ):
        """When project_id is not supplied, None is forwarded (no filter)."""
        from langflow.api.v1.deployments import list_deployments

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        mock_resolve_adapter.return_value = MagicMock()
        mapper = MagicMock()
        mapper.shape_deployment_list_items.return_value = []
        mock_get_mapper.return_value = mapper
        mock_list_synced.return_value = ([], 0)

        await list_deployments(
            provider_id=pa.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            params=SimpleNamespace(page=1, size=20),
            deployment_type=None,
            load_from_provider=False,
        )

        mock_list_synced.assert_awaited_once()
        assert mock_list_synced.call_args.kwargs["project_id"] is None


# ---------------------------------------------------------------------------
# config/snapshot passthrough listing routes
# ---------------------------------------------------------------------------


class TestConfigAndSnapshotListRoutes:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_configs_global_scope_uses_provider_id(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
    ):
        from langflow.api.v1.deployments import list_deployment_configs

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.list_configs.return_value = ConfigListResult(
            configs=[
                ConfigListItem(id="cfg-1", name="Config 1"),
                ConfigListItem(id="cfg-2", name="Config 2"),
            ],
            provider_result={},
        )
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        adapter_params = MagicMock()
        expected_response = MagicMock()
        mapper.resolve_config_list_adapter_params = AsyncMock(return_value=adapter_params)
        mapper.shape_config_list_result.return_value = expected_response
        mock_get_mapper.return_value = mapper

        result = await list_deployment_configs(
            provider_id=pa.id,
            deployment_id=None,
            page=1,
            size=10,
            session=AsyncMock(),
            current_user=_fake_user(),
        )

        assert result is expected_response
        mapper.resolve_config_list_adapter_params.assert_awaited_once_with(
            deployment_resource_key=None,
            provider_params=None,
            db=ANY,
        )
        call_params = adapter.list_configs.call_args.kwargs["params"]
        assert call_params is adapter_params
        mapper.shape_config_list_result.assert_called_once_with(
            adapter.list_configs.return_value,
            page=1,
            size=10,
        )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.get_deployment_row_or_404", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_snapshots_deployment_scope_resolves_provider_from_deployment(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_get_deployment,
    ):
        from langflow.api.v1.deployments import list_deployment_snapshots

        pa = _fake_provider_account()
        deployment = _fake_deployment_row(resource_key="dep-key")
        deployment.deployment_provider_account_id = pa.id
        mock_get_deployment.return_value = deployment
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.list_snapshots.return_value = SnapshotListResult(
            snapshots=[
                SnapshotItem(id="tool-1", name="Tool 1"),
                SnapshotItem(id="tool-2", name="Tool 2"),
            ],
            provider_result={"deployment_id": "dep-1"},
        )
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        adapter_params = MagicMock()
        expected_response = MagicMock()
        mapper.resolve_snapshot_list_adapter_params = AsyncMock(return_value=adapter_params)
        mapper.shape_snapshot_list_result.return_value = expected_response
        mock_get_mapper.return_value = mapper

        result = await list_deployment_snapshots(
            provider_id=pa.id,
            deployment_id=deployment.id,
            names=None,
            page=1,
            size=10,
            session=AsyncMock(),
            current_user=_fake_user(),
        )

        assert result is expected_response
        mapper.resolve_snapshot_list_adapter_params.assert_awaited_once_with(
            deployment_resource_key="dep-key",
            snapshot_names=None,
            provider_params=None,
            db=ANY,
        )
        call_params = adapter.list_snapshots.call_args.kwargs["params"]
        assert call_params is adapter_params
        mapper.shape_snapshot_list_result.assert_called_once_with(
            adapter.list_snapshots.return_value,
            page=1,
            size=10,
        )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_snapshots_rejects_deployment_id_with_names(
        self,
        mock_get_pa,
    ):
        from langflow.api.v1.deployments import list_deployment_snapshots

        with pytest.raises(HTTPException) as exc_info:
            await list_deployment_snapshots(
                provider_id=uuid4(),
                deployment_id=uuid4(),
                names=["my_tool"],
                page=1,
                size=10,
                session=AsyncMock(),
                current_user=_fake_user(),
            )

        assert exc_info.value.status_code == 422
        assert exc_info.value.detail == "filtering by both deployment_id and names is not supported."
        mock_get_pa.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_snapshots_passes_names_to_mapper(
        self,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
    ):
        from langflow.api.v1.deployments import list_deployment_snapshots

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        adapter.list_snapshots.return_value = SnapshotListResult(
            snapshots=[SnapshotItem(id="tool-1", name="my_tool")],
        )
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        adapter_params = MagicMock()
        expected_response = MagicMock()
        mapper.resolve_snapshot_list_adapter_params = AsyncMock(return_value=adapter_params)
        mapper.shape_snapshot_list_result.return_value = expected_response
        mock_get_mapper.return_value = mapper

        result = await list_deployment_snapshots(
            provider_id=pa.id,
            deployment_id=None,
            names=["my_tool", "other_tool"],
            page=1,
            size=10,
            session=AsyncMock(),
            current_user=_fake_user(),
        )

        assert result is expected_response
        mapper.resolve_snapshot_list_adapter_params.assert_awaited_once_with(
            deployment_resource_key=None,
            snapshot_names=["my_tool", "other_tool"],
            provider_params=None,
            db=ANY,
        )


# ---------------------------------------------------------------------------
# update_snapshot
# ---------------------------------------------------------------------------


class TestUpdateSnapshotRoute:
    @pytest.mark.asyncio
    @patch("langflow.services.database.models.flow_version.crud.get_flow_version_entry", new_callable=AsyncMock)
    @patch("langflow.services.database.models.deployment.crud.get_deployment", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.update_flow_version_by_provider_snapshot_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.get_attachment_by_provider_snapshot_id", new_callable=AsyncMock)
    async def test_updates_all_attachment_rows_for_snapshot(
        self,
        mock_get_attachment,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_rows,
        mock_get_deployment_row,
        mock_get_flow_version,
    ):
        from langflow.api.v1.deployments import update_snapshot

        user = _fake_user()
        flow_id = uuid4()
        target_flow_version_id = uuid4()
        attachment = SimpleNamespace(
            flow_version_id=uuid4(),
            deployment_id=uuid4(),
            provider_snapshot_id="tool-1",
        )
        deployment = _fake_deployment_row(
            id=attachment.deployment_id,
            deployment_provider_account_id=uuid4(),
        )
        provider_account = _fake_provider_account()
        provider_account.id = deployment.deployment_provider_account_id

        mock_get_attachment.return_value = attachment
        mock_get_deployment_row.return_value = deployment
        mock_get_flow_version.return_value = SimpleNamespace(id=target_flow_version_id, flow_id=flow_id, data={})
        mock_get_pa.return_value = provider_account
        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.resolve_snapshot_update_artifact.return_value = {"artifact": "payload"}
        mock_get_mapper.return_value = mapper
        mock_update_rows.return_value = 2

        session = AsyncMock()
        session.get.return_value = SimpleNamespace(id=flow_id)

        response = await update_snapshot(
            provider_snapshot_id=" tool-1 ",
            body=SnapshotUpdateRequest(flow_version_id=target_flow_version_id),
            session=session,
            current_user=user,
            telemetry=_fake_telemetry(),
        )

        assert response.flow_version_id == target_flow_version_id
        assert response.provider_snapshot_id == "tool-1"
        mock_get_attachment.assert_awaited_once_with(
            session,
            user_id=user.id,
            provider_snapshot_id="tool-1",
        )
        mock_update_rows.assert_awaited_once_with(
            session,
            user_id=user.id,
            provider_snapshot_id="tool-1",
            flow_version_id=target_flow_version_id,
        )
        session.commit.assert_awaited_once()
        session.rollback.assert_not_awaited()

    @pytest.mark.asyncio
    @patch("langflow.services.database.models.flow_version.crud.get_flow_version_entry", new_callable=AsyncMock)
    @patch("langflow.services.database.models.deployment.crud.get_deployment", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.update_flow_version_by_provider_snapshot_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.get_attachment_by_provider_snapshot_id", new_callable=AsyncMock)
    async def test_commit_failure_attempts_provider_compensation(
        self,
        mock_get_attachment,
        mock_get_pa,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_rows,
        mock_get_deployment_row,
        mock_get_flow_version,
    ):
        from langflow.api.v1.deployments import update_snapshot

        user = _fake_user()
        flow_id = uuid4()
        previous_flow_version_id = uuid4()
        target_flow_version_id = uuid4()
        attachment = SimpleNamespace(
            flow_version_id=previous_flow_version_id,
            deployment_id=uuid4(),
            provider_snapshot_id="tool-1",
        )
        deployment = _fake_deployment_row(
            id=attachment.deployment_id,
            deployment_provider_account_id=uuid4(),
        )
        provider_account = _fake_provider_account()
        provider_account.id = deployment.deployment_provider_account_id

        target_version = SimpleNamespace(id=target_flow_version_id, flow_id=flow_id, data={"nodes": []})
        previous_version = SimpleNamespace(id=previous_flow_version_id, flow_id=flow_id, data={"nodes": []})
        mock_get_flow_version.side_effect = [target_version, previous_version]
        mock_get_attachment.return_value = attachment
        mock_get_deployment_row.return_value = deployment
        mock_get_pa.return_value = provider_account
        adapter = AsyncMock()
        adapter.update_snapshot.return_value = None
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.resolve_snapshot_update_artifact.side_effect = [{"artifact": "new"}, {"artifact": "previous"}]
        mock_get_mapper.return_value = mapper
        mock_update_rows.return_value = 2

        session = AsyncMock()
        session.get.return_value = SimpleNamespace(id=flow_id)
        session.commit.side_effect = RuntimeError("commit failed")

        with pytest.raises(RuntimeError, match="commit failed"):
            await update_snapshot(
                provider_snapshot_id="tool-1",
                body=SnapshotUpdateRequest(flow_version_id=target_flow_version_id),
                session=session,
                current_user=user,
                telemetry=_fake_telemetry(),
            )

        session.commit.assert_awaited_once()
        session.rollback.assert_awaited_once()
        assert adapter.update_snapshot.await_count == 2


# ---------------------------------------------------------------------------
# list_deployment_flow_versions
# ---------------------------------------------------------------------------


class TestListDeploymentFlowVersionsRoute:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployment_flow_versions_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_returns_paginated_flow_versions(
        self,
        mock_resolve,
        mock_list_flow_versions_synced,
    ):
        from langflow.api.v1.deployments import list_deployment_flow_versions

        deployment_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        rows = [(SimpleNamespace(provider_snapshot_id="tool-1", created_at=None), SimpleNamespace())]
        snapshot_result = SnapshotListResult(
            snapshots=[
                SnapshotItem(
                    id="tool-1",
                    name="Tool 1",
                    provider_data={"connections": {"cfg-1": "conn-1"}},
                )
            ]
        )
        mock_resolve.return_value = (deployment_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")
        mock_list_flow_versions_synced.return_value = (rows, 7, snapshot_result)
        mapper.shape_flow_version_list_result.return_value = SimpleNamespace(
            page=2,
            size=5,
            total=7,
            flow_versions=[
                SimpleNamespace(
                    provider_snapshot_id="tool-1",
                    provider_data={"app_ids": ["cfg-1"]},
                )
            ],
        )

        flow_id = uuid4()
        response = await list_deployment_flow_versions(
            deployment_id=deployment_row.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            page=2,
            size=5,
            flow_ids=[flow_id],
        )

        assert response.page == 2
        assert response.size == 5
        assert response.total == 7
        assert len(response.flow_versions) == 1
        assert response.flow_versions[0].provider_snapshot_id == "tool-1"
        assert response.flow_versions[0].provider_data == {"app_ids": ["cfg-1"]}

        mock_list_flow_versions_synced.assert_awaited_once()
        helper_kwargs = mock_list_flow_versions_synced.call_args.kwargs
        assert helper_kwargs["provider_id"] == deployment_row.deployment_provider_account_id
        assert helper_kwargs["deployment_id"] == deployment_row.id
        assert helper_kwargs["page"] == 2
        assert helper_kwargs["size"] == 5
        assert helper_kwargs["flow_ids"] == [flow_id]
        mapper.shape_flow_version_list_result.assert_called_once_with(
            rows=rows,
            snapshot_result=snapshot_result,
            page=2,
            size=5,
            total=7,
        )


# ---------------------------------------------------------------------------
# provider account routes
# ---------------------------------------------------------------------------


class TestProviderAccountRoutes:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_update_provider_account_name_only_skips_credential_verification(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_provider_account,
    ):
        """PATCH skips credential verification when only name changes."""
        from langflow.api.v1.deployments import update_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        mapper = MagicMock()
        mapper.resolve_provider_account_update.return_value = {"name": "renamed"}
        mock_get_mapper.return_value = mapper
        mock_update_provider_account.return_value = existing_account

        session = AsyncMock()
        await update_provider_account(
            provider_id=existing_account.id,
            session=session,
            payload=DeploymentProviderAccountUpdateRequest(name="renamed"),
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        mapper.resolve_verify_credentials_for_update.assert_not_called()
        mock_resolve_adapter.assert_not_called()
        mock_update_provider_account.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_update_provider_account_verify_failure_returns_401(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_provider_account,
    ):
        """PATCH verifies new credentials before persisting them."""
        from langflow.api.v1.deployments import update_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account

        mapper = MagicMock()
        mapper.resolve_verify_credentials_for_update.return_value = MagicMock()
        mock_get_mapper.return_value = mapper

        adapter = AsyncMock()
        adapter.verify_credentials.side_effect = AuthenticationError(
            message="bad creds",
            error_code="authentication_error",
        )
        mock_resolve_adapter.return_value = adapter

        with pytest.raises(HTTPException) as exc_info:
            await update_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                payload=DeploymentProviderAccountUpdateRequest(provider_data={"api_key": "new-api-key"}),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 401
        mock_update_provider_account.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.create_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    async def test_create_provider_account_conflict_returns_409(
        self,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_create_provider_account,
    ):
        """POST converts duplicate provider-account conflicts into 409 responses."""
        from langflow.api.v1.deployments import create_provider_account

        mapper = MagicMock()
        mapper.resolve_verify_credentials_for_create.return_value = MagicMock()
        mapper.resolve_provider_account_create.return_value = MagicMock()
        mock_get_mapper.return_value = mapper

        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mock_create_provider_account.side_effect = ValueError("Provider account already exists")

        payload = DeploymentProviderAccountCreateRequest(
            name="prod",
            provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE,
            provider_data={
                "url": "https://api.us-south.wxo.cloud.ibm.com/instances/tenant-1",
                "api_key": "api-key",  # pragma: allowlist secret
            },
        )

        with pytest.raises(HTTPException) as exc_info:
            await create_provider_account(
                session=AsyncMock(),
                payload=payload,
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "Provider account is already tracked by user."

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_update_provider_account_conflict_returns_409(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_provider_account,
    ):
        """PATCH converts duplicate provider-account conflicts into 409 responses."""
        from langflow.api.v1.deployments import update_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account

        mapper = MagicMock()
        mapper.resolve_verify_credentials_for_update.return_value = None
        mapper.resolve_provider_account_update.return_value = {"name": "prod"}
        mock_get_mapper.return_value = mapper
        mock_resolve_adapter.return_value = AsyncMock()
        mock_update_provider_account.side_effect = ValueError(
            "Provider account update conflicts with an existing record"
        )

        with pytest.raises(HTTPException) as exc_info:
            await update_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                payload=DeploymentProviderAccountUpdateRequest(name="prod"),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "Provider account is already tracked by user."

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_update_provider_account_raw_guard_exception_propagates(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_provider_account,
    ):
        """PATCH preserves raw guard-shaped DB exceptions without rewriting them."""
        from langflow.api.v1.deployments import update_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account

        mapper = MagicMock()
        mapper.resolve_verify_credentials_for_update.return_value = None
        mapper.resolve_provider_account_update.return_value = {"provider_data": {"tenant_id": "tenant-renamed"}}
        mock_get_mapper.return_value = mapper
        mock_resolve_adapter.return_value = AsyncMock()
        mock_update_provider_account.side_effect = Exception(
            "DEPLOYMENT_GUARD:DEPLOYMENT_PROVIDER_ACCOUNT_IDENTITY_UPDATE:"
            "Cannot modify provider key, provider tenant id, or provider URL "
            "on an existing deployment provider account. "
            "Re-create the account instead."
        )

        with pytest.raises(
            Exception,
            match="DEPLOYMENT_GUARD:DEPLOYMENT_PROVIDER_ACCOUNT_IDENTITY_UPDATE",
        ):
            await update_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                payload=DeploymentProviderAccountUpdateRequest(provider_data={"tenant_id": "tenant-renamed"}),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_update_provider_account_propagates_deployment_guard_error_instance(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_update_provider_account,
    ):
        """PATCH preserves ORM-raised DeploymentGuardError exceptions."""
        from langflow.api.v1.deployments import update_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account

        mapper = MagicMock()
        mapper.resolve_verify_credentials_for_update.return_value = None
        mapper.resolve_provider_account_update.return_value = {"provider_data": {"tenant_id": "tenant-renamed"}}
        mock_get_mapper.return_value = mapper
        mock_resolve_adapter.return_value = AsyncMock()
        mock_update_provider_account.side_effect = DeploymentGuardError(
            code="DEPLOYMENT_PROVIDER_ACCOUNT_IDENTITY_UPDATE",
            technical_detail="Cannot modify provider key, provider tenant id, or provider URL.",
            detail="Cannot modify provider key, provider tenant id, or provider URL.",
        )

        with pytest.raises(
            DeploymentGuardError,
            match="Cannot modify provider key, provider tenant id, or provider URL",
        ):
            await update_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                payload=DeploymentProviderAccountUpdateRequest(provider_data={"tenant_id": "tenant-renamed"}),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock, return_value=([], 1))
    @patch(f"{ROUTES_MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_delete_provider_account_rejects_when_deployments_exist(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_count_deployments,
        mock_list_synced,
        mock_delete_provider_account,
    ):
        """DELETE refuses to remove provider accounts that still own deployments."""
        from langflow.api.v1.deployments import delete_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        mock_get_mapper.return_value = MagicMock()
        mock_resolve_adapter.return_value = AsyncMock()

        with pytest.raises(HTTPException) as exc_info:
            await delete_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 409
        mock_count_deployments.assert_awaited_once()
        mock_list_synced.assert_awaited_once()
        mock_delete_provider_account.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock, return_value=([], 0))
    @patch(f"{ROUTES_MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_delete_provider_account_prunes_stale_rows_before_delete(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_count_deployments,
        mock_list_synced,
        mock_delete_provider_account,
    ):
        """DELETE can proceed when reconciliation shows only stale local deployments remain."""
        from langflow.api.v1.deployments import delete_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        mock_get_mapper.return_value = MagicMock()
        mock_resolve_adapter.return_value = AsyncMock()
        session = AsyncMock()

        response = await delete_provider_account(
            provider_id=existing_account.id,
            session=session,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        assert response.status_code == 204
        mock_count_deployments.assert_awaited_once()
        mock_list_synced.assert_awaited_once()
        mock_delete_provider_account.assert_awaited_once_with(
            session,
            provider_account=existing_account,
        )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock, return_value=([], 0))
    @patch(f"{ROUTES_MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_delete_provider_account_sets_deployment_provider_scope_for_reconciliation(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_count_deployments,  # noqa: ARG002
        mock_list_synced,
        mock_delete_provider_account,  # noqa: ARG002
    ):
        """Reconciliation must run inside ``deployment_provider_scope`` so adapters can resolve credentials.

        Regression test: without the scope, the WxO adapter raises
        ``CredentialResolutionError("Deployment account context is not available...")``
        with status 401, the reconciliation is silently dropped, and the stale local
        count blocks every provider-account delete with a 409.
        """
        from langflow.api.v1.deployments import delete_provider_account
        from langflow.services.adapters.deployment.context import DeploymentProviderIDContext

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        mock_get_mapper.return_value = MagicMock()
        mock_resolve_adapter.return_value = AsyncMock()

        observed_provider_ids: list[object] = []

        async def capture_scope(*_args, **_kwargs):
            current = DeploymentProviderIDContext.get_current()
            observed_provider_ids.append(None if current is None else current.provider_id)
            return ([], 0)

        mock_list_synced.side_effect = capture_scope

        response = await delete_provider_account(
            provider_id=existing_account.id,
            session=AsyncMock(),
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        assert response.status_code == 204
        assert observed_provider_ids == [existing_account.id], (
            "list_deployments_synced must run inside deployment_provider_scope(provider_account.id) "
            "so the adapter can resolve credentials from DeploymentProviderIDContext."
        )
        # And the scope must not leak after the handler returns.
        assert DeploymentProviderIDContext.get_current() is None

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_delete_provider_account_falls_back_to_local_count_when_reconciliation_raises(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
        mock_count_deployments,
        mock_list_synced,
        mock_delete_provider_account,
    ):
        """If reconciliation fails (e.g. adapter credential error), fall back to the local count.

        The handler must surface the safe 409 (rather than crash) and must NOT delete the
        provider account row while local deployments are still tracked.
        """
        from langflow.api.v1.deployments import delete_provider_account
        from lfx.services.adapters.deployment.exceptions import CredentialResolutionError

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        mock_get_mapper.return_value = MagicMock()
        mock_resolve_adapter.return_value = AsyncMock()
        mock_list_synced.side_effect = CredentialResolutionError(
            message="Deployment account context is not available for adapter resolution."
        )

        with pytest.raises(HTTPException) as exc_info:
            await delete_provider_account(
                provider_id=existing_account.id,
                session=AsyncMock(),
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 409
        mock_count_deployments.assert_awaited_once()
        mock_list_synced.assert_awaited_once()
        mock_delete_provider_account.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_provider_account_row", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployments_synced", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=0)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_delete_provider_account_skips_reconciliation_when_no_local_deployments(
        self,
        mock_get_provider_account,
        mock_get_mapper,  # noqa: ARG002
        mock_resolve_adapter,  # noqa: ARG002
        mock_count_deployments,
        mock_list_synced,
        mock_delete_provider_account,
    ):
        """When the local count is 0 we must not hit the provider; just delete the row."""
        from langflow.api.v1.deployments import delete_provider_account

        existing_account = _fake_provider_account()
        mock_get_provider_account.return_value = existing_account
        session = AsyncMock()

        response = await delete_provider_account(
            provider_id=existing_account.id,
            session=session,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        assert response.status_code == 204
        mock_count_deployments.assert_awaited_once()
        mock_list_synced.assert_not_awaited()
        mock_delete_provider_account.assert_awaited_once_with(
            session,
            provider_account=existing_account,
        )


# ---------------------------------------------------------------------------
# update_deployment: rollback on commit failure
# ---------------------------------------------------------------------------


class TestUpdateDeploymentRollback:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.rollback_provider_update", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.update_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update", return_value=([], []))
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_rollback_called_on_commit_failure(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,  # noqa: ARG002
        mock_validate_fv,  # noqa: ARG002
        mock_resolve_snap,  # noqa: ARG002
        mock_apply_patch,  # noqa: ARG002
        mock_update_db,  # noqa: ARG002
        mock_rollback,
    ):
        """When session.commit() fails, rollback_provider_update is called."""
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()
        session.commit.side_effect = RuntimeError("DB commit failed")

        payload = MagicMock()
        payload.name = None
        payload.description = None

        with pytest.raises(RuntimeError, match="DB commit failed"):
            await update_deployment(
                deployment_id=dep_row.id,
                session=session,
                payload=payload,
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        mock_rollback.assert_awaited_once()
        assert mock_rollback.call_args.kwargs["deployment_db_id"] == dep_row.id
        assert mock_rollback.call_args.kwargs["deployment_resource_key"] == dep_row.resource_key
        assert (
            mock_rollback.call_args.kwargs["deployment_provider_account_id"] == dep_row.deployment_provider_account_id
        )

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.rollback_provider_update", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.update_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update", return_value=([], []))
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_no_rollback_on_success(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,  # noqa: ARG002
        mock_validate_fv,  # noqa: ARG002
        mock_resolve_snap,  # noqa: ARG002
        mock_apply_patch,  # noqa: ARG002
        mock_update_db,  # noqa: ARG002
        mock_rollback,
    ):
        """On successful commit, rollback is NOT called."""
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.name = None
        payload.description = None

        await update_deployment(
            deployment_id=dep_row.id,
            session=session,
            payload=payload,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        mock_rollback.assert_not_awaited()


class TestUpdateDeploymentAlreadyAttachedFiltering:
    """Bind operations for flow versions that already have DB attachments.

    These should not be passed to resolve_added_snapshot_bindings_for_update,
    since the adapter does not return new snapshot bindings for reused tools.
    """

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments_for_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update")
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_already_attached_flow_versions_excluded_from_snapshot_resolution(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,
        mock_validate_fv,  # noqa: ARG002
        mock_list_attachments,
        mock_resolve_snap,
        mock_apply_patch,  # noqa: ARG002
    ):
        """Only truly new flow version IDs are passed to snapshot resolution.

        When some bind flow version IDs already have DB attachments, only the
        truly new ones are passed to resolve_added_snapshot_bindings_for_update.
        """
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        reused_fv_id = uuid4()
        new_fv_id = uuid4()
        mock_resolve_fvp.return_value = ([reused_fv_id, new_fv_id], [])

        mock_list_attachments.return_value = [
            SimpleNamespace(flow_version_id=reused_fv_id, provider_snapshot_id="existing-tool-1"),
        ]

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.name = None
        payload.description = None

        await update_deployment(
            deployment_id=dep_row.id,
            session=session,
            payload=payload,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        mock_resolve_snap.assert_called_once()
        resolved_fv_ids = mock_resolve_snap.call_args.kwargs["added_flow_version_ids"]
        assert resolved_fv_ids == [new_fv_id]

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments_for_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update")
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_all_flow_versions_already_attached_passes_empty_list(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,
        mock_validate_fv,  # noqa: ARG002
        mock_list_attachments,
        mock_resolve_snap,
        mock_apply_patch,  # noqa: ARG002
    ):
        """An empty list is passed when all flow versions are already attached.

        When all bind flow versions already have attachments, an empty list
        is passed to resolve_added_snapshot_bindings_for_update.
        """
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        fv_id_1 = uuid4()
        fv_id_2 = uuid4()
        mock_resolve_fvp.return_value = ([fv_id_1, fv_id_2], [])

        mock_list_attachments.return_value = [
            SimpleNamespace(flow_version_id=fv_id_1, provider_snapshot_id="tool-1"),
            SimpleNamespace(flow_version_id=fv_id_2, provider_snapshot_id="tool-2"),
        ]

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.name = None
        payload.description = None

        await update_deployment(
            deployment_id=dep_row.id,
            session=session,
            payload=payload,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        resolved_fv_ids = mock_resolve_snap.call_args.kwargs["added_flow_version_ids"]
        assert resolved_fv_ids == []

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments_for_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update")
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_no_existing_attachments_passes_all_flow_versions(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,
        mock_validate_fv,  # noqa: ARG002
        mock_list_attachments,
        mock_resolve_snap,
        mock_apply_patch,  # noqa: ARG002
    ):
        """When no bind flow versions have existing attachments, all are passed through."""
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        fv_id_1 = uuid4()
        fv_id_2 = uuid4()
        mock_resolve_fvp.return_value = ([fv_id_1, fv_id_2], [])

        mock_list_attachments.return_value = []

        session = AsyncMock()
        session.commit.return_value = None

        payload = MagicMock()
        payload.name = None
        payload.description = None

        await update_deployment(
            deployment_id=dep_row.id,
            session=session,
            payload=payload,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        resolved_fv_ids = mock_resolve_snap.call_args.kwargs["added_flow_version_ids"]
        assert resolved_fv_ids == [fv_id_1, fv_id_2]


class TestUpdateDeploymentMetadataPersistence:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.update_deployment_db", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.apply_flow_version_patch_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_added_snapshot_bindings_for_update", return_value=[])
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update", return_value=([], []))
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_explicit_null_description_is_persisted(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,  # noqa: ARG002
        mock_validate_fv,  # noqa: ARG002
        mock_resolve_snap,  # noqa: ARG002
        mock_apply_patch,  # noqa: ARG002
        mock_update_db,
    ):
        """PATCH should persist an explicitly-cleared description alongside other spec updates."""
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row(name="old-name", description="old-description")
        updated_row = _fake_deployment_row(
            id=dep_row.id,
            resource_key=dep_row.resource_key,
            name="renamed",
            description=None,
            deployment_provider_account_id=dep_row.deployment_provider_account_id,
            project_id=dep_row.project_id,
        )
        adapter = AsyncMock()
        mapper = MagicMock()
        update_result = DeploymentUpdateResult(id="provider-dep-1")
        adapter.update.return_value = update_result
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mapper.shape_deployment_update_result.return_value = MagicMock()
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")
        mock_update_db.return_value = updated_row

        session = AsyncMock()
        session.commit.return_value = None

        payload = DeploymentUpdateRequest.model_validate(
            {
                "name": "renamed",
                "description": None,
            }
        )

        await update_deployment(
            deployment_id=dep_row.id,
            session=session,
            payload=payload,
            current_user=_fake_user(),
            telemetry=_fake_telemetry(),
        )

        mock_update_db.assert_awaited_once()
        update_kwargs = mock_update_db.call_args.kwargs
        assert update_kwargs["name"] == "renamed"
        assert "description" in update_kwargs
        assert update_kwargs["description"] is None


# ---------------------------------------------------------------------------
# get_deployment: deployment-level sync
# ---------------------------------------------------------------------------


class TestGetDeploymentSync:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock, return_value=[])
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_stale_row_deleted_when_provider_returns_not_found(
        self,
        mock_resolve,
        mock_delete_row,
        mock_list_att,  # noqa: ARG002
    ):
        """When the provider raises DeploymentNotFoundError, the DB row is deleted and 404 returned."""
        from langflow.api.v1.deployments import get_deployment
        from langflow.api.v1.mappers.deployments.base import BaseDeploymentMapper

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        adapter.get.side_effect = DeploymentNotFoundError(message="gone")
        mock_resolve.return_value = (dep_row, adapter, BaseDeploymentMapper(), "watsonx-orchestrate", "tenant-1")

        user = _fake_user()
        session = AsyncMock()

        with pytest.raises(HTTPException) as exc_info:
            await get_deployment(deployment_id=dep_row.id, session=session, current_user=user)

        assert exc_info.value.status_code == 404
        mock_delete_row.assert_awaited_once_with(session, user_id=user.id, deployment_id=dep_row.id)
        session.commit.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_non_404_adapter_error_does_not_delete_row(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """When adapter.get() raises a non-404 DeploymentServiceError, the DB row is kept."""
        from langflow.api.v1.deployments import get_deployment
        from langflow.api.v1.mappers.deployments.base import BaseDeploymentMapper

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        adapter.get.side_effect = AuthenticationError(message="bad creds", error_code="authentication_error")
        mock_resolve.return_value = (dep_row, adapter, BaseDeploymentMapper(), "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()

        with pytest.raises(HTTPException) as exc_info:
            await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert exc_info.value.status_code == 401
        mock_delete_row.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_service_unavailable_returns_503(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """When adapter.get() raises ServiceUnavailableError, 503 is returned and row is kept."""
        from langflow.api.v1.deployments import get_deployment
        from langflow.api.v1.mappers.deployments.base import BaseDeploymentMapper

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        adapter.get.side_effect = ServiceUnavailableError(message="provider down")
        mock_resolve.return_value = (dep_row, adapter, BaseDeploymentMapper(), "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()

        with pytest.raises(HTTPException) as exc_info:
            await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert exc_info.value.status_code == 503
        mock_delete_row.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock, return_value=[])
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_get_deployment_does_not_inject_resource_key_into_provider_data(
        self,
        mock_resolve,
        mock_list_att,  # noqa: ARG002
    ):
        from langflow.api.v1.deployments import get_deployment
        from langflow.api.v1.mappers.deployments.base import BaseDeploymentMapper

        class _MapperForGet(BaseDeploymentMapper):
            def shape_deployment_get_data(self, provider_data):
                if provider_data is None:
                    return None
                sanitized_provider_data = dict(provider_data)
                sanitized_provider_data.pop("tool_ids", None)
                return sanitized_provider_data or None

        created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
        updated_at = datetime(2026, 1, 3, 4, 5, 6, tzinfo=timezone.utc)
        dep_row = _fake_deployment_row(
            resource_key="provider-rk-1",
            name="db-owned-name",
            description="db-owned-description",
            deployment_type="agent",
            created_at=created_at,
            updated_at=updated_at,
        )
        adapter = AsyncMock()
        provider_deployment = MagicMock()
        provider_deployment.name = "provider-owned-name"
        provider_deployment.description = "provider-owned-description"
        provider_deployment.type = "worker"
        provider_deployment.created_at = None
        provider_deployment.updated_at = None
        provider_deployment.model_dump.return_value = {
            "provider_data": {
                "llm": "virtual-model/bedrock/openai.gpt-oss-120b-1:0",
                "tool_ids": ["tool-1", "tool-2"],
            }
        }
        adapter.get.return_value = provider_deployment
        mock_resolve.return_value = (dep_row, adapter, _MapperForGet(), "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()
        result = await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert result.resource_key == "provider-rk-1"
        assert result.name == "db-owned-name"
        assert result.description == "db-owned-description"
        assert result.type == "agent"
        assert result.created_at == created_at
        assert result.updated_at == updated_at
        assert "tool_ids" not in result.provider_data
        assert result.provider_data == {"llm": "virtual-model/bedrock/openai.gpt-oss-120b-1:0"}

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_unbound_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_snapshot_sync_corrects_attached_count(
        self,
        mock_resolve,
        mock_list_att,
        mock_delete_unbound,
    ):
        """Binding-aware sync corrects attached_count in the response."""
        from langflow.api.v1.deployments import get_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        provider_deployment = MagicMock()
        provider_deployment.name = "deployed-agent"
        provider_deployment.description = "desc"
        provider_deployment.type = "agent"
        provider_deployment.created_at = None
        provider_deployment.updated_at = None
        provider_deployment.model_dump.return_value = {}
        adapter.get.return_value = provider_deployment
        mapper.extract_snapshot_bindings_for_get.return_value = [
            ProviderSnapshotBinding(resource_key=dep_row.resource_key, snapshot_id="snap-1")
        ]
        mapper.shape_deployment_get_data.return_value = None
        mock_resolve.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        att_good = _fake_attachment(provider_snapshot_id="snap-1")
        mock_list_att.return_value = [att_good]

        session = AsyncMock()
        session.begin_nested = MagicMock(return_value=_AsyncNoopSavepoint())
        result = await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert result.attached_count == 1
        mock_delete_unbound.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_unbound_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_unsupported_provider_falls_back_to_unverified_count(
        self,
        mock_resolve,
        mock_list_att,
        mock_delete_unbound,
    ):
        """NotImplemented mapper sync falls back to unverified count."""
        from langflow.api.v1.deployments import get_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        provider_deployment = MagicMock()
        provider_deployment.name = "deployed-agent"
        provider_deployment.description = "desc"
        provider_deployment.type = "agent"
        provider_deployment.created_at = None
        provider_deployment.updated_at = None
        provider_deployment.model_dump.return_value = {}
        adapter.get.return_value = provider_deployment
        mapper.extract_snapshot_bindings_for_get.side_effect = NotImplementedError("not supported")
        mapper.shape_deployment_get_data.return_value = None
        mock_resolve.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")
        mock_list_att.return_value = [_fake_attachment(provider_snapshot_id=None)]

        session = AsyncMock()
        result = await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert result.attached_count == 1
        session.rollback.assert_not_awaited()
        mock_delete_unbound.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_unbound_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_binding_aware_sync_error_falls_back_to_unverified_count(
        self,
        mock_resolve,
        mock_list_att,
        mock_delete_unbound,
    ):
        """When binding-aware sync raises, response uses unverified attachment count."""
        from langflow.api.v1.deployments import get_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        provider_deployment = MagicMock()
        provider_deployment.name = "deployed-agent"
        provider_deployment.description = "desc"
        provider_deployment.type = "agent"
        provider_deployment.created_at = None
        provider_deployment.updated_at = None
        provider_deployment.model_dump.return_value = {}
        adapter.get.return_value = provider_deployment
        mapper.extract_snapshot_bindings_for_get.return_value = [
            ProviderSnapshotBinding(resource_key=dep_row.resource_key, snapshot_id="snap-1")
        ]
        mapper.shape_deployment_get_data.return_value = None
        mock_resolve.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        att1 = _fake_attachment(provider_snapshot_id="snap-1")
        att2 = _fake_attachment(provider_snapshot_id="snap-2")
        mock_list_att.return_value = [att1, att2]
        mock_delete_unbound.side_effect = RuntimeError("provider down")

        session = AsyncMock()
        session.begin_nested = MagicMock(return_value=_AsyncNoopSavepoint())
        result = await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert result.attached_count == 2
        session.rollback.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_unbound_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.list_deployment_attachments", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_binding_aware_sync_prunes_detached_attachments(
        self,
        mock_resolve,
        mock_list_att,
        mock_delete_unbound,
    ):
        """Binding-aware sync sends authoritative bindings and returns corrected count."""
        from langflow.api.v1.deployments import get_deployment

        dep_row = _fake_deployment_row(resource_key="agent-rk-1")
        adapter = AsyncMock()
        mapper = MagicMock()
        provider_deployment = MagicMock()
        provider_deployment.model_dump.return_value = {}
        adapter.get.return_value = provider_deployment
        mapper.extract_snapshot_bindings_for_get.return_value = [
            ProviderSnapshotBinding(resource_key="agent-rk-1", snapshot_id="snap-1")
        ]
        mapper.shape_deployment_get_data.return_value = None
        mock_resolve.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")
        mock_list_att.return_value = [_fake_attachment(provider_snapshot_id="snap-1")]

        session = AsyncMock()
        session.begin_nested = MagicMock(return_value=_AsyncNoopSavepoint())
        result = await get_deployment(deployment_id=dep_row.id, session=session, current_user=_fake_user())

        assert result.attached_count == 1
        mock_delete_unbound.assert_awaited_once_with(
            db=session,
            user_id=ANY,
            provider_account_id=dep_row.deployment_provider_account_id,
            deployment_ids=[dep_row.id],
            bindings=[ProviderSnapshotBinding(resource_key="agent-rk-1", snapshot_id="snap-1")],
        )


# ---------------------------------------------------------------------------
# delete_deployment: stale-row cleanup + commit retry
# ---------------------------------------------------------------------------


class TestDeleteDeployment:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_provider_not_found_still_deletes_local_row(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """Delete is idempotent when the provider agent is already gone."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        adapter.delete.side_effect = DeploymentNotFoundError(message="gone")
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        user = _fake_user()
        session = AsyncMock()

        response = await delete_deployment(
            deployment_id=dep_row.id, session=session, current_user=user, telemetry=_fake_telemetry()
        )

        assert response.status_code == 204
        mock_delete_row.assert_awaited_once_with(session, user_id=user.id, deployment_id=dep_row.id)
        session.commit.assert_awaited_once()
        session.rollback.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_non_404_provider_error_does_not_delete_local_row(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """Delete keeps the DB row when the provider call fails for non-404 reasons."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        adapter.delete.side_effect = AuthenticationError(message="bad creds", error_code="authentication_error")
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()

        with pytest.raises(HTTPException) as exc_info:
            await delete_deployment(
                deployment_id=dep_row.id, session=session, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 401
        mock_delete_row.assert_not_awaited()
        session.commit.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_commit_failure_retries_local_cleanup_once(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """A failed local commit after provider delete triggers one cleanup retry."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        user = _fake_user()
        session = AsyncMock()
        session.commit.side_effect = [RuntimeError("commit failed"), None]

        response = await delete_deployment(
            deployment_id=dep_row.id, session=session, current_user=user, telemetry=_fake_telemetry()
        )

        assert response.status_code == 204
        assert mock_delete_row.await_count == 2
        session.rollback.assert_awaited_once()
        assert session.commit.await_count == 2

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_repeated_local_cleanup_failure_returns_500(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """If cleanup still fails after retry, the route surfaces a 500."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        session = AsyncMock()
        session.commit.side_effect = [RuntimeError("commit failed"), RuntimeError("still failing")]

        with pytest.raises(HTTPException) as exc_info:
            await delete_deployment(
                deployment_id=dep_row.id, session=session, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 500
        assert mock_delete_row.await_count == 2
        assert session.rollback.await_count == 2
        assert session.commit.await_count == 2

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_include_provider_true_calls_adapter_delete(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """include_provider=True (default) calls adapter.delete to remove provider resources."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        user = _fake_user()
        session = AsyncMock()

        response = await delete_deployment(
            deployment_id=dep_row.id,
            session=session,
            current_user=user,
            include_provider=True,
            telemetry=_fake_telemetry(),
        )

        assert response.status_code == 204
        adapter.delete.assert_awaited_once()
        mock_delete_row.assert_awaited_once()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.delete_deployment_by_id", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_adapter_from_deployment", new_callable=AsyncMock)
    async def test_include_provider_false_skips_adapter_delete(
        self,
        mock_resolve,
        mock_delete_row,
    ):
        """include_provider=False skips the adapter entirely — only the DB row is removed."""
        from langflow.api.v1.deployments import delete_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mock_resolve.return_value = (dep_row, adapter, "watsonx-orchestrate", "tenant-1")

        user = _fake_user()
        session = AsyncMock()

        response = await delete_deployment(
            deployment_id=dep_row.id,
            session=session,
            current_user=user,
            include_provider=False,
            telemetry=_fake_telemetry(),
        )

        assert response.status_code == 204
        adapter.delete.assert_not_awaited()
        mock_delete_row.assert_awaited_once()


# ---------------------------------------------------------------------------
# create_deployment: duplicate name returns 409
# ---------------------------------------------------------------------------


class TestCreateDeploymentDuplicateName:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=True)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_duplicate_name_returns_409(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
    ):
        """When a deployment with the same name already exists, 409 is returned without calling the provider."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "duplicate-name"

        with pytest.raises(HTTPException) as exc_info:
            await create_deployment(
                session=AsyncMock(), payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 409
        assert "duplicate-name" in exc_info.value.detail

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=True)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_duplicate_name_does_not_call_provider(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_resolve_adapter,
    ):
        """When name already exists, the provider adapter is never invoked."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "taken"

        with pytest.raises(HTTPException):
            await create_deployment(
                session=AsyncMock(), payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        mock_resolve_adapter.assert_not_called()


# ---------------------------------------------------------------------------
# create_deployment: project-scoped flow version validation
# ---------------------------------------------------------------------------


class TestCreateDeploymentProjectValidation:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_validation_called_before_adapter(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
    ):
        """validate_project_scoped_flow_version_ids is called before the adapter create."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        fv_ids = [uuid4(), uuid4()]
        mapper.util_create_flow_version_ids.return_value = fv_ids
        mapper.util_existing_deployment_resource_key_for_create.return_value = None
        mock_get_mapper.return_value = mapper
        project_id = uuid4()
        mock_resolve_project.return_value = project_id

        mock_validate_fv.side_effect = HTTPException(status_code=404, detail="invalid")

        payload = MagicMock()
        payload.provider_id = pa.id

        with pytest.raises(HTTPException) as exc_info:
            await create_deployment(
                session=AsyncMock(), payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 404
        mock_validate_fv.assert_awaited_once()
        call_kwargs = mock_validate_fv.call_args.kwargs
        assert call_kwargs["flow_version_ids"] == fv_ids
        assert call_kwargs["project_id"] == project_id
        adapter.create.assert_not_awaited()

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_empty_flow_version_ids_skips_validation(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
    ):
        """When the mapper returns no flow_version_ids, validation still runs (it short-circuits internally)."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        create_result = DeploymentCreateResult(id="prov-1")
        adapter.create.return_value = create_result
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_create_flow_version_ids.return_value = []
        mapper.util_existing_deployment_resource_key_for_create.return_value = None
        mapper.resolve_deployment_create = AsyncMock(return_value=MagicMock())
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()

        payload = MagicMock()
        payload.provider_id = pa.id
        payload.name = "test"
        payload.type = "agent"
        payload.description = None

        session = AsyncMock()
        session.commit.return_value = None

        with (
            patch(f"{ROUTES_MODULE}.create_deployment_db", new_callable=AsyncMock, return_value=_fake_deployment_row()),
            patch(f"{ROUTES_MODULE}.attach_flow_versions", new_callable=AsyncMock),
        ):
            mapper.shape_deployment_create_result.return_value = MagicMock()
            await create_deployment(
                session=session, payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        mock_validate_fv.assert_awaited_once()
        assert mock_validate_fv.call_args.kwargs["flow_version_ids"] == []


class TestCreateDeploymentSchemaValidation:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_project_id_for_deployment_create", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.deployment_name_exists", new_callable=AsyncMock, return_value=False)
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_mapper_schema_validation_error_surfaces_422(
        self,
        mock_get_pa,
        mock_name_exists,  # noqa: ARG002
        mock_get_mapper,
        mock_resolve_adapter,
        mock_resolve_project,
        mock_validate_fv,
    ):
        """Mapper-level schema failures on create surface as HTTP 422."""
        from langflow.api.v1.deployments import create_deployment

        pa = _fake_provider_account()
        mock_get_pa.return_value = pa
        adapter = AsyncMock()
        mock_resolve_adapter.return_value = adapter
        mapper = MagicMock()
        mapper.util_create_flow_version_ids.return_value = [uuid4()]
        mapper.util_existing_deployment_resource_key_for_create.return_value = None
        mapper.resolve_deployment_create = AsyncMock(
            side_effect=HTTPException(
                status_code=422,
                detail="Invalid provider_data payload: simulated adapter validation failure",
            )
        )
        mock_get_mapper.return_value = mapper
        mock_resolve_project.return_value = uuid4()

        payload = MagicMock()
        payload.provider_id = pa.id

        with pytest.raises(HTTPException) as exc_info:
            await create_deployment(
                session=AsyncMock(), payload=payload, current_user=_fake_user(), telemetry=_fake_telemetry()
            )

        assert exc_info.value.status_code == 422
        mock_validate_fv.assert_awaited_once()
        adapter.create.assert_not_awaited()


# ---------------------------------------------------------------------------
# update_deployment: project-scoped flow version validation
# ---------------------------------------------------------------------------


class TestUpdateDeploymentProjectValidation:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.validate_project_scoped_flow_version_ids", new_callable=AsyncMock)
    @patch(f"{ROUTES_MODULE}.resolve_flow_version_patch_for_update")
    @patch(f"{ROUTES_MODULE}.resolve_adapter_mapper_from_deployment", new_callable=AsyncMock)
    async def test_validation_called_before_adapter(
        self,
        mock_resolve_amm,
        mock_resolve_fvp,
        mock_validate_fv,
    ):
        """validate_project_scoped_flow_version_ids is called before the adapter update."""
        from langflow.api.v1.deployments import update_deployment

        dep_row = _fake_deployment_row()
        adapter = AsyncMock()
        mapper = MagicMock()
        mapper.resolve_deployment_update = AsyncMock(return_value=MagicMock())
        mock_resolve_amm.return_value = (dep_row, adapter, mapper, "watsonx-orchestrate", "tenant-1")

        add_ids = [uuid4()]
        remove_ids = [uuid4()]
        mock_resolve_fvp.return_value = (add_ids, remove_ids)

        mock_validate_fv.side_effect = HTTPException(status_code=404, detail="out of project")

        payload = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            await update_deployment(
                deployment_id=dep_row.id,
                session=AsyncMock(),
                payload=payload,
                current_user=_fake_user(),
                telemetry=_fake_telemetry(),
            )

        assert exc_info.value.status_code == 404
        mock_validate_fv.assert_awaited_once()
        assert mock_validate_fv.call_args.kwargs["project_id"] == dep_row.project_id
        adapter.update.assert_not_awaited()


# ---------------------------------------------------------------------------
# list_deployment_llms
# ---------------------------------------------------------------------------


class TestListDeploymentLlms:
    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_llms_returns_shaped_response(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
    ):
        from langflow.api.v1.deployments import list_deployment_llms

        provider_account = _fake_provider_account()
        mock_get_provider_account.return_value = provider_account

        adapter = AsyncMock()
        llm_result = SimpleNamespace(
            provider_result={"models": [{"model_name": "model-a"}, {"model_name": "model-b"}]},
        )
        adapter.list_llms.return_value = llm_result
        mock_resolve_adapter.return_value = adapter

        mapper = MagicMock()
        mapper.shape_llm_list_result.return_value = DeploymentLlmListResponse(
            provider_data={"models": [{"model_name": "model-a"}, {"model_name": "model-b"}]}
        )
        mock_get_mapper.return_value = mapper

        response = await list_deployment_llms(
            provider_id=provider_account.id,
            session=AsyncMock(),
            current_user=_fake_user(),
        )

        assert response == DeploymentLlmListResponse(
            provider_data={"models": [{"model_name": "model-a"}, {"model_name": "model-b"}]}
        )
        adapter.list_llms.assert_awaited_once()
        mapper.shape_llm_list_result.assert_called_once_with(llm_result)

    @pytest.mark.asyncio
    @patch(f"{ROUTES_MODULE}.resolve_deployment_adapter")
    @patch(f"{ROUTES_MODULE}.get_deployment_mapper")
    @patch(f"{ROUTES_MODULE}.get_owned_provider_account_or_404", new_callable=AsyncMock)
    async def test_list_llms_maps_adapter_error(
        self,
        mock_get_provider_account,
        mock_get_mapper,
        mock_resolve_adapter,
    ):
        from langflow.api.v1.deployments import list_deployment_llms

        provider_account = _fake_provider_account()
        mock_get_provider_account.return_value = provider_account

        adapter = AsyncMock()
        adapter.list_llms.side_effect = ServiceUnavailableError(message="provider down")
        mock_resolve_adapter.return_value = adapter

        mapper = MagicMock()
        mock_get_mapper.return_value = mapper

        with pytest.raises(HTTPException) as exc_info:
            await list_deployment_llms(
                provider_id=provider_account.id,
                session=AsyncMock(),
                current_user=_fake_user(),
            )

        assert exc_info.value.status_code == 503
        mapper.shape_llm_list_result.assert_not_called()


# ---------------------------------------------------------------------------
# resolve_deployment_adapter
# ---------------------------------------------------------------------------


class TestResolveDeploymentAdapter:
    def test_empty_provider_key_raises_400(self):
        from langflow.api.v1.mappers.deployments.helpers import resolve_deployment_adapter

        with pytest.raises(HTTPException) as exc_info:
            resolve_deployment_adapter("")

        assert exc_info.value.status_code == 400
        assert "provider_key" in exc_info.value.detail.lower()

    def test_whitespace_only_provider_key_raises_400(self):
        from langflow.api.v1.mappers.deployments.helpers import resolve_deployment_adapter

        with pytest.raises(HTTPException) as exc_info:
            resolve_deployment_adapter("   ")

        assert exc_info.value.status_code == 400

    @patch(f"{HELPERS_MODULE}.get_deployment_adapter", return_value=None)
    def test_unknown_provider_key_raises_503(self, _mock_get):  # noqa: PT019
        from langflow.api.v1.mappers.deployments.helpers import resolve_deployment_adapter

        with pytest.raises(HTTPException) as exc_info:
            resolve_deployment_adapter("nonexistent_provider")

        assert exc_info.value.status_code == 503
        assert "nonexistent_provider" in exc_info.value.detail

    @patch(f"{HELPERS_MODULE}.get_deployment_adapter", side_effect=RuntimeError("registry boom"))
    def test_adapter_lookup_error_raises_500(self, _mock_get):  # noqa: PT019
        from langflow.api.v1.mappers.deployments.helpers import resolve_deployment_adapter

        with pytest.raises(HTTPException) as exc_info:
            resolve_deployment_adapter("bad_key")

        assert exc_info.value.status_code == 500

    @patch(f"{HELPERS_MODULE}.get_deployment_adapter")
    def test_valid_provider_key_returns_adapter(self, mock_get):
        from langflow.api.v1.mappers.deployments.helpers import resolve_deployment_adapter

        sentinel = MagicMock()
        mock_get.return_value = sentinel

        result = resolve_deployment_adapter("watsonx_orchestrate")

        assert result is sentinel
        mock_get.assert_called_once_with("watsonx_orchestrate")


# ---------------------------------------------------------------------------
# handle_adapter_errors: context manager integration
# ---------------------------------------------------------------------------


class TestHandleAdapterErrors:
    def test_maps_authentication_error_to_401(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise AuthenticationError(message="bad creds", error_code="authentication_error")

        assert exc_info.value.status_code == 401
        assert "bad creds" in exc_info.value.detail

    def test_maps_service_unavailable_to_503(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise ServiceUnavailableError(message="provider down")

        assert exc_info.value.status_code == 503

    def test_maps_not_found_to_404(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise DeploymentNotFoundError(message="gone")

        assert exc_info.value.status_code == 404

    def test_maps_conflict_with_mapper_formatter(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        deployment_mapper = MagicMock()
        deployment_mapper.format_conflict_detail.return_value = "friendly detail"

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors(mapper=deployment_mapper):
            raise ResourceConflictError(message="raw provider conflict")

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "friendly detail"
        deployment_mapper.format_conflict_detail.assert_called_once_with(
            "raw provider conflict",
            resource=None,
            resource_name=None,
        )

    def test_maps_conflict_passes_structured_resource_to_mapper_formatter(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        deployment_mapper = MagicMock()
        deployment_mapper.format_conflict_detail.return_value = "friendly detail"

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors(mapper=deployment_mapper):
            raise ResourceConflictError(message="raw provider conflict", resource="tool")

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "friendly detail"
        deployment_mapper.format_conflict_detail.assert_called_once_with(
            "raw provider conflict",
            resource="tool",
            resource_name=None,
        )

    def test_maps_conflict_passes_structured_resource_name_to_mapper_formatter(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        deployment_mapper = MagicMock()
        deployment_mapper.format_conflict_detail.return_value = "friendly detail"

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors(mapper=deployment_mapper):
            raise ResourceConflictError(
                message="raw provider conflict",
                resource="tool",
                resource_name="Simple_Agent",
            )

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "friendly detail"
        deployment_mapper.format_conflict_detail.assert_called_once_with(
            "raw provider conflict",
            resource="tool",
            resource_name="Simple_Agent",
        )

    def test_maps_conflict_without_mapper_passthrough(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise ResourceConflictError(message="raw provider conflict")

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail == "raw provider conflict"

    def test_passes_through_http_exception(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise HTTPException(status_code=418, detail="teapot")

        assert exc_info.value.status_code == 418

    def test_maps_not_implemented_to_501(self):
        from langflow.api.v1.mappers.deployments.helpers import handle_adapter_errors

        msg = "nope"
        with pytest.raises(HTTPException) as exc_info, handle_adapter_errors():
            raise NotImplementedError(msg)

        assert exc_info.value.status_code == 501
