"""Tests for deployment provider-sync helpers. Covers: - fetch_provider_resource_keys: ID-only matching, error handling - list_deployments_synced: cursor-based sync with inline stale-row deletion """ from __future__ import annotations from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID, uuid4 import pytest from fastapi import HTTPException from langflow.api.v1.mappers.deployments.base import BaseDeploymentMapper from langflow.api.v1.mappers.deployments.contracts import ( CreateSnapshotBinding, CreateSnapshotBindings, UpdateSnapshotBinding, UpdateSnapshotBindings, ) from langflow.api.v1.mappers.deployments.watsonx_orchestrate import WatsonxOrchestrateDeploymentMapper from langflow.api.v1.schemas.deployments import DeploymentUpdateRequest from lfx.services.adapters.deployment.exceptions import ServiceUnavailableError from lfx.services.adapters.deployment.schema import ( DeploymentCreateResult, DeploymentGetResult, DeploymentType, DeploymentUpdateResult, SnapshotItem, SnapshotListResult, ) MODULE = "langflow.api.v1.mappers.deployments.helpers" SYNC_MODULE = "langflow.api.v1.mappers.deployments.sync" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _mock_item(*, item_id: str, name: str = "irrelevant") -> SimpleNamespace: return SimpleNamespace(id=item_id, name=name) def _mock_provider_view(items: list) -> SimpleNamespace: return SimpleNamespace(deployments=items) def _wxo_deployment_provider_data( *, tool_ids: list[str], environments: list[str] | None = None, display_name: str = "Agent Display", description: str = "Agent description", llm: str = "meta-llm", ) -> dict[str, Any]: return { "display_name": display_name, "description": description, "tool_ids": tool_ids, "llm": llm, "environments": environments or [], } def _mock_deployment_row( resource_key: str, deployment_type: str | None = None, *, user_id: UUID | None = None, ) -> SimpleNamespace: return SimpleNamespace( id=uuid4(), resource_key=resource_key, display_name="Deployment", description=None, deployment_type=deployment_type, deployment_provider_account_id=uuid4(), user_id=user_id or uuid4(), ) class _NoSnapshotBindingMapper(BaseDeploymentMapper): def extract_snapshot_bindings(self, provider_view) -> list[Any]: _ = provider_view return [] def extract_metadata_for_list(self, provider_view) -> dict[str, dict[str, Any]]: if provider_view is None: return {} return { str(item.id): { "display_name": (getattr(item, "provider_data", {}) or {}).get( "display_name", getattr(item, "name", str(item.id)) ), "description": (getattr(item, "provider_data", {}) or {}).get("description"), } for item in provider_view.deployments } class _AsyncNoopSavepoint: async def __aenter__(self): return None async def __aexit__(self, exc_type, exc, tb): return False def _mock_async_db() -> MagicMock: db = MagicMock() db.begin_nested.return_value = _AsyncNoopSavepoint() db.exec = AsyncMock(return_value=MagicMock(all=MagicMock(return_value=[]))) return db @pytest.mark.asyncio @patch(f"{MODULE}.update_deployment_metadata", new_callable=AsyncMock) async def test_sync_deployment_display_metadata_passes_provider_metadata(mock_update_metadata): from langflow.api.v1.mappers.deployments.helpers import sync_deployment_display_metadata deployment = _mock_deployment_row("rk-1") long_description = "x" * 501 mock_update_metadata.return_value = deployment result = await sync_deployment_display_metadata( MagicMock(), deployment=deployment, provider_metadata={ "display_name": "Provider Name", "description": long_description, }, user_id=uuid4(), ) assert result is deployment mock_update_metadata.assert_awaited_once() assert mock_update_metadata.call_args.kwargs["display_name"] == "Provider Name" assert mock_update_metadata.call_args.kwargs["description"] == long_description # --------------------------------------------------------------------------- # fetch_provider_resource_keys # --------------------------------------------------------------------------- class TestFetchProviderResourceKeys: @pytest.mark.asyncio async def test_returns_ids_only(self): """Provider items matched by str(item.id), not item.name.""" adapter = AsyncMock() adapter.list.return_value = _mock_provider_view( [ _mock_item(item_id="rk-1", name="deploy-one"), _mock_item(item_id="rk-2", name="deploy-two"), ] ) from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys known_keys, provider_view = await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1", "rk-2"], ) assert known_keys == {"rk-1", "rk-2"} assert provider_view is adapter.list.return_value @pytest.mark.asyncio async def test_does_not_match_by_name(self): """If a name matches a resource_key but the id doesn't, it should NOT be included.""" adapter = AsyncMock() # Provider returns item with id="different" but name="rk-1" adapter.list.return_value = _mock_provider_view( [ _mock_item(item_id="different", name="rk-1"), ] ) from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys known_keys, provider_view = await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], ) # "rk-1" is NOT in the result because only id is checked, and id="different" assert "rk-1" not in known_keys assert known_keys == {"different"} assert provider_view is adapter.list.return_value @pytest.mark.asyncio async def test_empty_provider_id_raises_value_error(self): adapter = AsyncMock() adapter.list.return_value = _mock_provider_view( [ _mock_item(item_id="rk-1"), _mock_item(item_id=""), # falsy id ] ) from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys with pytest.raises(ValueError, match="empty id"): await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], ) @pytest.mark.asyncio async def test_empty_provider_response(self): adapter = AsyncMock() adapter.list.return_value = _mock_provider_view([]) from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys known_keys, provider_view = await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], ) assert known_keys == set() assert provider_view is adapter.list.return_value @pytest.mark.asyncio async def test_provider_error_raises_500_for_unhandled_exception(self): adapter = AsyncMock() adapter.list.side_effect = RuntimeError("provider down") from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys with pytest.raises(HTTPException) as exc_info: await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], ) assert exc_info.value.status_code == 500 @pytest.mark.asyncio async def test_provider_deployment_service_error_uses_mapped_http_status(self): adapter = AsyncMock() adapter.list.side_effect = ServiceUnavailableError("provider down") from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys with pytest.raises(HTTPException) as exc_info: await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], ) assert exc_info.value.status_code == 503 assert exc_info.value.detail == "provider down" @pytest.mark.asyncio async def test_passes_resource_keys_as_deployment_ids(self): adapter = AsyncMock() adapter.list.return_value = _mock_provider_view([]) from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys keys = ["rk-1", "rk-2", "rk-3"] await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=keys, ) call_args = adapter.list.call_args params = call_args.kwargs["params"] assert params.deployment_ids == keys assert params.provider_params is None assert params.deployment_types is None @pytest.mark.asyncio async def test_empty_resource_keys_returns_empty_set(self): adapter = AsyncMock() from langflow.api.v1.mappers.deployments.helpers import fetch_provider_resource_keys known_keys, provider_view = await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=[], ) assert known_keys == set() assert provider_view.deployments == [] adapter.list.assert_not_awaited() @pytest.mark.asyncio async def test_passes_deployment_type_filter(self): """When deployment_type is provided, it's forwarded to the provider.""" adapter = AsyncMock() adapter.list.return_value = _mock_provider_view([]) from langflow.api.v1.mappers.deployments.helpers import DeploymentType, fetch_provider_resource_keys await fetch_provider_resource_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), resource_keys=["rk-1"], deployment_type=DeploymentType("agent"), ) params = adapter.list.call_args.kwargs["params"] assert params.deployment_types == [DeploymentType("agent")] # --------------------------------------------------------------------------- # list_deployments_synced # --------------------------------------------------------------------------- class TestListDeploymentsSynced: @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=2) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_keeps_provider_confirmed_rows( self, mock_list, mock_fetch, mock_delete_unbound, mock_count_attachments, mock_count, ): """Rows whose resource_key is in the provider's known set are kept.""" owner_id = uuid4() row1 = _mock_deployment_row("rk-1", user_id=owner_id) row2 = _mock_deployment_row("rk-2", user_id=owner_id) fv_id = uuid4() mock_list.side_effect = [[(row1, 0, []), (row2, 1, [(fv_id, "snap-1")])], []] mock_fetch.return_value = ( {"rk-1", "rk-2"}, _mock_provider_view([SimpleNamespace(id="rk-1"), SimpleNamespace(id="rk-2")]), ) mock_count_attachments.return_value = {row1.id: 0, row2.id: 1} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, total, provider_data_by_resource_key = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_NoSnapshotBindingMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 2 assert accepted[0][0] is row1 assert accepted[1][0] is row2 assert total == 2 assert provider_data_by_resource_key == {} mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.delete_deployment_by_id", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_deletes_stale_rows( self, mock_list, mock_fetch, mock_delete, mock_delete_unbound, mock_count_attachments, mock_count, ): """Rows not recognised by the provider are deleted.""" uid = uuid4() stale_row = _mock_deployment_row("rk-stale", user_id=uid) good_row = _mock_deployment_row("rk-good", user_id=uid) db = _mock_async_db() # First call returns stale + good; second call returns empty (all consumed) mock_list.side_effect = [ [(stale_row, 0, []), (good_row, 1, [])], [], ] mock_fetch.return_value = ( {"rk-good"}, _mock_provider_view([SimpleNamespace(id="rk-good")]), ) # only rk-good is known mock_count_attachments.return_value = {good_row.id: 1} from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, provider_data_by_resource_key = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_NoSnapshotBindingMapper(), user_id=uid, provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 1 assert accepted[0][0] is good_row assert provider_data_by_resource_key == {} mock_delete.assert_awaited_once_with( db, user_id=uid, deployment_id=stale_row.id, ) mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=2) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.delete_deployment_by_id", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_merges_list_item_provider_data_across_refill_rounds( self, mock_list, mock_fetch, mock_delete, mock_delete_unbound, mock_count_attachments, mock_count, ): """Per-item provider_data is collected from each provider sync round.""" class _ProviderDataMapper(_NoSnapshotBindingMapper): def extract_list_item_provider_data(self, provider_view) -> dict[str, dict[str, Any]]: return {str(item.id): item.provider_data for item in provider_view.deployments} owner_id = uuid4() stale_row = _mock_deployment_row("rk-stale", user_id=owner_id) row1 = _mock_deployment_row("rk-1", user_id=owner_id) row2 = _mock_deployment_row("rk-2", user_id=owner_id) mock_list.side_effect = [ [(stale_row, 0, []), (row1, 0, [])], [(row2, 0, [])], ] mock_fetch.side_effect = [ ( {"rk-1"}, _mock_provider_view([SimpleNamespace(id="rk-1", provider_data={"environments": ["draft"]})]), ), ( {"rk-2"}, _mock_provider_view([SimpleNamespace(id="rk-2", provider_data={"environments": ["live"]})]), ), ] mock_count_attachments.return_value = {row1.id: 0, row2.id: 0} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, total, provider_data_by_resource_key = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_ProviderDataMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=2, deployment_type=None, ) assert [row.resource_key for row, _, _ in accepted] == ["rk-1", "rk-2"] assert total == 2 assert provider_data_by_resource_key == { "rk-1": {"environments": ["draft"]}, "rk-2": {"environments": ["live"]}, } mock_delete.assert_awaited_once() mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.update_deployment_metadata_batch", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_passes_provider_description_before_batch_metadata_sync( self, mock_list, mock_fetch, mock_update_metadata_batch, mock_delete_unbound, mock_count_attachments, mock_count, ): """Provider descriptions are passed through before local metadata batch sync.""" long_description = "x" * 501 class _MetadataMapper(_NoSnapshotBindingMapper): def extract_metadata_for_list(self, provider_view) -> dict[str, dict[str, Any]]: _ = provider_view return { "rk-1": { "display_name": "Provider Name", "description": long_description, } } row = _mock_deployment_row("rk-1") mock_list.return_value = [(row, 0, [])] mock_fetch.return_value = ({"rk-1"}, _mock_provider_view([SimpleNamespace(id="rk-1")])) mock_count_attachments.return_value = {row.id: 0} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, total, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_MetadataMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=1, deployment_type=None, ) assert [row.resource_key for row, _, _ in accepted] == ["rk-1"] assert total == 1 mock_update_metadata_batch.assert_awaited_once() updates = mock_update_metadata_batch.call_args.kwargs["deployment_updates"] assert updates[0].display_name == "Provider Name" assert updates[0].description == long_description mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_empty_batch_stops_loop(self, mock_list, mock_fetch, mock_count): """An empty batch from the DB ends the loop.""" mock_list.return_value = [] from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, total, provider_data_by_resource_key = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=BaseDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), page=1, size=10, deployment_type=None, ) assert accepted == [] assert total == 0 assert provider_data_by_resource_key == {} mock_fetch.assert_not_awaited() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=2) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.delete_deployment_by_id", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_mismatched_type_not_in_known_skips_without_deleting( self, mock_list, mock_fetch, mock_delete, mock_delete_unbound, mock_count_attachments, mock_count, ): """Row not in known but with a different local type is skipped, not deleted. The provider was filtered by type, so absence only means the type didn't match — the resource may still exist on the provider. """ row_match = _mock_deployment_row("rk-1", deployment_type="agent") row_other = _mock_deployment_row("rk-2", deployment_type="other") mock_list.side_effect = [ [(row_match, 0, []), (row_other, 0, [])], [], ] # Provider filtered by "agent" — only rk-1 returned mock_fetch.return_value = ({"rk-1"}, _mock_provider_view([SimpleNamespace(id="rk-1")])) mock_count_attachments.return_value = {row_match.id: 0} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import DeploymentType, list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_NoSnapshotBindingMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=DeploymentType("agent"), ) assert len(accepted) == 1 assert accepted[0][0] is row_match # rk-2 was NOT deleted — just skipped mock_delete.assert_not_awaited() mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.delete_deployment_by_id", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_cursor_does_not_advance_on_delete( self, mock_list, mock_fetch, mock_delete, mock_delete_unbound, mock_count_attachments, mock_count, ): """When a stale row is deleted the cursor stays put (offset doesn't increment).""" stale = _mock_deployment_row("rk-stale") good = _mock_deployment_row("rk-good") # First batch: stale row only. Cursor should stay at 0 for next fetch. # Second batch: good row at the same offset (because deletion shifted it). mock_list.side_effect = [ [(stale, 0, [])], [(good, 0, [])], [], ] mock_fetch.side_effect = [ (set(), None), # stale not known ({"rk-good"}, _mock_provider_view([SimpleNamespace(id="rk-good")])), # good is known ] mock_count_attachments.return_value = {good.id: 0} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_NoSnapshotBindingMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) # Both list_deployments_page calls should use offset=0 offsets = [call.kwargs["offset"] for call in mock_list.call_args_list[:2]] assert offsets == [0, 0], f"Expected cursor to stay at 0 after deletion, got offsets={offsets}" assert len(accepted) == 1 assert accepted[0][0] is good mock_delete.assert_awaited_once() mock_delete_unbound.assert_awaited_once() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=5) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_page_offset_calculation(self, mock_list, mock_fetch, mock_count): """Page 2 with size 5 should start at offset 5.""" mock_list.return_value = [] from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=BaseDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), page=2, size=5, deployment_type=None, ) assert mock_list.call_args.kwargs["offset"] == 5 mock_fetch.assert_not_awaited() mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.delete_deployment_by_id", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_stops_after_one_refill(self, mock_list, mock_fetch, mock_delete, mock_count): """Only two sync rounds run: initial fetch + one refill.""" stale = _mock_deployment_row("rk-stale") # Always return a stale row — the function should still stop after 2 rounds. mock_list.return_value = [(stale, 0, [])] mock_fetch.return_value = (set(), None) # never known from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=_NoSnapshotBindingMapper(), user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), page=1, size=2, deployment_type=None, ) assert accepted == [] assert mock_list.call_count == 2 assert mock_delete.await_count == 2 mock_count.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=3) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_passes_flow_version_ids_through(self, mock_list, mock_fetch, mock_count): """flow_version_ids are forwarded to list_deployments_page and count.""" mock_list.return_value = [] fv_ids = [uuid4(), uuid4()] from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=BaseDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), page=1, size=10, deployment_type=None, flow_version_ids=fv_ids, ) assert mock_list.call_args.kwargs["flow_version_ids"] == fv_ids assert mock_count.call_args.kwargs["flow_version_ids"] == fv_ids mock_fetch.assert_not_awaited() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_passes_project_id_through(self, mock_list, mock_fetch, mock_count): """project_id is forwarded to list_deployments_page and count.""" mock_list.return_value = [] project_id = uuid4() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=BaseDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), page=1, size=10, deployment_type=None, project_id=project_id, ) assert mock_list.call_args.kwargs["project_id"] == project_id assert mock_count.call_args.kwargs["project_id"] == project_id mock_fetch.assert_not_awaited() class _FakeMapper(BaseDeploymentMapper): def util_create_snapshot_bindings(self, *, result) -> CreateSnapshotBindings: # type: ignore[override] provider_result = result.provider_result if not isinstance(provider_result, dict): return CreateSnapshotBindings() snapshot_bindings = provider_result.get("snapshot_bindings") if not isinstance(snapshot_bindings, list): return CreateSnapshotBindings() resolved: list[CreateSnapshotBinding] = [] for binding in snapshot_bindings: if not isinstance(binding, dict): continue source_ref = str(binding.get("source_ref") or "").strip() snapshot_id = str(binding.get("snapshot_id") or "").strip() if source_ref and snapshot_id: resolved.append( CreateSnapshotBinding( source_ref=source_ref, snapshot_id=snapshot_id, ) ) return CreateSnapshotBindings(snapshot_bindings=resolved) def util_update_snapshot_bindings(self, *, result) -> UpdateSnapshotBindings: # type: ignore[override] provider_result = result.provider_result if not isinstance(provider_result, dict): return UpdateSnapshotBindings() snapshot_bindings = provider_result.get("added_snapshot_bindings") if not isinstance(snapshot_bindings, list): return UpdateSnapshotBindings() resolved: list[UpdateSnapshotBinding] = [] for binding in snapshot_bindings: if not isinstance(binding, dict): continue source_ref = str(binding.get("source_ref") or "").strip() snapshot_id = str(binding.get("snapshot_id") or "").strip() if source_ref and snapshot_id: resolved.append( UpdateSnapshotBinding( source_ref=source_ref, snapshot_id=snapshot_id, ) ) return UpdateSnapshotBindings(snapshot_bindings=resolved) class TestCreateSnapshotMapping: def test_resolve_snapshot_map_for_create_maps_source_ref_to_flow_version_ids(self): from langflow.api.v1.mappers.deployments.helpers import resolve_snapshot_map_for_create flow_version_ids = [uuid4(), uuid4()] resolved = resolve_snapshot_map_for_create( deployment_mapper=_FakeMapper(), result=DeploymentCreateResult( id="provider-id", type=DeploymentType.AGENT, provider_result={ "snapshot_bindings": [ {"source_ref": str(flow_version_ids[0]), "snapshot_id": "snap-1"}, {"source_ref": str(flow_version_ids[1]), "snapshot_id": "snap-2"}, ] }, ), flow_version_ids=flow_version_ids, ) assert resolved == { flow_version_ids[0]: "snap-1", flow_version_ids[1]: "snap-2", } def test_resolve_snapshot_map_for_create_rejects_missing_bindings(self): from langflow.api.v1.mappers.deployments.helpers import resolve_snapshot_map_for_create with pytest.raises(HTTPException, match="missing required snapshot bindings"): resolve_snapshot_map_for_create( deployment_mapper=_FakeMapper(), result=DeploymentCreateResult(id="provider-id", type=DeploymentType.AGENT), flow_version_ids=[uuid4()], ) class TestUpdateSnapshotMapping: def test_resolve_added_snapshot_bindings_for_update_maps_source_ref(self): from langflow.api.v1.mappers.deployments.helpers import resolve_added_snapshot_bindings_for_update flow_version_ids = [uuid4(), uuid4()] resolved = resolve_added_snapshot_bindings_for_update( deployment_mapper=_FakeMapper(), added_flow_version_ids=flow_version_ids, result=DeploymentUpdateResult( id="provider-id", provider_result={ "name": "agent_api_name", "added_snapshot_bindings": [ {"source_ref": str(flow_version_ids[0]), "snapshot_id": "snap-1"}, {"source_ref": str(flow_version_ids[1]), "snapshot_id": "snap-2"}, ], }, ), ) assert resolved == [ (flow_version_ids[0], "snap-1"), (flow_version_ids[1], "snap-2"), ] def test_resolve_added_snapshot_bindings_for_update_rejects_unexpected_source_ref(self): from langflow.api.v1.mappers.deployments.helpers import resolve_added_snapshot_bindings_for_update with pytest.raises(HTTPException, match="Unexpected source_ref in update snapshot bindings"): resolve_added_snapshot_bindings_for_update( deployment_mapper=_FakeMapper(), added_flow_version_ids=[uuid4()], result=DeploymentUpdateResult( id="provider-id", provider_result={ "name": "agent_api_name", "added_snapshot_bindings": [ {"source_ref": "other", "snapshot_id": "snap-extra"}, ], }, ), ) def test_resolve_added_snapshot_bindings_for_update_rejects_missing_expected_source_ref(self): from langflow.api.v1.mappers.deployments.helpers import resolve_added_snapshot_bindings_for_update with pytest.raises(HTTPException, match="Missing snapshot bindings for added flow versions"): resolve_added_snapshot_bindings_for_update( deployment_mapper=_FakeMapper(), added_flow_version_ids=[uuid4()], result=DeploymentUpdateResult( id="provider-id", provider_result={ "name": "agent_api_name", "added_snapshot_bindings": [], }, ), ) def test_resolve_snapshot_map_for_create_rejects_unexpected_source_ref(self): from langflow.api.v1.mappers.deployments.helpers import resolve_snapshot_map_for_create with pytest.raises(HTTPException, match="Unexpected source_ref"): resolve_snapshot_map_for_create( deployment_mapper=_FakeMapper(), result=DeploymentCreateResult( id="provider-id", type=DeploymentType.AGENT, provider_result={"snapshot_bindings": [{"source_ref": "other-ref", "snapshot_id": "snap-1"}]}, ), flow_version_ids=[uuid4()], ) def test_watsonx_mapper_util_created_snapshot_ids_uses_adapter_slot(): created = WatsonxOrchestrateDeploymentMapper().util_created_snapshot_ids( result=DeploymentUpdateResult( id="provider-id", provider_result={ "name": "agent_api_name", "display_name": "Agent API Name", "created_snapshot_ids": ["snap-1", "snap-2"], }, ), ) assert created.ids == ["snap-1", "snap-2"] def test_watsonx_mapper_extract_snapshot_bindings_from_provider_view(): mapper = WatsonxOrchestrateDeploymentMapper() provider_view = _mock_provider_view( [ SimpleNamespace(id="agent-1", provider_data=_wxo_deployment_provider_data(tool_ids=["tool-1", "tool-2"])), SimpleNamespace(id="agent-2", provider_data=_wxo_deployment_provider_data(tool_ids=["tool-3"])), ] ) bindings = mapper.extract_snapshot_bindings(provider_view) assert [(binding.resource_key, binding.snapshot_id) for binding in bindings] == [ ("agent-1", "tool-1"), ("agent-1", "tool-2"), ("agent-2", "tool-3"), ] def test_watsonx_mapper_extract_snapshot_bindings_requires_contract_fields(): mapper = WatsonxOrchestrateDeploymentMapper() provider_view = _mock_provider_view( [ SimpleNamespace(id="agent-1", provider_data={}), ] ) with pytest.raises(HTTPException) as exc_info: mapper.extract_snapshot_bindings(provider_view) assert exc_info.value.status_code == 500 assert "Missing required field 'display_name'" in str(exc_info.value.detail) def test_watsonx_mapper_extract_snapshot_bindings_requires_provider_data_object(): mapper = WatsonxOrchestrateDeploymentMapper() provider_view = _mock_provider_view( [ SimpleNamespace(id="agent-1", provider_data="bad-payload-type"), ] ) with pytest.raises(HTTPException) as exc_info: mapper.extract_snapshot_bindings(provider_view) assert exc_info.value.status_code == 500 assert "Unexpected result while reading deployment list item metadata" in str(exc_info.value.detail) def test_watsonx_mapper_extract_snapshot_bindings_allows_empty_tool_ids(): mapper = WatsonxOrchestrateDeploymentMapper() provider_view = _mock_provider_view( [ SimpleNamespace(id="agent-1", provider_data=_wxo_deployment_provider_data(tool_ids=[])), ] ) bindings = mapper.extract_snapshot_bindings(provider_view) assert bindings == [] def test_watsonx_mapper_extract_snapshot_bindings_for_get(): mapper = WatsonxOrchestrateDeploymentMapper() get_result = DeploymentGetResult( id="agent-1", name="agent-name", type="agent", provider_data={"tool_ids": ["tool-1", "tool-2"]}, ) bindings = mapper.extract_snapshot_bindings_for_get(get_result, resource_key="agent-1") assert [(binding.resource_key, binding.snapshot_id) for binding in bindings] == [ ("agent-1", "tool-1"), ("agent-1", "tool-2"), ] def test_watsonx_mapper_extract_snapshot_bindings_for_get_requires_tool_ids(): mapper = WatsonxOrchestrateDeploymentMapper() get_result = DeploymentGetResult( id="agent-1", name="agent-name", type="agent", provider_data={}, ) with pytest.raises( ValueError, match=r"^An internal error occured\. provider_data must contain 'tool_ids' from wxO adapter for get\(\)\.$", ): mapper.extract_snapshot_bindings_for_get(get_result, resource_key="agent-1") def test_watsonx_mapper_extract_snapshot_bindings_for_get_requires_provider_data(): mapper = WatsonxOrchestrateDeploymentMapper() get_result = DeploymentGetResult( id="agent-1", name="agent-name", type="agent", provider_data=None, ) with pytest.raises( ValueError, match=r"^An internal error occured\. provider_data is required from wxO adapter for get\(\)\.$", ): mapper.extract_snapshot_bindings_for_get(get_result, resource_key="agent-1") def test_watsonx_mapper_extract_snapshot_bindings_for_get_requires_tool_ids_list_type(): mapper = WatsonxOrchestrateDeploymentMapper() get_result = DeploymentGetResult( id="agent-1", name="agent-name", type="agent", provider_data={"tool_ids": "tool-1"}, ) with pytest.raises( ValueError, match=( r"^An internal error occured\. provider_data\['tool_ids'\] must be a list " r"from wxO adapter for get\(\)\.$" ), ): mapper.extract_snapshot_bindings_for_get(get_result, resource_key="agent-1") def test_watsonx_mapper_shape_deployment_get_data_hides_internal_sync_fields(): mapper = WatsonxOrchestrateDeploymentMapper() shaped = mapper.shape_deployment_get_data( { "llm": "my-llm", "display_name": "Agent Name", "description": "Provider description", "tool_ids": ["tool-1", "tool-2"], "environments": ["draft"], }, name="agent-name", ) assert shaped == { "llm": "my-llm", "name": "agent-name", "display_name": "Agent Name", "environments": ["draft"], } def test_watsonx_mapper_shape_deployment_get_data_requires_provider_data(): mapper = WatsonxOrchestrateDeploymentMapper() with pytest.raises(HTTPException) as exc_info: mapper.shape_deployment_get_data(None) assert exc_info.value.status_code == 500 assert "Empty result while reading deployment metadata" in str(exc_info.value.detail) def test_base_mapper_extract_snapshot_bindings_for_get_raises_not_implemented(): """Locks in the destructive-deletion safety contract for GET-sync. The base implementation MUST raise ``NotImplementedError`` rather than return ``[]``. An empty list would flow into ``delete_unbound_attachments`` (which treats empty ``bindings`` plus non-empty ``deployment_ids`` as a wipe of every local attachment for the GETted deployment). Raising prevents that destructive interpretation for any provider that hasn't overridden the method — the GET call site treats ``NotImplementedError`` as a mapper implementation error instead of wiping local attachment state. """ mapper = BaseDeploymentMapper() get_result = DeploymentGetResult( id="agent-1", name="agent-name", type="agent", provider_data={"tool_ids": ["tool-1"]}, ) with pytest.raises(NotImplementedError, match="not configured for syncing snapshots for a deployment"): mapper.extract_snapshot_bindings_for_get(get_result, resource_key="agent-1") def test_base_mapper_extract_snapshot_bindings_raises_not_implemented(): """Locks in the destructive-deletion safety contract for list-sync. The base implementation MUST raise ``NotImplementedError`` rather than return ``[]``. An empty list would flow into ``delete_unbound_attachments`` (which treats empty ``bindings`` plus non-empty ``deployment_ids`` as a wipe of every local attachment for those deployments). Raising prevents that destructive interpretation for any provider that hasn't overridden the method. """ mapper = BaseDeploymentMapper() provider_view = _mock_provider_view( [ SimpleNamespace(id="agent-1", provider_data={"tool_ids": ["tool-1"]}), ] ) with pytest.raises(NotImplementedError, match="not configured for syncing snapshots for multiple deployments"): mapper.extract_snapshot_bindings(provider_view) def test_resolve_flow_version_patch_for_update_watsonx_operations(): from langflow.api.v1.mappers.deployments.helpers import resolve_flow_version_patch_for_update add_id = uuid4() unbind_only_id = uuid4() remove_id = uuid4() add_ids, remove_ids = resolve_flow_version_patch_for_update( deployment_mapper=WatsonxOrchestrateDeploymentMapper(), payload=DeploymentUpdateRequest( provider_data={ "llm": "test-llm", "upsert_flows": [ { "flow_version_id": str(add_id), "add_app_ids": ["app-one"], "remove_app_ids": [], }, { "flow_version_id": str(unbind_only_id), "add_app_ids": [], "remove_app_ids": ["app-one"], }, ], "remove_flows": [str(remove_id)], } ), ) assert add_ids == [add_id] assert remove_ids == [remove_id] class _FakeCountExecResult: def __init__(self, count_value: int): self._count_value = count_value def one(self): return self._count_value class _FakeCountDb: def __init__(self, count_value: int): self._count_value = count_value async def exec(self, _statement): return _FakeCountExecResult(self._count_value) @pytest.mark.asyncio async def test_validate_project_scoped_flow_version_ids_accepts_all_ids(): from langflow.api.v1.mappers.deployments.helpers import validate_project_scoped_flow_version_ids flow_version_id = uuid4() await validate_project_scoped_flow_version_ids( flow_version_ids=[flow_version_id, flow_version_id], user_id=uuid4(), project_id=uuid4(), db=_FakeCountDb(count_value=1), # type: ignore[arg-type] ) @pytest.mark.asyncio async def test_validate_project_scoped_flow_version_ids_rejects_out_of_project_ids(): from langflow.api.v1.mappers.deployments.helpers import validate_project_scoped_flow_version_ids with pytest.raises(HTTPException, match="selected project"): await validate_project_scoped_flow_version_ids( flow_version_ids=[uuid4(), uuid4()], user_id=uuid4(), project_id=uuid4(), db=_FakeCountDb(count_value=1), # type: ignore[arg-type] ) # --------------------------------------------------------------------------- # fetch_provider_snapshot_keys # --------------------------------------------------------------------------- def _mock_snapshot_item(*, item_id: str, name: str = "irrelevant") -> SimpleNamespace: return SimpleNamespace(id=item_id, name=name) def _mock_snapshot_view(items: list) -> SimpleNamespace: return SimpleNamespace(snapshots=items) class TestFetchProviderSnapshotKeys: @pytest.mark.asyncio async def test_returns_recognized_snapshot_ids(self): adapter = AsyncMock() adapter.list_snapshots.return_value = _mock_snapshot_view( [_mock_snapshot_item(item_id="snap-1"), _mock_snapshot_item(item_id="snap-2")] ) from langflow.api.v1.mappers.deployments.sync import fetch_provider_snapshot_keys result = await fetch_provider_snapshot_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), snapshot_ids=["snap-1", "snap-2", "snap-3"], ) assert result == {"snap-1", "snap-2"} @pytest.mark.asyncio async def test_empty_snapshot_ids_returns_empty_set(self): adapter = AsyncMock() from langflow.api.v1.mappers.deployments.sync import fetch_provider_snapshot_keys result = await fetch_provider_snapshot_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), snapshot_ids=[], ) assert result == set() adapter.list_snapshots.assert_not_awaited() @pytest.mark.asyncio async def test_provider_error_raises_500(self): adapter = AsyncMock() adapter.list_snapshots.side_effect = RuntimeError("provider down") from langflow.api.v1.mappers.deployments.sync import fetch_provider_snapshot_keys with pytest.raises(HTTPException) as exc_info: await fetch_provider_snapshot_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), snapshot_ids=["snap-1"], ) assert exc_info.value.status_code == 500 @pytest.mark.asyncio async def test_falsy_snapshot_id_from_provider_raises_value_error(self): adapter = AsyncMock() adapter.list_snapshots.return_value = _mock_snapshot_view([_mock_snapshot_item(item_id="")]) from langflow.api.v1.mappers.deployments.sync import fetch_provider_snapshot_keys with pytest.raises(ValueError, match="snapshot with an empty id"): await fetch_provider_snapshot_keys( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), db=AsyncMock(), snapshot_ids=["snap-1"], ) # --------------------------------------------------------------------------- # sync_attachment_snapshot_ids # --------------------------------------------------------------------------- def _mock_attachment(*, flow_version_id=None, provider_snapshot_id=None, deployment_id=None) -> SimpleNamespace: return SimpleNamespace( flow_version_id=flow_version_id or uuid4(), provider_snapshot_id=provider_snapshot_id, deployment_id=deployment_id or uuid4(), ) class TestSyncAttachmentSnapshotIds: @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployment_attachments_by_keys", new_callable=AsyncMock) async def test_deletes_stale_attachments(self, mock_delete_att): """Attachments with snapshot IDs not in the known set are deleted.""" dep_id = uuid4() uid = uuid4() fv_good = uuid4() fv_stale_a = uuid4() fv_stale_b = uuid4() attachments = [ _mock_attachment(flow_version_id=fv_good, provider_snapshot_id="snap-good", deployment_id=dep_id), _mock_attachment(flow_version_id=fv_stale_a, provider_snapshot_id="snap-stale", deployment_id=dep_id), _mock_attachment(flow_version_id=fv_stale_b, provider_snapshot_id="snap-stale", deployment_id=dep_id), ] from langflow.api.v1.mappers.deployments.sync import sync_attachment_snapshot_ids counts = await sync_attachment_snapshot_ids( user_id=uid, attachments=attachments, known_snapshot_ids={"snap-good"}, db=AsyncMock(), ) assert counts[dep_id] == 1 mock_delete_att.assert_awaited_once() call_kwargs = mock_delete_att.call_args.kwargs assert call_kwargs["user_id"] == uid key_batch = call_kwargs["attachment_key_batch"] key_pairs = [(key.deployment_id, key.flow_version_id) for key in key_batch.keys] assert key_pairs == [ (dep_id, fv_stale_a), (dep_id, fv_stale_b), ] @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployment_attachments_by_keys", new_callable=AsyncMock) async def test_raises_on_falsy_snapshot_id(self, mock_delete_att): """Attachments with empty provider_snapshot_id raise data-integrity errors.""" dep_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id=None, deployment_id=dep_id), _mock_attachment(provider_snapshot_id="", deployment_id=dep_id), ] from langflow.api.v1.mappers.deployments.sync import sync_attachment_snapshot_ids with pytest.raises(ValueError, match="provider_snapshot_id must be non-empty"): await sync_attachment_snapshot_ids( user_id=uuid4(), attachments=attachments, known_snapshot_ids=set(), db=AsyncMock(), ) mock_delete_att.assert_not_awaited() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployment_attachments_by_keys", new_callable=AsyncMock) async def test_falsy_snapshot_short_circuits_before_batch_delete(self, mock_delete_att): """If any attachment has a falsy snapshot id, sync aborts before deleting stale keys.""" dep_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id="snap-known", deployment_id=dep_id), _mock_attachment(provider_snapshot_id="snap-stale", deployment_id=dep_id), _mock_attachment(provider_snapshot_id=" ", deployment_id=dep_id), ] from langflow.api.v1.mappers.deployments.sync import sync_attachment_snapshot_ids with pytest.raises(ValueError, match="provider_snapshot_id must be non-empty"): await sync_attachment_snapshot_ids( user_id=uuid4(), attachments=attachments, known_snapshot_ids={"snap-known"}, db=AsyncMock(), ) mock_delete_att.assert_not_awaited() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployment_attachments_by_keys", new_callable=AsyncMock) async def test_all_snapshots_known_nothing_deleted(self, mock_delete_att): dep_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=dep_id), _mock_attachment(provider_snapshot_id="snap-2", deployment_id=dep_id), ] from langflow.api.v1.mappers.deployments.sync import sync_attachment_snapshot_ids counts = await sync_attachment_snapshot_ids( user_id=uuid4(), attachments=attachments, known_snapshot_ids={"snap-1", "snap-2"}, db=AsyncMock(), ) assert counts[dep_id] == 2 mock_delete_att.assert_not_awaited() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployment_attachments_by_keys", new_callable=AsyncMock) async def test_extra_known_provider_snapshots_are_ignored(self, mock_delete_att): dep_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=dep_id), _mock_attachment(provider_snapshot_id="snap-2", deployment_id=dep_id), ] from langflow.api.v1.mappers.deployments.sync import sync_attachment_snapshot_ids counts = await sync_attachment_snapshot_ids( user_id=uuid4(), attachments=attachments, known_snapshot_ids={"snap-1", "snap-2", "snap-extra"}, db=AsyncMock(), ) assert counts[dep_id] == 2 mock_delete_att.assert_not_awaited() # --------------------------------------------------------------------------- # sync_flow_version_attachments # --------------------------------------------------------------------------- class TestSyncFlowVersionAttachments: @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.get_deployment_adapter") @patch("langflow.api.v1.mappers.deployments.registry.get_deployment_mapper") @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_snapshot_cleanup_runs_inside_savepoint( self, mock_list_rows, mock_get_mapper, mock_get_adapter, mock_fetch_resource_keys, mock_delete_unbound, ): """Binding cleanup is isolated so a failed best-effort sync cannot dirty the outer transaction.""" deployment = _mock_deployment_row("rk-1") provider_view = _mock_provider_view([_mock_item(item_id="rk-1")]) mock_list_rows.return_value = [(deployment, "watsonx-orchestrate")] mapper = MagicMock() mapper.extract_snapshot_bindings.return_value = [] mock_get_mapper.return_value = mapper mock_get_adapter.return_value = AsyncMock() mock_fetch_resource_keys.return_value = ({"rk-1"}, provider_view) mock_delete_unbound.side_effect = RuntimeError("delete failed") db = _mock_async_db() from langflow.api.v1.mappers.deployments.sync import sync_flow_version_attachments await sync_flow_version_attachments( db=db, flow_id=uuid4(), user_id=uuid4(), ) db.begin_nested.assert_called_once() mock_delete_unbound.assert_awaited_once() # --------------------------------------------------------------------------- # _sync_deployments_and_attachments_by_provider # --------------------------------------------------------------------------- class TestSyncDeploymentsAndAttachmentsByProvider: @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_deployments_by_ids", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.get_deployment_adapter") @patch("langflow.api.v1.mappers.deployments.registry.get_deployment_mapper") async def test_batches_stale_deletions_once_per_provider_group( self, mock_get_mapper, mock_get_adapter, mock_fetch_resource_keys, mock_delete_unbound, mock_delete_deployments, ): provider_account_id = uuid4() stale_a = SimpleNamespace( id=uuid4(), resource_key="rk-stale-a", deployment_provider_account_id=provider_account_id, ) stale_b = SimpleNamespace( id=uuid4(), resource_key="rk-stale-b", deployment_provider_account_id=provider_account_id, ) surviving = SimpleNamespace( id=uuid4(), resource_key="rk-live", deployment_provider_account_id=provider_account_id, ) provider_view = _mock_provider_view([_mock_item(item_id="rk-live")]) mock_get_adapter.return_value = AsyncMock() mock_fetch_resource_keys.return_value = ({"rk-live"}, provider_view) mapper = MagicMock() mapper.extract_snapshot_bindings.return_value = ["binding-1"] mock_get_mapper.return_value = mapper db = _mock_async_db() from langflow.api.v1.mappers.deployments.sync import _sync_deployments_and_attachments_by_provider await _sync_deployments_and_attachments_by_provider( db=db, user_id=uuid4(), deployments_with_provider=[ (stale_a, "watsonx-orchestrate"), (stale_b, "watsonx-orchestrate"), (surviving, "watsonx-orchestrate"), ], stale_scope_label="flow", failure_log_message="ignored", failure_scope_value=uuid4(), ) mock_delete_deployments.assert_awaited_once() assert set(mock_delete_deployments.await_args.kwargs["deployment_ids"]) == {stale_a.id, stale_b.id} mock_delete_unbound.assert_awaited_once() assert mock_delete_unbound.await_args.kwargs["deployment_ids"] == [surviving.id] assert mock_delete_unbound.await_args.kwargs["bindings"] == ["binding-1"] # --------------------------------------------------------------------------- # provider-account scoped sync entry points # --------------------------------------------------------------------------- class TestProviderAccountScopedSync: @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_deployment_state_returns_early_when_flow_ids_empty( self, mock_list_deployments, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_deployment_state await sync_flow_deployment_state( db=AsyncMock(), flow_ids=[], user_id=uuid4(), ) mock_delete_orphans.assert_not_awaited() mock_list_deployments.assert_not_awaited() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_deployment_state_deduplicates_flow_ids( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_deployment_state flow_id = uuid4() mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_flow_deployment_state( db=AsyncMock(), flow_ids=[flow_id, flow_id], user_id=uuid4(), ) assert mock_delete_orphans.await_args.kwargs["flow_ids"] == [flow_id] assert mock_list_deployments.await_args.kwargs["flow_ids"] == [flow_id] mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_project", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_project_deployments_with_provider_info", new_callable=AsyncMock, return_value=[]) async def test_sync_project_deployments_returns_when_no_deployments( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_project_deployments project_id = uuid4() await sync_project_deployments( db=AsyncMock(), project_id=project_id, user_id=uuid4(), ) mock_delete_orphans.assert_awaited_once() mock_list_deployments.assert_awaited_once() mock_sync_by_provider.assert_not_awaited() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_deployment_state_passes_provider_account_filter( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_deployment_state provider_account_id = uuid4() mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_flow_deployment_state( db=AsyncMock(), flow_ids=[uuid4()], user_id=uuid4(), deployment_provider_account_id=provider_account_id, ) assert mock_list_deployments.call_args.kwargs["provider_account_id"] == provider_account_id mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_version_attachments_passes_provider_account_filter( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_version_attachments provider_account_id = uuid4() mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_flow_version_attachments( db=AsyncMock(), flow_id=uuid4(), user_id=uuid4(), deployment_provider_account_id=provider_account_id, ) assert mock_list_deployments.call_args.kwargs["provider_account_id"] == provider_account_id mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_project", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_project_deployments_with_provider_info", new_callable=AsyncMock) async def test_sync_project_deployments_passes_provider_account_filter( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_project_deployments provider_account_id = uuid4() mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_project_deployments( db=AsyncMock(), project_id=uuid4(), user_id=uuid4(), deployment_provider_account_id=provider_account_id, ) assert mock_list_deployments.call_args.kwargs["provider_account_id"] == provider_account_id mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock, side_effect=Exception("db")) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_deployment_state_continues_when_orphan_cleanup_fails( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_deployment_state mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_flow_deployment_state( db=AsyncMock(), flow_ids=[uuid4()], user_id=uuid4(), ) mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_flow_ids", new_callable=AsyncMock, side_effect=Exception("db")) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_deployments_for_flows_with_provider_info", new_callable=AsyncMock) async def test_sync_flow_version_attachments_continues_when_orphan_cleanup_fails( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_flow_version_attachments mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_flow_version_attachments( db=AsyncMock(), flow_id=uuid4(), user_id=uuid4(), ) mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() @pytest.mark.asyncio @patch(f"{SYNC_MODULE}.delete_orphan_attachments_for_project", new_callable=AsyncMock, side_effect=Exception("db")) @patch(f"{SYNC_MODULE}._sync_deployments_and_attachments_by_provider", new_callable=AsyncMock) @patch(f"{SYNC_MODULE}.list_project_deployments_with_provider_info", new_callable=AsyncMock) async def test_sync_project_deployments_continues_when_orphan_cleanup_fails( self, mock_list_deployments, mock_sync_by_provider, mock_delete_orphans, ): from langflow.api.v1.mappers.deployments.sync import sync_project_deployments mock_list_deployments.return_value = [(_mock_deployment_row("rk-1"), "watsonx-orchestrate")] await sync_project_deployments( db=AsyncMock(), project_id=uuid4(), user_id=uuid4(), ) mock_delete_orphans.assert_awaited_once() mock_sync_by_provider.assert_awaited_once() # --------------------------------------------------------------------------- # rollback_provider_create # --------------------------------------------------------------------------- class TestRollbackProviderCreate: @pytest.mark.asyncio async def test_calls_adapter_delete(self): """Compensating delete is called with the provider resource ID.""" adapter = AsyncMock() from langflow.api.v1.mappers.deployments.helpers import rollback_provider_create await rollback_provider_create( deployment_adapter=adapter, provider_id=uuid4(), resource_id="provider-resource-123", user_id=uuid4(), db=AsyncMock(), ) adapter.delete.assert_awaited_once() assert adapter.delete.call_args.kwargs["deployment_id"] == "provider-resource-123" @pytest.mark.asyncio async def test_uses_extended_provider_result_rollback_when_available(self): """Adapters can clean up create-time side resources before falling back to delete.""" adapter = AsyncMock() from langflow.api.v1.mappers.deployments.helpers import rollback_provider_create provider_result = {"app_ids": ["app-1"], "tools_with_refs": [{"tool_id": "tool-1", "source_ref": "fv-1"}]} await rollback_provider_create( deployment_adapter=adapter, provider_id=uuid4(), resource_id="provider-resource-123", provider_result=provider_result, user_id=uuid4(), db=AsyncMock(), ) adapter.rollback_create_result.assert_awaited_once() assert adapter.rollback_create_result.call_args.kwargs["provider_result"] == provider_result adapter.delete.assert_not_awaited() @pytest.mark.asyncio async def test_falls_back_to_delete_when_extended_rollback_fails(self): """If the extended cleanup path fails, the helper still attempts the basic delete.""" adapter = AsyncMock() adapter.rollback_create_result.side_effect = RuntimeError("rollback boom") from langflow.api.v1.mappers.deployments.helpers import rollback_provider_create await rollback_provider_create( deployment_adapter=adapter, provider_id=uuid4(), resource_id="provider-resource-123", provider_result={"app_ids": ["app-1"]}, user_id=uuid4(), db=AsyncMock(), ) adapter.rollback_create_result.assert_awaited_once() adapter.delete.assert_awaited_once() @pytest.mark.asyncio async def test_swallows_delete_failure(self): """If the compensating delete fails, exception is swallowed (logged).""" adapter = AsyncMock() adapter.delete.side_effect = RuntimeError("provider unreachable") from langflow.api.v1.mappers.deployments.helpers import rollback_provider_create # Should not raise await rollback_provider_create( deployment_adapter=adapter, provider_id=uuid4(), resource_id="orphaned-resource", user_id=uuid4(), db=AsyncMock(), ) @pytest.mark.asyncio async def test_skips_delete_fallback_when_disabled_and_extended_rollback_fails(self): """When delete fallback is disabled, extended rollback failures do not trigger delete.""" adapter = AsyncMock() adapter.rollback_create_result.side_effect = RuntimeError("rollback boom") from langflow.api.v1.mappers.deployments.helpers import rollback_provider_create await rollback_provider_create( deployment_adapter=adapter, provider_id=uuid4(), resource_id="existing-resource-1", provider_result={"app_ids": ["app-1"]}, allow_delete_fallback=False, user_id=uuid4(), db=AsyncMock(), ) adapter.rollback_create_result.assert_awaited_once() adapter.delete.assert_not_awaited() # --------------------------------------------------------------------------- # rollback_provider_update # --------------------------------------------------------------------------- class TestRollbackProviderUpdate: @pytest.mark.asyncio async def test_returns_early_when_mapper_returns_none(self): """When the mapper cannot build a rollback payload, no adapter call is made.""" adapter = AsyncMock() mapper = AsyncMock(spec=BaseDeploymentMapper) mapper.resolve_rollback_update = AsyncMock(return_value=None) dep_row = _mock_deployment_row("rk-1") from langflow.api.v1.mappers.deployments.helpers import rollback_provider_update await rollback_provider_update( deployment_adapter=adapter, deployment_mapper=mapper, deployment_db_id=dep_row.id, deployment_resource_key=dep_row.resource_key, deployment_provider_account_id=dep_row.deployment_provider_account_id, user_id=uuid4(), db=AsyncMock(), ) adapter.update.assert_not_awaited() @pytest.mark.asyncio async def test_calls_adapter_update_with_rollback_payload(self): """When mapper returns a payload, compensating update is issued.""" adapter = AsyncMock() rollback_payload = MagicMock() mapper = AsyncMock(spec=BaseDeploymentMapper) mapper.resolve_rollback_update = AsyncMock(return_value=rollback_payload) dep_row = _mock_deployment_row("rk-1") dep_row.deployment_provider_account_id = uuid4() from langflow.api.v1.mappers.deployments.helpers import rollback_provider_update await rollback_provider_update( deployment_adapter=adapter, deployment_mapper=mapper, deployment_db_id=dep_row.id, deployment_resource_key=dep_row.resource_key, deployment_provider_account_id=dep_row.deployment_provider_account_id, user_id=uuid4(), db=AsyncMock(), ) adapter.update.assert_awaited_once() assert adapter.update.call_args.kwargs["payload"] is rollback_payload @pytest.mark.asyncio async def test_swallows_adapter_update_failure(self): """If the compensating update fails, exception is swallowed.""" adapter = AsyncMock() adapter.update.side_effect = RuntimeError("provider error") mapper = AsyncMock(spec=BaseDeploymentMapper) mapper.resolve_rollback_update = AsyncMock(return_value=MagicMock()) dep_row = _mock_deployment_row("rk-1") dep_row.deployment_provider_account_id = uuid4() from langflow.api.v1.mappers.deployments.helpers import rollback_provider_update # Should not raise await rollback_provider_update( deployment_adapter=adapter, deployment_mapper=mapper, deployment_db_id=dep_row.id, deployment_resource_key=dep_row.resource_key, deployment_provider_account_id=dep_row.deployment_provider_account_id, user_id=uuid4(), db=AsyncMock(), ) @pytest.mark.asyncio async def test_swallows_mapper_failure(self): """If mapper.resolve_rollback_update raises, exception is swallowed.""" adapter = AsyncMock() mapper = AsyncMock(spec=BaseDeploymentMapper) mapper.resolve_rollback_update = AsyncMock(side_effect=RuntimeError("mapper error")) dep_row = _mock_deployment_row("rk-1") from langflow.api.v1.mappers.deployments.helpers import rollback_provider_update # Should not raise await rollback_provider_update( deployment_adapter=adapter, deployment_mapper=mapper, deployment_db_id=dep_row.id, deployment_resource_key=dep_row.resource_key, deployment_provider_account_id=dep_row.deployment_provider_account_id, user_id=uuid4(), db=AsyncMock(), ) adapter.update.assert_not_awaited() # --------------------------------------------------------------------------- # WatsonxOrchestrateDeploymentMapper.resolve_rollback_update # --------------------------------------------------------------------------- DEP_CRUD_MODULE = "langflow.services.database.models.deployment.crud" ATT_CRUD_MODULE = "langflow.services.database.models.flow_version_deployment_attachment.crud" class TestWxoResolveRollbackUpdate: @pytest.mark.asyncio @patch(f"{ATT_CRUD_MODULE}.list_deployment_attachments", new_callable=AsyncMock) @patch(f"{DEP_CRUD_MODULE}.get_deployment", new_callable=AsyncMock) async def test_returns_none_when_deployment_not_found(self, mock_get_dep, mock_list_att): """If the deployment row no longer exists, returns None.""" mock_get_dep.return_value = None mapper = WatsonxOrchestrateDeploymentMapper() result = await mapper.resolve_rollback_update( user_id=uuid4(), deployment_db_id=uuid4(), deployment_resource_key="rk-1", db=AsyncMock(), ) assert result is None mock_list_att.assert_not_awaited() @pytest.mark.asyncio @patch(f"{ATT_CRUD_MODULE}.list_deployment_attachments", new_callable=AsyncMock) @patch(f"{DEP_CRUD_MODULE}.get_deployment", new_callable=AsyncMock) async def test_builds_put_tools_from_attachment_snapshot_ids(self, mock_get_dep, mock_list_att): """Rollback payload contains put_tools with provider_snapshot_ids from DB attachments. att3/att4 simulate legacy rows that may still have NULL/blank snapshot IDs in the DB from before the non-empty invariant was enforced at the app level. """ dep = MagicMock() dep.display_name = "test-dep" dep.description = "desc" mock_get_dep.return_value = dep att1 = MagicMock() att1.provider_snapshot_id = "tool-1" att2 = MagicMock() att2.provider_snapshot_id = "tool-2" att3 = MagicMock() att3.provider_snapshot_id = None att4 = MagicMock() att4.provider_snapshot_id = " " mock_list_att.return_value = [att1, att2, att3, att4] mapper = WatsonxOrchestrateDeploymentMapper() result = await mapper.resolve_rollback_update( user_id=uuid4(), deployment_db_id=uuid4(), deployment_resource_key="rk-1", db=AsyncMock(), ) assert result is not None assert result.spec is not None assert "name" not in result.spec.model_fields_set assert result.spec.description == "desc" provider_data = result.provider_data assert provider_data["put_tools"] == ["tool-1", "tool-2"] assert provider_data["display_name"] == "test-dep" @pytest.mark.asyncio @patch(f"{ATT_CRUD_MODULE}.list_deployment_attachments", new_callable=AsyncMock) @patch(f"{DEP_CRUD_MODULE}.get_deployment", new_callable=AsyncMock) async def test_empty_attachments_produces_empty_put_tools(self, mock_get_dep, mock_list_att): """When no attachments exist, put_tools is an empty list (clears all tools).""" dep = MagicMock() dep.display_name = "test-dep" dep.description = " " mock_get_dep.return_value = dep mock_list_att.return_value = [] mapper = WatsonxOrchestrateDeploymentMapper() result = await mapper.resolve_rollback_update( user_id=uuid4(), deployment_db_id=uuid4(), deployment_resource_key="rk-1", db=AsyncMock(), ) assert result is not None assert result.provider_data["put_tools"] == [] assert result.provider_data["display_name"] == "test-dep" assert result.spec is not None assert result.spec.description is None # --------------------------------------------------------------------------- # list_deployments_synced: phase-2 binding sync # --------------------------------------------------------------------------- class TestListDeploymentsSyncedBindingPhase: @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_binding_sync_corrects_attached_count( self, mock_list, mock_fetch_rk, mock_delete_unbound, mock_count_attachments, mock_count, # noqa: ARG002 ): """Phase 2 corrects attached_count using binding-level cleanup + recount.""" row = _mock_deployment_row("rk-1") mock_list.side_effect = [[(row, 3, [])], []] mock_fetch_rk.return_value = ( {"rk-1"}, _mock_provider_view( [ SimpleNamespace( id="rk-1", name="agent_api_name", provider_data={ "display_name": "Agent One", "description": "Provider description", "tool_ids": ["snap-1"], "llm": "meta-llm", "environments": [], }, ) ] ), ) mock_count_attachments.return_value = {row.id: 2} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, provider_data_by_resource_key = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=WatsonxOrchestrateDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 1 assert accepted[0][1] == 2 assert provider_data_by_resource_key == { "rk-1": { "name": "agent_api_name", "display_name": "Agent One", "environments": [], "llm": "meta-llm", } } db.begin_nested.assert_called_once() mock_delete_unbound.assert_awaited_once() mock_count_attachments.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_binding_cleanup_runs_with_empty_bindings( self, mock_list, mock_fetch_rk, mock_delete_unbound, mock_count_attachments, mock_count, # noqa: ARG002 ): """Phase 2 still runs when mapper extracts no bindings.""" row = _mock_deployment_row("rk-1") mock_list.side_effect = [[(row, 0, [])], []] mock_fetch_rk.return_value = ( {"rk-1"}, _mock_provider_view( [SimpleNamespace(id="rk-1", name="Agent 1", provider_data=_wxo_deployment_provider_data(tool_ids=[]))] ), ) mock_count_attachments.return_value = {row.id: 0} db = _mock_async_db() provider_id = uuid4() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=WatsonxOrchestrateDeploymentMapper(), user_id=uuid4(), provider_id=provider_id, db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 1 mock_delete_unbound.assert_awaited_once() assert mock_delete_unbound.await_args.kwargs["bindings"] == [] assert mock_delete_unbound.await_args.kwargs["provider_account_id"] == provider_id mock_count_attachments.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_binding_sync_error_preserves_original_counts( self, mock_list, mock_fetch_rk, mock_delete_unbound, mock_count_attachments, mock_count, # noqa: ARG002 ): """When Phase 2 raises, accepted rows keep their original attached_count.""" row = _mock_deployment_row("rk-1") mock_list.side_effect = [[(row, 3, [])], []] mock_fetch_rk.return_value = ( {"rk-1"}, _mock_provider_view( [ SimpleNamespace( id="rk-1", name="Agent 1", provider_data=_wxo_deployment_provider_data(tool_ids=["snap-1"]), ) ] ), ) mock_delete_unbound.side_effect = RuntimeError("cleanup failed") db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=WatsonxOrchestrateDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 1 assert accepted[0][1] == 3 mock_count_attachments.assert_not_awaited() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployments_by_provider", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.count_attachments_by_deployment_ids", new_callable=AsyncMock) @patch(f"{MODULE}.delete_unbound_attachments", new_callable=AsyncMock) @patch(f"{MODULE}.fetch_provider_resource_keys", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployments_page", new_callable=AsyncMock) async def test_binding_cleanup_runs_inside_savepoint( self, mock_list, mock_fetch_rk, mock_delete_unbound, mock_count_attachments, mock_count, # noqa: ARG002 ): """Phase 2 cleanup uses a savepoint so failures don't leak partial writes.""" row = _mock_deployment_row("rk-1") mock_list.side_effect = [[(row, 3, [])], []] mock_fetch_rk.return_value = ( {"rk-1"}, _mock_provider_view( [ SimpleNamespace( id="rk-1", name="Agent 1", provider_data=_wxo_deployment_provider_data(tool_ids=["snap-1"]), ) ] ), ) mock_count_attachments.return_value = {row.id: 3} db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployments_synced accepted, _, _ = await list_deployments_synced( deployment_adapter=AsyncMock(), deployment_mapper=WatsonxOrchestrateDeploymentMapper(), user_id=uuid4(), provider_id=uuid4(), db=db, page=1, size=10, deployment_type=None, ) assert len(accepted) == 1 assert accepted[0][1] == 3 db.begin_nested.assert_called_once() mock_delete_unbound.assert_awaited_once() # --------------------------------------------------------------------------- # list_deployment_flow_versions_synced # --------------------------------------------------------------------------- class TestListDeploymentFlowVersionsSynced: @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=2) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock) @patch(f"{MODULE}.sync_attachment_snapshot_ids", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock) async def test_syncs_snapshots_with_snapshot_ids_and_returns_enrichment( self, mock_list_attachments, mock_sync_snapshot_ids, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=deployment_id), _mock_attachment(provider_snapshot_id="snap-stale", deployment_id=deployment_id), ] mock_list_attachments.return_value = attachments rows = [(SimpleNamespace(provider_snapshot_id="snap-1"), SimpleNamespace())] mock_list_with_versions.return_value = rows adapter = AsyncMock() adapter.list_snapshots.return_value = SnapshotListResult( snapshots=[ SnapshotItem( id="snap-1", name="tool-1", provider_data={"connections": {"cfg-1": "conn-1"}}, ) ] ) db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced out_rows, total, snapshot_result = await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=db, page=2, size=3, ) assert out_rows == rows assert total == 2 assert isinstance(snapshot_result, SnapshotListResult) assert [snapshot.id for snapshot in snapshot_result.snapshots] == ["snap-1"] adapter.list_snapshots.assert_awaited_once() adapter_params = adapter.list_snapshots.call_args.kwargs["params"] assert adapter_params.snapshot_ids == ["snap-1", "snap-stale"] mock_sync_snapshot_ids.assert_awaited_once() assert mock_sync_snapshot_ids.call_args.kwargs["known_snapshot_ids"] == {"snap-1"} assert mock_list_with_versions.call_args.kwargs["offset"] == 3 assert mock_list_with_versions.call_args.kwargs["limit"] == 3 mock_count_attachments.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock) @patch(f"{MODULE}.sync_attachment_snapshot_ids", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock) async def test_fails_fast_when_attachment_snapshot_ids_are_falsy( self, mock_list_attachments, mock_sync_snapshot_ids, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() attachments = [ _mock_attachment(provider_snapshot_id=None, deployment_id=deployment_id), _mock_attachment(provider_snapshot_id="", deployment_id=deployment_id), ] mock_list_attachments.return_value = attachments adapter = AsyncMock() from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced with pytest.raises(ValueError, match="provider_snapshot_id must be non-empty"): await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=AsyncMock(), page=1, size=10, ) adapter.list_snapshots.assert_not_awaited() mock_sync_snapshot_ids.assert_not_awaited() mock_count_attachments.assert_not_awaited() mock_list_with_versions.assert_not_awaited() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock) @patch(f"{MODULE}.sync_attachment_snapshot_ids", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock) async def test_snapshot_sync_savepoint_failure_clears_enrichment_payload( self, mock_list_attachments, mock_sync_snapshot_ids, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() mock_list_attachments.return_value = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=deployment_id) ] rows = [(SimpleNamespace(provider_snapshot_id="snap-1"), SimpleNamespace())] mock_list_with_versions.return_value = rows adapter = AsyncMock() adapter.list_snapshots.return_value = SnapshotListResult( snapshots=[ SnapshotItem( id="snap-1", name="tool-1", provider_data={"connections": {"cfg-1": "conn-1"}}, ) ] ) mock_sync_snapshot_ids.side_effect = RuntimeError("savepoint failed") db = _mock_async_db() from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced out_rows, total, snapshot_result = await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=db, page=1, size=20, ) assert out_rows == rows assert total == 1 assert snapshot_result is None adapter.list_snapshots.assert_awaited_once() mock_sync_snapshot_ids.assert_awaited_once() mock_count_attachments.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock) @patch(f"{MODULE}.sync_attachment_snapshot_ids", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock) async def test_snapshot_sync_errors_fall_back_to_db_rows_without_enrichment( self, mock_list_attachments, mock_sync_snapshot_ids, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() mock_list_attachments.return_value = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=deployment_id) ] rows = [(SimpleNamespace(provider_snapshot_id="snap-1"), SimpleNamespace())] mock_list_with_versions.return_value = rows adapter = AsyncMock() adapter.list_snapshots.side_effect = RuntimeError("provider down") from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced out_rows, total, snapshot_result = await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=AsyncMock(), page=1, size=20, ) assert out_rows == rows assert total == 1 assert snapshot_result is None mock_sync_snapshot_ids.assert_not_awaited() mock_count_attachments.assert_awaited_once() @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=0) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock, return_value=[]) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock, return_value=[]) async def test_forwards_flow_ids_filter_to_attachment_queries( self, mock_list_attachments, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() flow_id = uuid4() adapter = AsyncMock() from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=AsyncMock(), page=1, size=10, flow_ids=[flow_id], ) assert mock_list_attachments.call_args.kwargs["flow_ids"] == [flow_id] assert mock_list_with_versions.call_args.kwargs["flow_ids"] == [flow_id] assert mock_count_attachments.call_args.kwargs["flow_ids"] == [flow_id] @pytest.mark.asyncio @patch(f"{MODULE}.count_deployment_attachments", new_callable=AsyncMock, return_value=1) @patch(f"{MODULE}.list_deployment_attachments_with_versions", new_callable=AsyncMock) @patch(f"{MODULE}.sync_attachment_snapshot_ids", new_callable=AsyncMock) @patch(f"{MODULE}.list_deployment_attachments", new_callable=AsyncMock) async def test_provider_snapshot_result_with_falsy_id_falls_back_without_enrichment( self, mock_list_attachments, mock_sync_snapshot_ids, mock_list_with_versions, mock_count_attachments, ): deployment_id = uuid4() mock_list_attachments.return_value = [ _mock_attachment(provider_snapshot_id="snap-1", deployment_id=deployment_id) ] rows = [(SimpleNamespace(provider_snapshot_id="snap-1"), SimpleNamespace())] mock_list_with_versions.return_value = rows adapter = AsyncMock() adapter.list_snapshots.return_value = _mock_snapshot_view([_mock_snapshot_item(item_id="")]) from langflow.api.v1.mappers.deployments.helpers import list_deployment_flow_versions_synced out_rows, total, snapshot_result = await list_deployment_flow_versions_synced( deployment_adapter=adapter, user_id=uuid4(), provider_id=uuid4(), deployment_id=deployment_id, db=AsyncMock(), page=1, size=20, ) assert out_rows == rows assert total == 1 assert snapshot_result is None adapter.list_snapshots.assert_awaited_once() mock_sync_snapshot_ids.assert_not_awaited() mock_list_with_versions.assert_awaited_once() mock_count_attachments.assert_awaited_once() # --------------------------------------------------------------------------- # flow_version_deployment_attachment CRUD helpers # --------------------------------------------------------------------------- class _FakeAllResult: def __init__(self, rows): self._rows = rows def all(self): return self._rows class _FakeOneResult: def __init__(self, value): self._value = value def one(self): return self._value class _CaptureDb: def __init__(self, result): self._result = result self.statements = [] async def exec(self, statement): self.statements.append(statement) return self._result class TestFlowVersionDeploymentAttachmentCrud: @pytest.mark.asyncio async def test_list_deployment_attachments_with_versions_orders_by_attachment_timestamps(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( list_deployment_attachments_with_versions, ) rows = [(SimpleNamespace(), SimpleNamespace())] db = _CaptureDb(_FakeAllResult(rows)) result = await list_deployment_attachments_with_versions( db, user_id=uuid4(), deployment_id=uuid4(), offset=5, limit=10, ) assert result == rows statement_text = str(db.statements[0]).lower() assert "order by" in statement_text assert "created_at desc" in statement_text assert "updated_at desc" in statement_text @pytest.mark.asyncio async def test_list_deployment_attachments_with_versions_skips_query_when_limit_non_positive(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( list_deployment_attachments_with_versions, ) db = _CaptureDb(_FakeAllResult([])) result = await list_deployment_attachments_with_versions( db, user_id=uuid4(), deployment_id=uuid4(), offset=0, limit=0, ) assert result == [] assert db.statements == [] @pytest.mark.asyncio async def test_count_deployment_attachments_returns_scalar_count(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( count_deployment_attachments, ) db = _CaptureDb(_FakeOneResult(4)) total = await count_deployment_attachments( db, user_id=uuid4(), deployment_id=uuid4(), ) assert total == 4 assert len(db.statements) == 1 # String-match on compiled SQL because these tests use captured # statement objects without a real database engine. statement_text = str(db.statements[0]).lower() assert "join flow_version" in statement_text @pytest.mark.asyncio async def test_count_attachments_by_deployment_ids_joins_flow_version(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( count_attachments_by_deployment_ids, ) deployment_id = uuid4() db = _CaptureDb(_FakeAllResult([])) counts = await count_attachments_by_deployment_ids( db, user_id=uuid4(), deployment_ids=[deployment_id], ) assert counts == {deployment_id: 0} assert len(db.statements) == 1 statement_text = str(db.statements[0]).lower() assert "join flow_version" in statement_text @pytest.mark.asyncio async def test_delete_orphan_attachments_for_flow_ids_joins_flow_version_and_deployment(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( delete_orphan_attachments_for_flow_ids, ) db = _CaptureDb(SimpleNamespace(rowcount=2)) deleted = await delete_orphan_attachments_for_flow_ids( db, user_id=uuid4(), flow_ids=[uuid4()], ) assert deleted == 2 statement_text = str(db.statements[0]).lower() assert "delete from flow_version_deployment_attachment" in statement_text assert "join flow_version" in statement_text assert "join deployment" in statement_text @pytest.mark.asyncio async def test_delete_orphan_attachments_for_project_joins_flow_scope_and_deployment(self): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( delete_orphan_attachments_for_project, ) db = _CaptureDb(SimpleNamespace(rowcount=1)) deleted = await delete_orphan_attachments_for_project( db, user_id=uuid4(), project_id=uuid4(), ) assert deleted == 1 statement_text = str(db.statements[0]).lower() assert "delete from flow_version_deployment_attachment" in statement_text assert "join flow_version" in statement_text assert "join flow" in statement_text assert "join deployment" in statement_text