"""Tests for the Flow Versions API endpoints. Tests cover: - Creating snapshots - Listing version entries - Getting a single version entry with full data - Activating a version (auto-snapshot, overwrite) - Deleting version entries - Edge cases: empty versions, nonexistent IDs, cross-user isolation - Cascade deletion when the parent flow is deleted - Realistic data payloads using starter project JSON """ import json import pytest from fastapi import status from httpx import AsyncClient # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- async def _create_flow(client: AsyncClient, headers: dict, name: str = "version-test-flow") -> dict: """Create a minimal flow and return the JSON response.""" payload = { "name": name, "description": "flow for version tests", "data": {"nodes": [], "edges": []}, "is_component": False, } resp = await client.post("api/v1/flows/", json=payload, headers=headers) assert resp.status_code == status.HTTP_201_CREATED return resp.json() async def _create_snapshot(client: AsyncClient, headers: dict, flow_id: str, description: str | None = None) -> dict: """POST a snapshot and return the JSON response.""" body = {"description": description} if description else {} resp = await client.post(f"api/v1/flows/{flow_id}/versions/", json=body, headers=headers) assert resp.status_code == status.HTTP_201_CREATED return resp.json() async def _list_versions(client: AsyncClient, headers: dict, flow_id: str, *, params: dict | None = None) -> list[dict]: resp = await client.get(f"api/v1/flows/{flow_id}/versions/", headers=headers, params=params) assert resp.status_code == status.HTTP_200_OK body = resp.json() assert "entries" in body assert "max_entries" in body return body["entries"] async def _patch_flow_data(client: AsyncClient, headers: dict, flow_id: str, data: dict) -> dict: """PATCH the flow to change its data (simulates canvas auto-save).""" resp = await client.patch(f"api/v1/flows/{flow_id}", json={"data": data}, headers=headers) assert resp.status_code == status.HTTP_200_OK return resp.json() def _set_deployments_feature_flag(monkeypatch: pytest.MonkeyPatch, *, enabled: bool) -> None: from lfx.services.settings.feature_flags import FEATURE_FLAGS monkeypatch.setattr(FEATURE_FLAGS, "wxo_deployments", enabled) # --------------------------------------------------------------------------- # Basic CRUD # --------------------------------------------------------------------------- async def test_create_snapshot(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"], description="initial save") assert snap["flow_id"] == flow["id"] assert snap["version_number"] == 1 assert snap["version_tag"] == "v1" assert snap["description"] == "initial save" assert "created_at" in snap # List endpoint should NOT include data assert "data" not in snap async def test_create_snapshot_without_description(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) assert snap["description"] is None assert snap["version_number"] == 1 async def test_list_versions_empty(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) entries = await _list_versions(client, logged_in_headers, flow["id"]) assert entries == [] async def test_list_versions_response_includes_max_entries(client: AsyncClient, logged_in_headers, monkeypatch): """The list endpoint should return max_entries from settings.""" from langflow.services.deps import get_settings_service settings = get_settings_service().settings monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 25) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK body = resp.json() assert body["max_entries"] == 25 assert isinstance(body["entries"], list) assert len(body["entries"]) == 1 async def test_list_versions_returns_entries_newest_first(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"], description="first") await _create_snapshot(client, logged_in_headers, flow["id"], description="second") await _create_snapshot(client, logged_in_headers, flow["id"], description="third") entries = await _list_versions(client, logged_in_headers, flow["id"]) assert len(entries) == 3 # newest first → highest version_number first assert entries[0]["version_number"] == 3 assert entries[1]["version_number"] == 2 assert entries[2]["version_number"] == 1 async def test_version_numbers_auto_increment(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) s1 = await _create_snapshot(client, logged_in_headers, flow["id"]) s2 = await _create_snapshot(client, logged_in_headers, flow["id"]) s3 = await _create_snapshot(client, logged_in_headers, flow["id"]) assert s1["version_number"] == 1 assert s2["version_number"] == 2 assert s3["version_number"] == 3 async def test_get_single_version_entry_includes_data(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK full = resp.json() assert "data" in full assert full["data"] == flow["data"] # snapshot captured the flow's data assert full["version_tag"] == "v1" async def test_snapshot_captures_current_flow_data(client: AsyncClient, logged_in_headers): """Snapshot data should be a deep copy of flow.data at the time of snapshot.""" flow = await _create_flow(client, logged_in_headers) # Take snapshot of the initial empty data s1 = await _create_snapshot(client, logged_in_headers, flow["id"], description="empty") # Modify the flow new_data = {"nodes": [{"id": "node-1"}], "edges": []} await _patch_flow_data(client, logged_in_headers, flow["id"], new_data) # Take another snapshot s2 = await _create_snapshot(client, logged_in_headers, flow["id"], description="with node") # Fetch full snapshots and compare r1 = await client.get(f"api/v1/flows/{flow['id']}/versions/{s1['id']}", headers=logged_in_headers) r2 = await client.get(f"api/v1/flows/{flow['id']}/versions/{s2['id']}", headers=logged_in_headers) assert r1.json()["data"] == {"nodes": [], "edges": []} assert r2.json()["data"] == new_data async def test_delete_version_entry(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.delete(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_204_NO_CONTENT entries = await _list_versions(client, logged_in_headers, flow["id"]) assert len(entries) == 0 # --------------------------------------------------------------------------- # Activate version # --------------------------------------------------------------------------- async def test_activate_version_overwrites_flow_data(client: AsyncClient, logged_in_headers): """Activating a saved version should replace the current flow.data.""" flow = await _create_flow(client, logged_in_headers) original_data = flow["data"] # {"nodes": [], "edges": []} # Snapshot the original state snap = await _create_snapshot(client, logged_in_headers, flow["id"], description="v1 original") # Modify the flow modified_data = {"nodes": [{"id": "added-node"}], "edges": []} await _patch_flow_data(client, logged_in_headers, flow["id"], modified_data) # Activate the old snapshot resp = await client.post(f"api/v1/flows/{flow['id']}/versions/{snap['id']}/activate", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK updated_flow = resp.json() assert updated_flow["data"] == original_data async def test_activate_creates_auto_snapshot(client: AsyncClient, logged_in_headers): """Activation should auto-snapshot the current draft before overwriting.""" flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) # Modify flow so auto-snapshot has different data await _patch_flow_data(client, logged_in_headers, flow["id"], {"nodes": [{"id": "x"}], "edges": []}) # Activate — this should create an auto-snapshot first await client.post(f"api/v1/flows/{flow['id']}/versions/{snap['id']}/activate", headers=logged_in_headers) entries = await _list_versions(client, logged_in_headers, flow["id"]) # 1 manual snapshot + 1 auto-snapshot = 2 assert len(entries) == 2 auto_snap = next(e for e in entries if "Auto-saved" in (e.get("description") or "")) assert auto_snap is not None async def test_activate_skips_auto_snapshot_when_save_draft_false(client: AsyncClient, logged_in_headers): """Activation with save_draft=false should NOT create an auto-snapshot.""" flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) # Modify flow so we can verify the draft was NOT saved await _patch_flow_data(client, logged_in_headers, flow["id"], {"nodes": [{"id": "x"}], "edges": []}) # Activate with save_draft=false resp = await client.post( f"api/v1/flows/{flow['id']}/versions/{snap['id']}/activate", params={"save_draft": False}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_200_OK entries = await _list_versions(client, logged_in_headers, flow["id"]) # Only the 1 manual snapshot — no auto-snapshot created assert len(entries) == 1 assert not any("Auto-saved" in (e.get("description") or "") for e in entries) # --------------------------------------------------------------------------- # Cascade deletion # --------------------------------------------------------------------------- async def test_deleting_flow_cascades_to_versions(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) # Delete the flow resp = await client.delete(f"api/v1/flows/{flow['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK # Versions endpoint for the deleted flow should 404 resp = await client.get(f"api/v1/flows/{flow['id']}/versions/", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND # --------------------------------------------------------------------------- # Error / edge cases # --------------------------------------------------------------------------- async def test_get_versions_for_nonexistent_flow(client: AsyncClient, logged_in_headers): fake_id = "00000000-0000-0000-0000-000000000000" resp = await client.get(f"api/v1/flows/{fake_id}/versions/", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_get_nonexistent_version_entry(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) fake_id = "00000000-0000-0000-0000-000000000000" resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{fake_id}", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_activate_nonexistent_version_entry(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) fake_id = "00000000-0000-0000-0000-000000000000" resp = await client.post(f"api/v1/flows/{flow['id']}/versions/{fake_id}/activate", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_delete_nonexistent_version_entry(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) fake_id = "00000000-0000-0000-0000-000000000000" resp = await client.delete(f"api/v1/flows/{flow['id']}/versions/{fake_id}", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_activate_entry_belonging_to_different_flow(client: AsyncClient, logged_in_headers): """A version entry from flow A should not be activatable on flow B.""" flow_a = await _create_flow(client, logged_in_headers, name="flow-a") flow_b = await _create_flow(client, logged_in_headers, name="flow-b") snap_a = await _create_snapshot(client, logged_in_headers, flow_a["id"]) # Try to activate snap_a on flow_b resp = await client.post(f"api/v1/flows/{flow_b['id']}/versions/{snap_a['id']}/activate", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_delete_entry_belonging_to_different_flow(client: AsyncClient, logged_in_headers): """A version entry from flow A should not be deletable via flow B's endpoint.""" flow_a = await _create_flow(client, logged_in_headers, name="flow-a-del") flow_b = await _create_flow(client, logged_in_headers, name="flow-b-del") snap_a = await _create_snapshot(client, logged_in_headers, flow_a["id"]) resp = await client.delete(f"api/v1/flows/{flow_b['id']}/versions/{snap_a['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND async def test_unauthenticated_request_is_rejected(client: AsyncClient): fake_id = "00000000-0000-0000-0000-000000000000" resp = await client.get(f"api/v1/flows/{fake_id}/versions/") assert resp.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN) # --------------------------------------------------------------------------- # Pagination # --------------------------------------------------------------------------- async def test_list_versions_pagination(client: AsyncClient, logged_in_headers): flow = await _create_flow(client, logged_in_headers) # Create 5 snapshots for i in range(5): await _create_snapshot(client, logged_in_headers, flow["id"], description=f"snap-{i}") # Fetch with limit=2 resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"limit": 2, "offset": 0}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_200_OK page1 = resp.json()["entries"] assert len(page1) == 2 assert page1[0]["version_number"] == 5 # newest first assert page1[1]["version_number"] == 4 # Second page resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"limit": 2, "offset": 2}, headers=logged_in_headers, ) page2 = resp.json()["entries"] assert len(page2) == 2 assert page2[0]["version_number"] == 3 # --------------------------------------------------------------------------- # Multi-activate scenario (full lifecycle) # --------------------------------------------------------------------------- async def test_full_lifecycle(client: AsyncClient, logged_in_headers): """End-to-end: create, snapshot, edit, snapshot, activate old, verify.""" # 1. Create flow with initial data flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] initial_data = flow["data"] # 2. Save v1 snapshot v1 = await _create_snapshot(client, logged_in_headers, flow_id, description="version 1") assert v1["version_number"] == 1 # 3. Edit the flow data_v2 = {"nodes": [{"id": "n1"}, {"id": "n2"}], "edges": [{"id": "e1"}]} await _patch_flow_data(client, logged_in_headers, flow_id, data_v2) # 4. Save v2 snapshot v2 = await _create_snapshot(client, logged_in_headers, flow_id, description="version 2") assert v2["version_number"] == 2 # 5. Edit the flow again data_v3 = {"nodes": [{"id": "n1"}, {"id": "n2"}, {"id": "n3"}], "edges": []} await _patch_flow_data(client, logged_in_headers, flow_id, data_v3) # 6. Activate v1 — should auto-snapshot current state, then revert to v1's data resp = await client.post(f"api/v1/flows/{flow_id}/versions/{v1['id']}/activate", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK activated_flow = resp.json() assert activated_flow["data"] == initial_data # 7. Verify versions now has 3 entries: v1, v2, + auto-snapshot (v3) entries = await _list_versions(client, logged_in_headers, flow_id) assert len(entries) == 3 # 8. The auto-snapshot should contain data_v3 auto = next(e for e in entries if "Auto-saved" in (e.get("description") or "")) auto_full = await client.get(f"api/v1/flows/{flow_id}/versions/{auto['id']}", headers=logged_in_headers) assert auto_full.json()["data"] == data_v3 # 9. Activate v2 resp2 = await client.post(f"api/v1/flows/{flow_id}/versions/{v2['id']}/activate", headers=logged_in_headers) assert resp2.status_code == status.HTTP_200_OK assert resp2.json()["data"] == data_v2 # --------------------------------------------------------------------------- # Realistic data payloads (starter project) # --------------------------------------------------------------------------- async def test_snapshot_and_activate_with_complex_flow_data(client: AsyncClient, logged_in_headers): """Snapshot / activate round-trip with a real multi-node flow from test data.""" complex_json = json.loads(pytest.COMPLEX_EXAMPLE_PATH.read_text(encoding="utf-8")) complex_data = complex_json.get("data", complex_json) # Create a flow with the complex data payload = { "name": "complex-version-test", "description": "flow with real nodes", "data": complex_data, "is_component": False, } resp = await client.post("api/v1/flows/", json=payload, headers=logged_in_headers) assert resp.status_code == status.HTTP_201_CREATED flow = resp.json() flow_id = flow["id"] # Verify the flow has meaningful data assert len(flow["data"]["nodes"]) > 1 assert len(flow["data"]["edges"]) > 0 # Snapshot the complex state snap = await _create_snapshot(client, logged_in_headers, flow_id, description="complex v1") # Overwrite the flow with minimal data minimal_data = {"nodes": [{"id": "single-node"}], "edges": []} await _patch_flow_data(client, logged_in_headers, flow_id, minimal_data) # Verify flow was actually changed get_resp = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers) assert len(get_resp.json()["data"]["nodes"]) == 1 # Activate the complex snapshot — should restore the full graph activate_resp = await client.post( f"api/v1/flows/{flow_id}/versions/{snap['id']}/activate", headers=logged_in_headers ) assert activate_resp.status_code == status.HTTP_200_OK restored = activate_resp.json() assert restored["data"] == complex_data async def test_snapshot_preserves_full_node_metadata(client: AsyncClient, logged_in_headers): """Verify that node IDs, positions, and edge handles survive snapshot. Note: the GET endpoint strips API keys server-side, so template password field values will be None. This test focuses on structural preservation (IDs, positions, edge count) rather than exact template value equality. """ complex_json = json.loads(pytest.COMPLEX_EXAMPLE_PATH.read_text(encoding="utf-8")) complex_data = complex_json.get("data", complex_json) payload = { "name": "metadata-preservation-test", "description": "test", "data": complex_data, "is_component": False, } resp = await client.post("api/v1/flows/", json=payload, headers=logged_in_headers) flow = resp.json() snap = await _create_snapshot(client, logged_in_headers, flow["id"]) # Fetch the full snapshot and compare node-by-node full = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) snapshot_data = full.json()["data"] assert len(snapshot_data["nodes"]) == len(complex_data["nodes"]) assert len(snapshot_data["edges"]) == len(complex_data["edges"]) for i, original_node in enumerate(complex_data["nodes"]): snap_node = snapshot_data["nodes"][i] assert snap_node["id"] == original_node["id"] assert snap_node.get("position") == original_node.get("position") # --------------------------------------------------------------------------- # Version limit enforcement # --------------------------------------------------------------------------- async def test_version_limit_enforcement(client: AsyncClient, logged_in_headers, monkeypatch): """Creating snapshots beyond the configured limit prunes the oldest entries.""" # Set a small limit for testing from langflow.services.deps import get_settings_service settings = get_settings_service().settings monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 3) flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] # Create 5 snapshots — limit is 3, so oldest 2 should be pruned for i in range(5): await _create_snapshot(client, logged_in_headers, flow_id, description=f"snap-{i}") entries = await _list_versions(client, logged_in_headers, flow_id) assert len(entries) == 3 async def test_version_limit_keeps_newest(client: AsyncClient, logged_in_headers, monkeypatch): """After pruning, the remaining entries should be the most recent by version_number.""" from langflow.services.deps import get_settings_service settings = get_settings_service().settings monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 3) flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] for i in range(5): await _create_snapshot(client, logged_in_headers, flow_id, description=f"snap-{i}") entries = await _list_versions(client, logged_in_headers, flow_id) version_numbers = [e["version_number"] for e in entries] # Should have versions 5, 4, 3 (newest first) — versions 1 and 2 were pruned assert version_numbers == [5, 4, 3] async def test_pruning_deletes_oldest_by_data_content(client: AsyncClient, logged_in_headers, monkeypatch): """Verify that pruned entries are truly the oldest by checking surviving data content.""" from langflow.services.deps import get_settings_service settings = get_settings_service().settings monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 2) flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] # Create 3 snapshots, each with distinct flow data so we can verify which survived. for i in range(3): data = {"nodes": [{"id": f"node-from-snap-{i}"}], "edges": []} await _patch_flow_data(client, logged_in_headers, flow_id, data) await _create_snapshot(client, logged_in_headers, flow_id, description=f"snap-{i}") # Only the 2 newest should survive (snap-1 and snap-2); snap-0 should be pruned. entries = await _list_versions(client, logged_in_headers, flow_id) assert len(entries) == 2 assert entries[0]["description"] == "snap-2" assert entries[1]["description"] == "snap-1" # Fetch full data for each survivor and confirm it matches the expected snapshot data. for entry, expected_idx in zip(entries, [2, 1], strict=False): resp = await client.get( f"api/v1/flows/{flow_id}/versions/{entry['id']}", headers=logged_in_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["nodes"][0]["id"] == f"node-from-snap-{expected_idx}" async def test_lowered_limit_prunes_excess_on_next_snapshot(client: AsyncClient, logged_in_headers, monkeypatch): """Lowering the limit after entries exist should prune excess on the next snapshot.""" from langflow.services.deps import get_settings_service settings = get_settings_service().settings # Start with a generous limit and create 5 snapshots. monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 10) flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] for i in range(5): await _create_snapshot(client, logged_in_headers, flow_id, description=f"snap-{i}") entries = await _list_versions(client, logged_in_headers, flow_id) assert len(entries) == 5 # Now lower the limit to 2 — existing entries remain untouched until next snapshot. monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 2) entries_before = await _list_versions(client, logged_in_headers, flow_id) assert len(entries_before) == 5 # Still 5 — no pruning yet # Create one more snapshot — should trigger pruning down to 2. await _create_snapshot(client, logged_in_headers, flow_id, description="after-limit-change") entries_after = await _list_versions(client, logged_in_headers, flow_id) assert len(entries_after) == 2 # The two survivors should be the newest: "after-limit-change" and "snap-4" assert entries_after[0]["description"] == "after-limit-change" assert entries_after[1]["description"] == "snap-4" # --------------------------------------------------------------------------- # Activate version with null data # --------------------------------------------------------------------------- async def test_activate_version_with_null_data(client: AsyncClient, logged_in_headers): """Activating a version whose data is None should return 400.""" from uuid import UUID from langflow.services.database.models.flow_version.model import FlowVersion from langflow.services.deps import session_scope from sqlmodel import select flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] snap = await _create_snapshot(client, logged_in_headers, flow_id) # Manually set the snapshot's data to None in the DB async with session_scope() as session: result = await session.exec(select(FlowVersion).where(FlowVersion.id == UUID(snap["id"]))) entry = result.first() if entry: entry.data = None session.add(entry) resp = await client.post(f"api/v1/flows/{flow_id}/versions/{snap['id']}/activate", headers=logged_in_headers) assert resp.status_code == status.HTTP_400_BAD_REQUEST assert "no data" in resp.json()["detail"].lower() # --------------------------------------------------------------------------- # Cross-user isolation # --------------------------------------------------------------------------- async def test_version_entry_scoped_to_flow(client: AsyncClient, logged_in_headers): """Getting a version entry via the wrong flow's endpoint should 404.""" flow_a = await _create_flow(client, logged_in_headers, name="scope-a") flow_b = await _create_flow(client, logged_in_headers, name="scope-b") snap_a = await _create_snapshot(client, logged_in_headers, flow_a["id"]) # Try to access snap_a through flow_b's endpoint resp = await client.get(f"api/v1/flows/{flow_b['id']}/versions/{snap_a['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_404_NOT_FOUND # --------------------------------------------------------------------------- # Edge case: limit boundary values # --------------------------------------------------------------------------- async def test_list_versions_limit_of_one(client: AsyncClient, logged_in_headers): """Limit=1 should return exactly one entry.""" flow = await _create_flow(client, logged_in_headers) for i in range(3): await _create_snapshot(client, logged_in_headers, flow["id"], description=f"s-{i}") resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"limit": 1}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_200_OK assert len(resp.json()["entries"]) == 1 assert resp.json()["entries"][0]["version_number"] == 3 # newest async def test_list_versions_invalid_limit_zero(client: AsyncClient, logged_in_headers): """Limit=0 should be rejected by validation (ge=1).""" flow = await _create_flow(client, logged_in_headers) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"limit": 0}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY async def test_list_versions_limit_exceeds_max(client: AsyncClient, logged_in_headers): """Limit > 100 should be rejected by validation (le=100).""" flow = await _create_flow(client, logged_in_headers) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"limit": 101}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY async def test_get_single_entry_strips_api_keys(client: AsyncClient, logged_in_headers): """GET /versions/{id} should strip API keys from the returned data. The single-entry endpoint is used by the frontend for previewing and per-version export. API keys are stripped server-side to prevent leakage through the preview path, matching the export endpoint's behaviour. """ api_key_data = { "nodes": [ { "id": "node-1", "data": { "node": { "template": { "api_key": { "value": "sk-secret-99999", "name": "api_key", "password": True, } } } }, } ], "edges": [], } payload = { "name": "single-entry-api-key-test", "description": "test", "data": api_key_data, "is_component": False, } resp = await client.post("api/v1/flows/", json=payload, headers=logged_in_headers) assert resp.status_code == status.HTTP_201_CREATED flow = resp.json() snap = await _create_snapshot(client, logged_in_headers, flow["id"]) # Fetch the single entry — API key should be stripped (value set to None) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers, ) assert resp.status_code == status.HTTP_200_OK entry_data = resp.json()["data"] template = entry_data["nodes"][0]["data"]["node"]["template"] assert template["api_key"]["value"] is None async def test_create_snapshot_rejects_long_description(client: AsyncClient, logged_in_headers): """POST with description > 500 chars should be rejected with 422.""" flow = await _create_flow(client, logged_in_headers) long_description = "x" * 501 resp = await client.post( f"api/v1/flows/{flow['id']}/versions/", json={"description": long_description}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY async def test_create_snapshot_accepts_max_description(client: AsyncClient, logged_in_headers): """POST with description exactly 500 chars should succeed.""" flow = await _create_flow(client, logged_in_headers) max_description = "x" * 500 snap = await _create_snapshot(client, logged_in_headers, flow["id"], description=max_description) assert snap["description"] == max_description async def test_activate_with_deeply_nested_data(client: AsyncClient, logged_in_headers): """Activate should work correctly with deeply nested flow data.""" # Create deeply nested data nested = {"level": 0} current = nested for i in range(1, 20): current["child"] = {"level": i} current = current["child"] deep_data = {"nodes": [{"id": "deep-node", "data": nested}], "edges": []} payload = { "name": "deep-nested-test", "description": "test", "data": deep_data, "is_component": False, } resp = await client.post("api/v1/flows/", json=payload, headers=logged_in_headers) assert resp.status_code == status.HTTP_201_CREATED flow = resp.json() flow_id = flow["id"] # Snapshot the deeply nested data snap = await _create_snapshot(client, logged_in_headers, flow_id) # Modify the flow await _patch_flow_data(client, logged_in_headers, flow_id, {"nodes": [], "edges": []}) # Activate the old version — triggers deepcopy of both current and target data resp = await client.post( f"api/v1/flows/{flow_id}/versions/{snap['id']}/activate", headers=logged_in_headers, ) assert resp.status_code == status.HTTP_200_OK restored = resp.json() assert restored["data"] == deep_data async def test_version_number_is_always_positive(client: AsyncClient, logged_in_headers): """get_next_version_number always returns >= 1, even for a brand-new flow.""" from langflow.services.database.models.flow_version.crud import get_next_version_number from langflow.services.deps import session_scope flow = await _create_flow(client, logged_in_headers) async with session_scope() as session: from uuid import UUID next_ver = await get_next_version_number(session, UUID(flow["id"])) assert next_ver >= 1 # After creating a snapshot, next version should still be >= 1 await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: next_ver = await get_next_version_number(session, UUID(flow["id"])) assert next_ver >= 1 assert next_ver == 2 async def test_rapid_snapshots_with_low_limit(client: AsyncClient, logged_in_headers, monkeypatch): """Creating many snapshots rapidly with a low limit should work without errors.""" from langflow.services.deps import get_settings_service settings = get_settings_service().settings monkeypatch.setattr(settings, "max_flow_version_entries_per_flow", 2) flow = await _create_flow(client, logged_in_headers) flow_id = flow["id"] # Create 5 snapshots in quick succession for i in range(5): resp = await client.post( f"api/v1/flows/{flow_id}/versions/", json={"description": f"rapid-{i}"}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_201_CREATED # Should have exactly 2 entries (the limit) entries = await _list_versions(client, logged_in_headers, flow_id) assert len(entries) == 2 # The newest ones should survive assert entries[0]["version_number"] == 5 assert entries[1]["version_number"] == 4 # --------------------------------------------------------------------------- # Deployment-aware: provider-scoped is_deployed indicator # --------------------------------------------------------------------------- async def _attach_version_to_deployment(session, *, user_id, flow_version_id, deployment_id, provider_snapshot_id=None): """Create a FlowVersionDeploymentAttachment row directly in the DB.""" from langflow.services.database.models.flow_version_deployment_attachment.crud import ( create_deployment_attachment, ) return await create_deployment_attachment( session, user_id=user_id, flow_version_id=flow_version_id, deployment_id=deployment_id, provider_snapshot_id=provider_snapshot_id or f"snap-{flow_version_id}", ) async def _create_deployment_row(session, *, user_id, project_id, provider_account_id, name, resource_key): """Create a Deployment row directly in the DB.""" from langflow.services.database.models.deployment.crud import create_deployment from lfx.services.adapters.deployment.schema import DeploymentType return await create_deployment( session, user_id=user_id, project_id=project_id, deployment_provider_account_id=provider_account_id, resource_key=resource_key, name=name, deployment_type=DeploymentType.AGENT, ) async def _create_provider_account_row(session, *, user_id): """Create a DeploymentProviderAccount row directly in the DB.""" from uuid import uuid4 from langflow.services.database.models.deployment_provider_account.crud import create_provider_account suffix = uuid4() return await create_provider_account( session, user_id=user_id, name=f"test-provider-account-{suffix}", provider_tenant_id=None, provider_key="watsonx-orchestrate", provider_url=f"https://test-{suffix}.example.com", api_key="test-key", # pragma: allowlist secret ) async def test_list_versions_omits_is_deployed_without_provider_scope( client: AsyncClient, logged_in_headers, monkeypatch ): """Without provider scoping, deployment status is omitted.""" _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) entries = await _list_versions(client, logged_in_headers, flow["id"]) assert len(entries) == 1 assert "is_deployed" not in entries[0] async def test_list_versions_is_deployed_scoped_to_provider(client: AsyncClient, logged_in_headers, monkeypatch): """Provider-scoped status only reflects deployments under the selected provider.""" from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) snap2 = await _create_snapshot(client, logged_in_headers, flow["id"], description="undeployed") async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() user_id = flow_row.user_id project_id = flow_row.folder_id provider_account_a = await _create_provider_account_row(session, user_id=user_id) provider_account_b = await _create_provider_account_row(session, user_id=user_id) provider_account_a_id = str(provider_account_a.id) provider_account_b_id = str(provider_account_b.id) deployment = await _create_deployment_row( session, user_id=user_id, project_id=project_id, provider_account_id=provider_account_a.id, name="test-deployment", resource_key="rk-1", ) await _attach_version_to_deployment( session, user_id=user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, ) entries = await _list_versions( client, logged_in_headers, flow["id"], params={"deployment_provider_id": provider_account_a_id}, ) assert len(entries) == 2 deployed_entry = next(e for e in entries if e["id"] == snap["id"]) undeployed_entry = next(e for e in entries if e["id"] == snap2["id"]) assert deployed_entry["is_deployed"] is True assert undeployed_entry["is_deployed"] is False entries_other_provider = await _list_versions( client, logged_in_headers, flow["id"], params={"deployment_provider_id": provider_account_b_id}, ) assert len(entries_other_provider) == 2 assert all(entry["is_deployed"] is False for entry in entries_other_provider) async def test_list_versions_plain_mode_returns_all(client: AsyncClient, logged_in_headers, monkeypatch): """Without provider scoping, all versions are returned without deployment status.""" from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"], description="deployed") await _create_snapshot(client, logged_in_headers, flow["id"], description="draft") async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() user_id = flow_row.user_id project_id = flow_row.folder_id provider_account = await _create_provider_account_row(session, user_id=user_id) deployment = await _create_deployment_row( session, user_id=user_id, project_id=project_id, provider_account_id=provider_account.id, name="test-no-filter", resource_key="rk-no-filter", ) entries = await _list_versions(client, logged_in_headers, flow["id"]) snap_deployed = next(e for e in entries if e["description"] == "deployed") await _attach_version_to_deployment( session, user_id=user_id, flow_version_id=UUID(snap_deployed["id"]), deployment_id=deployment.id, ) # No filter — should return both versions entries = await _list_versions(client, logged_in_headers, flow["id"]) assert len(entries) == 2 deployed = next(e for e in entries if e["description"] == "deployed") draft = next(e for e in entries if e["description"] == "draft") assert "is_deployed" not in deployed assert "is_deployed" not in draft async def test_create_snapshot_omits_is_deployed(client: AsyncClient, logged_in_headers, monkeypatch): """Create snapshot response should not include deployment status.""" _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) assert "is_deployed" not in snap async def test_list_versions_hides_deployment_state_when_feature_disabled( client: AsyncClient, logged_in_headers, monkeypatch ): from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=False) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() user_id = flow_row.user_id project_id = flow_row.folder_id provider_account = await _create_provider_account_row(session, user_id=user_id) deployment = await _create_deployment_row( session, user_id=user_id, project_id=project_id, provider_account_id=provider_account.id, name="disabled-flag-list", resource_key="rk-disabled-list", ) await _attach_version_to_deployment( session, user_id=user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, ) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert "is_deployed" not in resp.json()["entries"][0] async def test_list_versions_rejects_provider_id_when_feature_disabled( client: AsyncClient, logged_in_headers, monkeypatch ): from uuid import uuid4 _set_deployments_feature_flag(monkeypatch, enabled=False) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"deployment_provider_id": str(uuid4())}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_400_BAD_REQUEST assert resp.json()["detail"] == "Cannot use deployment_provider_id: the wxo_deployments feature flag is disabled" async def test_list_versions_rejects_unknown_provider_id(client: AsyncClient, logged_in_headers, monkeypatch): from uuid import uuid4 _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"deployment_provider_id": str(uuid4())}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_404_NOT_FOUND assert resp.json()["detail"] == "Deployment provider account not found." async def test_list_versions_rejects_foreign_provider_id(client: AsyncClient, logged_in_headers, monkeypatch): from uuid import uuid4 from langflow.services.database.models.user.model import User from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: foreign_user = User( username=f"foreign-provider-owner-{uuid4()}", password=str(uuid4()), is_active=True, ) session.add(foreign_user) await session.flush() foreign_provider = await _create_provider_account_row(session, user_id=foreign_user.id) foreign_provider_id = str(foreign_provider.id) resp = await client.get( f"api/v1/flows/{flow['id']}/versions/", params={"deployment_provider_id": foreign_provider_id}, headers=logged_in_headers, ) assert resp.status_code == status.HTTP_404_NOT_FOUND assert resp.json()["detail"] == "Deployment provider account not found." # --------------------------------------------------------------------------- # Sync-on-read: stale attachment pruning via provider verification # --------------------------------------------------------------------------- SYNC_MODULE = "langflow.api.v1.flow_version.sync_flow_version_attachments" async def test_list_versions_sync_prunes_stale_attachment(client: AsyncClient, logged_in_headers, monkeypatch): """When sync removes a stale attachment, is_deployed should flip to False.""" from unittest.mock import AsyncMock, patch from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) # Set up a deployed version with a provider_snapshot_id. async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() user_id = flow_row.user_id project_id = flow_row.folder_id provider_account = await _create_provider_account_row(session, user_id=user_id) provider_account_id = str(provider_account.id) deployment = await _create_deployment_row( session, user_id=user_id, project_id=project_id, provider_account_id=provider_account.id, name="sync-prune-test", resource_key="rk-sync-prune", ) await _attach_version_to_deployment( session, user_id=user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, provider_snapshot_id="tool-abc-123", ) deployment_id = deployment.id # Confirm is_deployed=True before sync prunes it. with patch(SYNC_MODULE, new_callable=AsyncMock): entries = await _list_versions( client, logged_in_headers, flow["id"], params={"deployment_provider_id": provider_account_id}, ) assert entries[0]["is_deployed"] is True async def _prune_sync(*_args, **_kwargs): from langflow.services.database.models.flow_version_deployment_attachment.crud import ( delete_deployment_attachment, ) async with session_scope() as session: await delete_deployment_attachment( session, user_id=user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment_id, ) with patch(SYNC_MODULE, new_callable=AsyncMock, side_effect=_prune_sync): entries = await _list_versions( client, logged_in_headers, flow["id"], params={"deployment_provider_id": provider_account_id}, ) assert len(entries) == 1 assert entries[0]["is_deployed"] is False async def test_list_versions_sync_failure_does_not_block_response(client: AsyncClient, logged_in_headers, monkeypatch): """If sync raises, the list endpoint should still return results.""" from unittest.mock import AsyncMock, patch from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() provider_account = await _create_provider_account_row(session, user_id=flow_row.user_id) provider_account_id = str(provider_account.id) with patch(SYNC_MODULE, new_callable=AsyncMock, side_effect=RuntimeError("provider down")): entries = await _list_versions( client, logged_in_headers, flow["id"], params={"deployment_provider_id": provider_account_id}, ) assert len(entries) == 1 async def test_get_single_version_reports_unknown_deployment_status_when_unattached( client: AsyncClient, logged_in_headers, monkeypatch ): """Single-version reads return deployment status as unknown.""" _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert resp.json()["is_deployed"] is None async def test_get_single_version_reports_unknown_deployment_status_when_attached( client: AsyncClient, logged_in_headers, monkeypatch ): """Single-version reads return unknown deployment status even when attached.""" from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() provider_account = await _create_provider_account_row(session, user_id=flow_row.user_id) deployment = await _create_deployment_row( session, user_id=flow_row.user_id, project_id=flow_row.folder_id, provider_account_id=provider_account.id, name="single-get-attached", resource_key="rk-single-get-attached", ) await _attach_version_to_deployment( session, user_id=flow_row.user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, ) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert resp.json()["is_deployed"] is None async def test_get_single_version_reports_unknown_deployment_status_when_feature_disabled( client: AsyncClient, logged_in_headers, monkeypatch ): """Feature-flag state does not change single-version deployment status behavior.""" from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=False) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() provider_account = await _create_provider_account_row(session, user_id=flow_row.user_id) deployment = await _create_deployment_row( session, user_id=flow_row.user_id, project_id=flow_row.folder_id, provider_account_id=provider_account.id, name="single-get-disabled", resource_key="rk-single-get-disabled", ) await _attach_version_to_deployment( session, user_id=flow_row.user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, ) resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert resp.json()["is_deployed"] is None async def test_get_single_version_sync_failure_does_not_block_response( client: AsyncClient, logged_in_headers, monkeypatch ): """Single-version reads remain healthy if the list sync helper would fail.""" from unittest.mock import AsyncMock, patch _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) with patch(SYNC_MODULE, new_callable=AsyncMock, side_effect=RuntimeError("provider down")) as mock_sync: resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert resp.json()["id"] == snap["id"] assert resp.json()["is_deployed"] is None mock_sync.assert_not_awaited() async def test_get_single_version_does_not_trigger_sync_prune(client: AsyncClient, logged_in_headers, monkeypatch): """Single-version reads should not trigger sync/prune side effects.""" from unittest.mock import AsyncMock, patch from uuid import UUID from langflow.services.deps import session_scope _set_deployments_feature_flag(monkeypatch, enabled=True) flow = await _create_flow(client, logged_in_headers) snap = await _create_snapshot(client, logged_in_headers, flow["id"]) async with session_scope() as session: from langflow.services.database.models.flow.model import Flow from langflow.services.database.models.flow_version_deployment_attachment.model import ( FlowVersionDeploymentAttachment, ) from sqlmodel import select flow_row = (await session.exec(select(Flow).where(Flow.id == UUID(flow["id"])))).one() user_id = flow_row.user_id provider_account = await _create_provider_account_row(session, user_id=user_id) deployment = await _create_deployment_row( session, user_id=user_id, project_id=flow_row.folder_id, provider_account_id=provider_account.id, name="single-get-no-sync", resource_key="rk-single-get-no-sync", ) await _attach_version_to_deployment( session, user_id=user_id, flow_version_id=UUID(snap["id"]), deployment_id=deployment.id, provider_snapshot_id="tool-xyz-789", ) with patch(SYNC_MODULE, new_callable=AsyncMock) as mock_sync: resp = await client.get(f"api/v1/flows/{flow['id']}/versions/{snap['id']}", headers=logged_in_headers) assert resp.status_code == status.HTTP_200_OK assert resp.json()["is_deployed"] is None mock_sync.assert_not_awaited() async with session_scope() as session: from langflow.services.database.models.flow_version_deployment_attachment.model import ( FlowVersionDeploymentAttachment, ) from sqlmodel import select attachment = ( await session.exec( select(FlowVersionDeploymentAttachment).where( FlowVersionDeploymentAttachment.flow_version_id == UUID(snap["id"]), ) ) ).first() assert attachment is not None async def test_get_flow_version_entries_by_ids_rejects_oversized_batch(): from unittest.mock import AsyncMock from uuid import uuid4 from langflow.services.database.models.flow_version.crud import get_flow_version_entries_by_ids session = AsyncMock() version_ids = [uuid4() for _ in range(51)] with pytest.raises(ValueError, match="version_ids supports at most 50 values"): await get_flow_version_entries_by_ids( session=session, version_ids=version_ids, user_id=uuid4(), ) session.exec.assert_not_called()