import io import json import zipfile from unittest.mock import MagicMock, patch from uuid import UUID, uuid4 import pytest from fastapi import status from httpx import AsyncClient from langflow.initial_setup.constants import STARTER_FOLDER_NAME from langflow.services.database.models.deployment.model import Deployment from langflow.services.database.models.deployment_provider_account.model import ( DeploymentProviderAccount, DeploymentProviderKey, ) from langflow.services.database.models.flow.model import Flow, FlowCreate from langflow.services.database.models.flow_version.model import FlowVersion from langflow.services.database.models.flow_version_deployment_attachment.model import ( FlowVersionDeploymentAttachment, ) from langflow.services.deps import session_scope from lfx.services.adapters.deployment.schema import DeploymentType CYRILLIC_NAME = "Новый проект" CYRILLIC_DESC = "Описание проекта с кириллицей" # noqa: RUF001 @pytest.fixture def basic_case(): return { "name": "New Project", "description": "", "flows_list": [], "components_list": [], } async def test_create_project(client: AsyncClient, logged_in_headers, basic_case): response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_201_CREATED assert isinstance(result, dict), "The result must be a dictionary" assert "name" in result, "The dictionary must contain a key called 'name'" assert "description" in result, "The dictionary must contain a key called 'description'" assert "id" in result, "The dictionary must contain a key called 'id'" assert "parent_id" in result, "The dictionary must contain a key called 'parent_id'" async def test_create_project_duplicate_name_escapes_like_wildcards(client: AsyncClient, logged_in_headers): unrelated = { "name": "proj_a (7)", "description": "", "flows_list": [], "components_list": [], } wildcard = { "name": "proj_%", "description": "", "flows_list": [], "components_list": [], } response = await client.post("api/v1/projects/", json=unrelated, headers=logged_in_headers) assert response.status_code == status.HTTP_201_CREATED response = await client.post("api/v1/projects/", json=wildcard, headers=logged_in_headers) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "proj_%" response = await client.post("api/v1/projects/", json=wildcard, headers=logged_in_headers) assert response.status_code == status.HTTP_201_CREATED assert response.json()["name"] == "proj_% (1)" async def test_read_projects(client: AsyncClient, logged_in_headers): response = await client.get("api/v1/projects/", headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_200_OK assert isinstance(result, list), "The result must be a list" assert len(result) > 0, "The list must not be empty" async def test_read_project(client: AsyncClient, logged_in_headers, basic_case): # Create a project first response_ = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) id_ = response_.json()["id"] # Get the project response = await client.get(f"api/v1/projects/{id_}", headers=logged_in_headers) result = response.json() # The response structure may be different depending on whether pagination is enabled if isinstance(result, dict) and "folder" in result: # Handle paginated project response folder_data = result["folder"] assert response.status_code == status.HTTP_200_OK assert isinstance(folder_data, dict), "The folder data must be a dictionary" assert "name" in folder_data, "The dictionary must contain a key called 'name'" assert "description" in folder_data, "The dictionary must contain a key called 'description'" assert "id" in folder_data, "The dictionary must contain a key called 'id'" elif isinstance(result, dict) and "project" in result: # Handle paginated project response project_data = result["project"] assert response.status_code == status.HTTP_200_OK assert isinstance(project_data, dict), "The project data must be a dictionary" assert "name" in project_data, "The dictionary must contain a key called 'name'" assert "description" in project_data, "The dictionary must contain a key called 'description'" assert "id" in project_data, "The dictionary must contain a key called 'id'" else: # Handle direct project response assert response.status_code == status.HTTP_200_OK assert isinstance(result, dict), "The result must be a dictionary" assert "name" in result, "The dictionary must contain a key called 'name'" assert "description" in result, "The dictionary must contain a key called 'description'" assert "id" in result, "The dictionary must contain a key called 'id'" async def test_update_project(client: AsyncClient, logged_in_headers, basic_case): update_case = basic_case.copy() update_case["name"] = "Updated Project" # Create a project first response_ = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) id_ = response_.json()["id"] # Update the project response = await client.patch(f"api/v1/projects/{id_}", json=update_case, headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_200_OK assert isinstance(result, dict), "The result must be a dictionary" assert "name" in result, "The dictionary must contain a key called 'name'" assert "description" in result, "The dictionary must contain a key called 'description'" assert "id" in result, "The dictionary must contain a key called 'id'" assert "parent_id" in result, "The dictionary must contain a key called 'parent_id'" async def test_create_project_validation_error(client: AsyncClient, logged_in_headers, basic_case): invalid_case = basic_case.copy() invalid_case.pop("name") response = await client.post("api/v1/projects/", json=invalid_case, headers=logged_in_headers) assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY async def test_delete_project_then_404(client: AsyncClient, logged_in_headers, basic_case): create_resp = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) proj_id = create_resp.json()["id"] del_resp = await client.delete(f"api/v1/projects/{proj_id}", headers=logged_in_headers) assert del_resp.status_code == status.HTTP_204_NO_CONTENT get_resp = await client.get(f"api/v1/projects/{proj_id}", headers=logged_in_headers) assert get_resp.status_code == status.HTTP_404_NOT_FOUND async def test_read_project_invalid_id_format(client: AsyncClient, logged_in_headers): bad_id = "not-a-uuid" response = await client.get(f"api/v1/projects/{bad_id}", headers=logged_in_headers) assert response.status_code in (status.HTTP_422_UNPROCESSABLE_ENTITY, status.HTTP_400_BAD_REQUEST) async def test_read_projects_pagination(client: AsyncClient, logged_in_headers): response = await client.get("api/v1/projects/?limit=1&offset=0", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() if isinstance(result, list): assert len(result) <= 1 else: assert "items" in result assert result.get("limit") == 1 async def test_read_projects_empty(client: AsyncClient, logged_in_headers): # Ensure DB is clean by fetching with a random header that forces each test transactional isolation random_headers = {**logged_in_headers, "X-Transaction-ID": str(uuid4())} response = await client.get("api/v1/projects/", headers=random_headers) if response.json(): pytest.skip("Pre-existing projects found; skipping empty list assertion") assert response.status_code == status.HTTP_200_OK assert response.json() == [] async def test_create_and_read_project_cyrillic(client: AsyncClient, logged_in_headers): """Ensure that the API correctly handles non-ASCII (Cyrillic) characters during project creation and retrieval.""" payload = { "name": CYRILLIC_NAME, "description": CYRILLIC_DESC, "flows_list": [], "components_list": [], } # Create the project with Cyrillic characters create_resp = await client.post("api/v1/projects/", json=payload, headers=logged_in_headers) assert create_resp.status_code == status.HTTP_201_CREATED created = create_resp.json() assert created["name"] == CYRILLIC_NAME assert created["description"] == CYRILLIC_DESC proj_id = created["id"] # Fetch the project back to verify round-trip UTF-8 integrity get_resp = await client.get(f"api/v1/projects/{proj_id}", headers=logged_in_headers) assert get_resp.status_code == status.HTTP_200_OK fetched = get_resp.json() # Handle potential pagination/envelope variations already seen in other tests if isinstance(fetched, dict) and "folder" in fetched: fetched = fetched["folder"] elif isinstance(fetched, dict) and "project" in fetched: fetched = fetched["project"] assert fetched["name"] == CYRILLIC_NAME assert fetched["description"] == CYRILLIC_DESC async def test_update_project_preserves_flows(client: AsyncClient, logged_in_headers): """Test that renaming a project preserves all associated flows (regression test for flow loss bug).""" # Create a project project_payload = { "name": "Project with Flows", "description": "Testing flow preservation", "flows_list": [], "components_list": [], } create_resp = await client.post("api/v1/projects/", json=project_payload, headers=logged_in_headers) assert create_resp.status_code == status.HTTP_201_CREATED project = create_resp.json() project_id = project["id"] # Create flows in the project flow1_payload = { "name": "Test Flow 1", "description": "First test flow", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": False, } flow2_payload = { "name": "Test Flow 2", "description": "Second test flow", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": False, } flow1_resp = await client.post("api/v1/flows/", json=flow1_payload, headers=logged_in_headers) flow2_resp = await client.post("api/v1/flows/", json=flow2_payload, headers=logged_in_headers) assert flow1_resp.status_code == status.HTTP_201_CREATED assert flow2_resp.status_code == status.HTTP_201_CREATED flow1_id = flow1_resp.json()["id"] flow2_id = flow2_resp.json()["id"] # Get project to verify flows are associated get_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert get_resp.status_code == status.HTTP_200_OK project_data = get_resp.json() # Current behavior: all flows (including components) are in the flows field flows_before = project_data.get("flows", []) # Filter only actual flows (not components) actual_flows_before = [f for f in flows_before if not f.get("is_component", False)] assert len(actual_flows_before) == 2 flow_ids_before = [f["id"] for f in actual_flows_before] assert str(flow1_id) in flow_ids_before assert str(flow2_id) in flow_ids_before # Update project name (the bug scenario) update_payload = {"name": "Renamed Project with Flows", "description": "Testing flow preservation after rename"} update_resp = await client.patch(f"api/v1/projects/{project_id}", json=update_payload, headers=logged_in_headers) assert update_resp.status_code == status.HTTP_200_OK # Verify project was renamed updated_project = update_resp.json() assert updated_project["name"] == "Renamed Project with Flows" # Critical test: Verify flows are still associated after rename get_after_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert get_after_resp.status_code == status.HTTP_200_OK project_after = get_after_resp.json() flows_after = project_after.get("flows", []) actual_flows_after = [f for f in flows_after if not f.get("is_component", False)] # This was the bug: flows were being lost after project rename assert len(actual_flows_after) == 2, f"Expected 2 flows after rename, got {len(actual_flows_after)}. Flows lost!" flow_ids_after = [f["id"] for f in actual_flows_after] assert str(flow1_id) in flow_ids_after, "Flow 1 was lost after project rename!" assert str(flow2_id) in flow_ids_after, "Flow 2 was lost after project rename!" # Verify individual flows still exist and are accessible flow1_get_resp = await client.get(f"api/v1/flows/{flow1_id}", headers=logged_in_headers) flow2_get_resp = await client.get(f"api/v1/flows/{flow2_id}", headers=logged_in_headers) assert flow1_get_resp.status_code == status.HTTP_200_OK assert flow2_get_resp.status_code == status.HTTP_200_OK # Verify flows still reference the correct project flow1_data = flow1_get_resp.json() flow2_data = flow2_get_resp.json() assert str(flow1_data["folder_id"]) == str(project_id) assert str(flow2_data["folder_id"]) == str(project_id) async def test_update_project_preserves_components(client: AsyncClient, logged_in_headers): """Test that renaming a project preserves all associated components.""" # Create a project project_payload = { "name": "Project with Components", "description": "Testing component preservation", "flows_list": [], "components_list": [], } create_resp = await client.post("api/v1/projects/", json=project_payload, headers=logged_in_headers) assert create_resp.status_code == status.HTTP_201_CREATED project = create_resp.json() project_id = project["id"] # Create components in the project comp1_payload = { "name": "Test Component 1", "description": "First test component", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": True, # This makes it a component } comp2_payload = { "name": "Test Component 2", "description": "Second test component", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": True, # This makes it a component } comp1_resp = await client.post("api/v1/flows/", json=comp1_payload, headers=logged_in_headers) comp2_resp = await client.post("api/v1/flows/", json=comp2_payload, headers=logged_in_headers) assert comp1_resp.status_code == status.HTTP_201_CREATED assert comp2_resp.status_code == status.HTTP_201_CREATED comp1_id = comp1_resp.json()["id"] comp2_id = comp2_resp.json()["id"] # Get project to verify components are associated get_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert get_resp.status_code == status.HTTP_200_OK project_data = get_resp.json() # Current behavior: all flows (including components) are in the flows field flows_before = project_data.get("flows", []) # Filter only components components_before = [f for f in flows_before if f.get("is_component", False)] assert len(components_before) == 2 component_ids_before = [c["id"] for c in components_before] assert str(comp1_id) in component_ids_before assert str(comp2_id) in component_ids_before # Update project name update_payload = {"name": "Renamed Project with Components"} update_resp = await client.patch(f"api/v1/projects/{project_id}", json=update_payload, headers=logged_in_headers) assert update_resp.status_code == status.HTTP_200_OK # Verify components are still associated after rename get_after_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert get_after_resp.status_code == status.HTTP_200_OK project_after = get_after_resp.json() flows_after = project_after.get("flows", []) components_after = [f for f in flows_after if f.get("is_component", False)] assert len(components_after) == 2, ( f"Expected 2 components after rename, got {len(components_after)}. Components lost!" ) component_ids_after = [c["id"] for c in components_after] assert str(comp1_id) in component_ids_after, "Component 1 was lost after project rename!" assert str(comp2_id) in component_ids_after, "Component 2 was lost after project rename!" async def test_update_project_preserves_mixed_flows_and_components(client: AsyncClient, logged_in_headers): """Test that renaming a project preserves both flows and components correctly.""" # Create a project project_payload = { "name": "Mixed Project", "description": "Testing mixed flows and components preservation", "flows_list": [], "components_list": [], } create_resp = await client.post("api/v1/projects/", json=project_payload, headers=logged_in_headers) assert create_resp.status_code == status.HTTP_201_CREATED project = create_resp.json() project_id = project["id"] # Create flows and components flow_payload = { "name": "Regular Flow", "description": "A regular flow", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": False, } component_payload = { "name": "Custom Component", "description": "A custom component", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": True, } flow_resp = await client.post("api/v1/flows/", json=flow_payload, headers=logged_in_headers) comp_resp = await client.post("api/v1/flows/", json=component_payload, headers=logged_in_headers) assert flow_resp.status_code == status.HTTP_201_CREATED assert comp_resp.status_code == status.HTTP_201_CREATED flow_id = flow_resp.json()["id"] comp_id = comp_resp.json()["id"] # Verify initial state get_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) project_data = get_resp.json() flows_before = project_data.get("flows", []) actual_flows_before = [f for f in flows_before if not f.get("is_component", False)] components_before = [f for f in flows_before if f.get("is_component", False)] assert len(actual_flows_before) == 1 assert len(components_before) == 1 # Update project update_payload = {"name": "Renamed Mixed Project"} update_resp = await client.patch(f"api/v1/projects/{project_id}", json=update_payload, headers=logged_in_headers) assert update_resp.status_code == status.HTTP_200_OK # Verify both flows and components preserved get_after_resp = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) project_after = get_after_resp.json() flows_after = project_after.get("flows", []) actual_flows_after = [f for f in flows_after if not f.get("is_component", False)] components_after = [f for f in flows_after if f.get("is_component", False)] assert len(actual_flows_after) == 1, "Flow was lost after project rename!" assert len(components_after) == 1, "Component was lost after project rename!" flow_id_after = actual_flows_after[0]["id"] comp_id_after = components_after[0]["id"] assert str(flow_id) == flow_id_after assert str(comp_id) == comp_id_after # MCP-related tests class TestProjectMCPIntegration: """Test MCP integration features in projects API.""" @pytest.fixture def mock_mcp_settings_enabled(self): """Mock settings with MCP auto-add enabled.""" with patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings: mock_service = MagicMock() mock_service.settings.add_projects_to_mcp_servers = True mock_service.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_service yield mock_service @pytest.fixture def mock_mcp_settings_disabled(self): """Mock settings with MCP auto-add disabled.""" with patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings: mock_service = MagicMock() mock_service.settings.add_projects_to_mcp_servers = False mock_service.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_service yield mock_service async def test_create_project_with_mcp_auto_add_disabled( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_disabled, # noqa: ARG002 ): """Test project creation when MCP auto-add is disabled.""" response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_201_CREATED assert "name" in result assert result["name"] == basic_case["name"] async def test_create_project_with_mcp_auto_add_enabled_success( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test successful project creation with MCP server auto-add.""" with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.create_api_key") as mock_create_api_key, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings # Mock API key creation mock_api_key_response = MagicMock() mock_api_key_response.api_key = "test-api-key-123" # pragma: allowlist secret mock_create_api_key.return_value = mock_api_key_response # Mock validation - no conflict mock_validation_result = MagicMock() mock_validation_result.has_conflict = False mock_validation_result.should_skip = False mock_validation_result.server_name = "lf-new-project" mock_validate.return_value = mock_validation_result mock_update_server.return_value = None response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_201_CREATED assert "name" in result # Verify MCP server creation was attempted mock_validate.assert_called_once() mock_update_server.assert_called_once() async def test_create_project_with_mcp_auto_add_enabled_success_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test for project creation with MCP server auto-add.""" with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.create_api_key") as mock_create_api_key, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings # Mock API key creation mock_api_key_response = MagicMock() mock_api_key_response.api_key = "test-api-key-123" # pragma: allowlist secret mock_create_api_key.return_value = mock_api_key_response # Mock validation - no conflict mock_validation_result = MagicMock() mock_validation_result.has_conflict = False mock_validation_result.should_skip = False mock_validation_result.server_name = "lf-new-project" mock_validate.return_value = mock_validation_result mock_update_server.return_value = None response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) result = response.json() assert response.status_code == status.HTTP_201_CREATED assert "name" in result # Verify MCP server creation was attempted mock_validate.assert_called_once() mock_update_server.assert_called_once() async def test_create_project_with_mcp_server_conflict( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project creation failure due to MCP server name conflict.""" with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings # Mock validation - has conflict mock_validation_result = MagicMock() mock_validation_result.has_conflict = True mock_validation_result.conflict_message = ( "MCP server name conflict: 'lf-new-project' already exists " "for a different project. Cannot create MCP server for project " "'New Project' (ID: test-project-id)" ) mock_validate.return_value = mock_validation_result # The validation function should raise the HTTPException during project creation response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert response.status_code == status.HTTP_409_CONFLICT response_data = response.json() assert "detail" in response_data assert mock_validation_result.conflict_message == response_data["detail"] # Verify validation was called with correct parameters mock_validate.assert_called_once() async def test_create_project_with_mcp_server_conflict_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test verifying project creation failure due to MCP server name conflict.""" with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings # Mock validation - has conflict mock_validation_result = MagicMock() mock_validation_result.has_conflict = True mock_validation_result.conflict_message = ( "MCP server name conflict: 'lf-new-project' already exists " "for a different project. Cannot create MCP server for project " "'New Project' (ID: test-project-id)" ) mock_validate.return_value = mock_validation_result response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert response.status_code == status.HTTP_409_CONFLICT response_data = response.json() assert "detail" in response_data assert mock_validation_result.conflict_message == response_data["detail"] # Verify validation was called with correct parameters mock_validate.assert_called_once() async def test_create_project_oauth_not_implemented( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project creation with OAuth auth type raises NotImplementedError.""" oauth_case = basic_case.copy() oauth_case["auth_settings"] = {"auth_type": "oauth"} with ( patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks to trigger OAuth path mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock validation - no conflict but OAuth case will raise NotImplementedError mock_validation_result = MagicMock() mock_validation_result.has_conflict = False mock_validation_result.should_skip = False mock_validation_result.server_name = "lf-new-project" mock_validate.return_value = mock_validation_result response = await client.post("api/v1/projects/", json=oauth_case, headers=logged_in_headers) # Should still create project but log error about OAuth assert response.status_code == status.HTTP_201_CREATED async def test_create_project_oauth_not_implemented_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test verifying OAuth paths raise NotImplementedError during project creation.""" oauth_case = basic_case.copy() oauth_case["auth_settings"] = {"auth_type": "oauth"} with ( patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks to trigger OAuth path mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock validation - no conflict but OAuth case will raise NotImplementedError mock_validation_result = MagicMock() mock_validation_result.has_conflict = False mock_validation_result.should_skip = False mock_validation_result.server_name = "lf-new-project" mock_validate.return_value = mock_validation_result response = await client.post("api/v1/projects/", json=oauth_case, headers=logged_in_headers) # Should still create project but log error about OAuth assert response.status_code == status.HTTP_201_CREATED async def test_update_project_name_with_mcp_server_update( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project rename with MCP server name update.""" # First create a project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"), patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Now update the project name update_case = {"name": "Updated Project Name", "description": "Updated description"} with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock old server validation mock_old_validation = MagicMock() mock_old_validation.server_exists = True mock_old_validation.project_id_matches = True mock_old_validation.server_name = "lf-new-project" mock_old_validation.existing_config = {"command": "uvx", "args": ["mcp-proxy", "old-url"]} # Mock new server validation mock_new_validation = MagicMock() mock_new_validation.has_conflict = False mock_new_validation.server_name = "lf-updated-project-name" mock_validate.side_effect = [mock_old_validation, mock_new_validation] response = await client.patch(f"api/v1/projects/{project_id}", json=update_case, headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() assert result["name"] == "Updated Project Name" # Should validate both old and new server names assert mock_validate.call_count == 2 # Should update server twice (delete old, create new) assert mock_update_server.call_count == 2 async def test_update_project_name_with_mcp_server_update_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test for project rename with MCP server name update.""" # First create a project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Now update the project name update_case = {"name": "Updated Project Name", "description": "Updated description"} with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock old server validation mock_old_validation = MagicMock() mock_old_validation.server_exists = True mock_old_validation.project_id_matches = True mock_old_validation.server_name = "lf-new-project" mock_old_validation.existing_config = {"command": "uvx", "args": ["mcp-proxy", "old-url"]} # Mock new server validation mock_new_validation = MagicMock() mock_new_validation.has_conflict = False mock_new_validation.server_name = "lf-updated-project-name" mock_validate.side_effect = [mock_old_validation, mock_new_validation] response = await client.patch(f"api/v1/projects/{project_id}", json=update_case, headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() assert result["name"] == "Updated Project Name" # Should validate both old and new server names assert mock_validate.call_count == 2 # Should update server twice (delete old, create new) assert mock_update_server.call_count == 2 async def test_update_project_name_with_mcp_conflict( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project rename with MCP server name conflict.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"), patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Try to update to conflicting name update_case = {"name": "Conflicting Project"} with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock old server validation - exists and matches mock_old_validation = MagicMock() mock_old_validation.server_exists = True mock_old_validation.project_id_matches = True mock_old_validation.server_name = "lf-new-project" # Mock new server validation - has conflict mock_new_validation = MagicMock() mock_new_validation.has_conflict = True mock_new_validation.conflict_message = "Server name conflict with different project" mock_new_validation.server_name = "lf-conflicting-project" mock_validate.side_effect = [mock_old_validation, mock_new_validation] response = await client.patch(f"api/v1/projects/{project_id}", json=update_case, headers=logged_in_headers) assert response.status_code == status.HTTP_409_CONFLICT assert "conflict" in response.json()["detail"].lower() async def test_update_project_name_with_mcp_conflict_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test for project rename with MCP server name conflict.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Try to update to conflicting name update_case = {"name": "Conflicting Project"} with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock old server validation - exists and matches mock_old_validation = MagicMock() mock_old_validation.server_exists = True mock_old_validation.project_id_matches = True mock_old_validation.server_name = "lf-new-project" # Mock new server validation - has conflict mock_new_validation = MagicMock() mock_new_validation.has_conflict = True mock_new_validation.conflict_message = "Server name conflict with different project" mock_new_validation.server_name = "lf-conflicting-project" mock_validate.side_effect = [mock_old_validation, mock_new_validation] response = await client.patch(f"api/v1/projects/{project_id}", json=update_case, headers=logged_in_headers) assert response.status_code == status.HTTP_409_CONFLICT assert "conflict" in response.json()["detail"].lower() async def test_delete_project_with_mcp_server_cleanup( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project deletion with MCP server cleanup.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"), patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Delete the project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock validation - server exists and matches this project mock_validation = MagicMock() mock_validation.server_exists = True mock_validation.project_id_matches = True mock_validation.server_name = "lf-new-project" mock_validate.return_value = mock_validation response = await client.delete(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_204_NO_CONTENT # Should validate server for deletion mock_validate.assert_called_once() # Should call update_server with delete=True mock_update_server.assert_called_once() _, kwargs = mock_update_server.call_args assert kwargs.get("delete") is True async def test_delete_project_with_mcp_server_cleanup_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test for project deletion with MCP server cleanup.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Delete the project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock validation - server exists and matches this project mock_validation = MagicMock() mock_validation.server_exists = True mock_validation.project_id_matches = True mock_validation.server_name = "lf-new-project" mock_validate.return_value = mock_validation response = await client.delete(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_204_NO_CONTENT # Should validate server for deletion mock_validate.assert_called_once() # Should call update_server with delete=True mock_update_server.assert_called_once() _, kwargs = mock_update_server.call_args assert kwargs.get("delete") is True async def test_delete_project_mcp_server_different_project( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test project deletion when MCP server belongs to different project.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url"), patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Delete the project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock validation - server exists but belongs to different project mock_validation = MagicMock() mock_validation.server_exists = True mock_validation.project_id_matches = False mock_validation.server_name = "lf-new-project" mock_validate.return_value = mock_validation response = await client.delete(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_204_NO_CONTENT # Should validate server but not delete it mock_validate.assert_called_once() mock_update_server.assert_not_called() async def test_delete_project_mcp_server_different_project_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test for project deletion when MCP server belongs to different project.""" # Create project first with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate_create, patch("langflow.api.v1.projects_mcp_helpers.update_server"), patch("langflow.api.v1.projects_mcp_helpers.create_api_key"), patch("langflow.api.v1.projects_mcp_helpers.get_storage_service"), ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_validation_create = MagicMock() mock_validation_create.has_conflict = False mock_validation_create.should_skip = False mock_validation_create.server_name = "lf-new-project" mock_validate_create.return_value = mock_validation_create create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) project_id = create_response.json()["id"] # Delete the project with ( patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.update_server") as mock_update_server, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Mock settings to enable MCP auto-add mock_settings = MagicMock() mock_settings.settings.add_projects_to_mcp_servers = True mock_settings.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_settings mock_storage.return_value = MagicMock() # Mock validation - server exists but belongs to different project mock_validation = MagicMock() mock_validation.server_exists = True mock_validation.project_id_matches = False mock_validation.server_name = "lf-new-project" mock_validate.return_value = mock_validation response = await client.delete(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_204_NO_CONTENT # Should validate server but not delete it mock_validate.assert_called_once() mock_update_server.assert_not_called() async def test_create_project_auto_login_disabled_adds_api_key_auth( self, client: AsyncClient, logged_in_headers, basic_case ): """Test that projects get API key auth when AUTO_LOGIN is disabled.""" with patch("langflow.api.v1.projects.get_settings_service") as mock_get_settings: mock_service = MagicMock() mock_service.settings.add_projects_to_mcp_servers = False # Disable MCP to focus on auth mock_service.auth_settings.AUTO_LOGIN = False mock_get_settings.return_value = mock_service with patch("langflow.api.v1.projects.encrypt_auth_settings") as mock_encrypt: mock_encrypt.return_value = {"auth_type": "apikey"} response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert response.status_code == status.HTTP_201_CREATED # Verify encrypt_auth_settings was called with apikey auth mock_encrypt.assert_called_once_with({"auth_type": "apikey"}) async def test_project_mcp_exception_handling( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Test that MCP exceptions during project creation don't prevent project creation.""" with ( patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock validation to raise an exception mock_validate.side_effect = Exception("MCP validation failed") response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) # Project should still be created despite MCP error assert response.status_code == status.HTTP_201_CREATED result = response.json() assert "name" in result assert result["name"] == basic_case["name"] async def test_project_mcp_exception_handling_legacy_sse( self, client: AsyncClient, logged_in_headers, basic_case, mock_mcp_settings_enabled, # noqa: ARG002 ): """Legacy SSE test ensuring MCP exceptions don't block project creation.""" with ( patch("langflow.api.v1.mcp_projects.get_project_sse_url") as mock_sse_url, patch("langflow.api.v1.projects_mcp_helpers.get_project_streamable_http_url") as mock_streamable_url, patch("langflow.api.v1.projects_mcp_helpers.validate_mcp_server_for_project") as mock_validate, patch("langflow.api.v1.projects_mcp_helpers.get_storage_service") as mock_storage, ): # Setup mocks mock_sse_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/sse" mock_streamable_url.return_value = "http://localhost:7860/api/v1/mcp/project/test-id/streamable" mock_storage.return_value = MagicMock() # Mock validation to raise an exception mock_validate.side_effect = Exception("MCP validation failed") response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) # Project should still be created despite MCP error assert response.status_code == status.HTTP_201_CREATED result = response.json() assert "name" in result assert result["name"] == basic_case["name"] # Tests for the read_project bug fix class TestReadProjectBugFix: """Test the read_project endpoint fix for ASGI response bug.""" async def test_read_project_without_pagination_params(self, client: AsyncClient, logged_in_headers, basic_case): """Test read_project returns correct response when no pagination params are provided.""" # Create a project first create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] # Read project without pagination params response = await client.get(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return FolderReadWithFlows (direct project response) assert isinstance(result, dict) assert "name" in result assert "description" in result assert "id" in result assert "flows" in result assert result["name"] == basic_case["name"] async def test_read_project_with_pagination_params(self, client: AsyncClient, logged_in_headers, basic_case): """Test read_project returns paginated response when pagination params are provided.""" # Create a project first create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] # Read project with pagination params response = await client.get(f"api/v1/projects/{project_id}?page=1&size=10", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return FolderWithPaginatedFlows (paginated response) assert isinstance(result, dict) assert "folder" in result assert "flows" in result # Check folder structure folder = result["folder"] assert "name" in folder assert "description" in folder assert "id" in folder assert folder["name"] == basic_case["name"] # Check flows pagination structure flows = result["flows"] assert "items" in flows assert "total" in flows assert "page" in flows assert "size" in flows async def test_read_project_with_partial_pagination_params( self, client: AsyncClient, logged_in_headers, basic_case ): """Test read_project behavior when only some pagination params are provided.""" # Create a project first create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] # Test with only page param (no size) response = await client.get(f"api/v1/projects/{project_id}?page=1", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return non-paginated response (FolderReadWithFlows) assert isinstance(result, dict) assert "name" in result # Direct project response assert "flows" in result assert result["name"] == basic_case["name"] # Test with only size param (no page) response = await client.get(f"api/v1/projects/{project_id}?size=10", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return non-paginated response (FolderReadWithFlows) assert isinstance(result, dict) assert "name" in result # Direct project response assert "flows" in result assert result["name"] == basic_case["name"] async def test_read_project_with_filtering_params(self, client: AsyncClient, logged_in_headers, basic_case): """Test read_project with filtering parameters (is_component, is_flow, search).""" # Create a project first create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] # Create a flow and component in the project for filtering tests flow_payload = { "name": "Test Flow", "description": "A test flow", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": False, } component_payload = { "name": "Test Component", "description": "A test component", "folder_id": project_id, "data": {"nodes": [], "edges": []}, "is_component": True, } flow_response = await client.post("api/v1/flows/", json=flow_payload, headers=logged_in_headers) comp_response = await client.post("api/v1/flows/", json=component_payload, headers=logged_in_headers) assert flow_response.status_code == status.HTTP_201_CREATED assert comp_response.status_code == status.HTTP_201_CREATED # Test with filtering params but no pagination (should use non-paginated path) response = await client.get(f"api/v1/projects/{project_id}?is_flow=true", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return non-paginated response assert isinstance(result, dict) assert "name" in result assert "flows" in result # Test with filtering params AND pagination (should use paginated path) response = await client.get( f"api/v1/projects/{project_id}?is_flow=true&page=1&size=10", headers=logged_in_headers ) assert response.status_code == status.HTTP_200_OK result = response.json() # Should return paginated response assert isinstance(result, dict) assert "folder" in result assert "flows" in result assert "items" in result["flows"] async def test_read_project_consistent_response_structure(self, client: AsyncClient, logged_in_headers, basic_case): """Test that read_project returns consistent response structure in all cases.""" # Create a project first create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] # Test multiple request scenarios to ensure consistency test_cases = [ # No params - should return FolderReadWithFlows {"params": "", "expect_paginated": False}, # Only search - should return FolderReadWithFlows {"params": "?search=test", "expect_paginated": False}, # Only is_component - should return FolderReadWithFlows {"params": "?is_component=true", "expect_paginated": False}, # Only is_flow - should return FolderReadWithFlows {"params": "?is_flow=true", "expect_paginated": False}, # Only page - should return FolderReadWithFlows {"params": "?page=1", "expect_paginated": False}, # Only size - should return FolderReadWithFlows {"params": "?size=10", "expect_paginated": False}, # Both page and size - should return FolderWithPaginatedFlows {"params": "?page=1&size=10", "expect_paginated": True}, # Page, size and filters - should return FolderWithPaginatedFlows {"params": "?page=1&size=10&is_flow=true", "expect_paginated": True}, ] for test_case in test_cases: response = await client.get(f"api/v1/projects/{project_id}{test_case['params']}", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK, f"Failed for params: {test_case['params']}" result = response.json() assert isinstance(result, dict), f"Result should be dict for params: {test_case['params']}" if test_case["expect_paginated"]: # Paginated response structure assert "folder" in result, f"Paginated response missing 'folder' for params: {test_case['params']}" assert "flows" in result, f"Paginated response missing 'flows' for params: {test_case['params']}" assert "items" in result["flows"], f"Paginated flows missing 'items' for params: {test_case['params']}" assert "total" in result["flows"], f"Paginated flows missing 'total' for params: {test_case['params']}" else: # Non-paginated response structure assert "name" in result, f"Non-paginated response missing 'name' for params: {test_case['params']}" assert "flows" in result, f"Non-paginated response missing 'flows' for params: {test_case['params']}" # Should NOT have pagination structure assert "folder" not in result, ( f"Non-paginated response should not have 'folder' for params: {test_case['params']}" ) async def test_read_project_error_handling_consistency(self, client: AsyncClient, logged_in_headers): """Test that error handling is consistent across both response paths.""" import uuid non_existent_id = str(uuid.uuid4()) # Test both pagination and non-pagination paths with non-existent project test_cases = [ "", # Non-paginated path "?page=1&size=10", # Paginated path ] for params in test_cases: response = await client.get(f"api/v1/projects/{non_existent_id}{params}", headers=logged_in_headers) assert response.status_code == status.HTTP_404_NOT_FOUND, f"Should return 404 for params: {params}" result = response.json() assert "detail" in result, f"Error response should have 'detail' for params: {params}" assert "not found" in result["detail"].lower(), ( f"Error message should mention 'not found' for params: {params}" ) async def test_download_file_starter_project(client: AsyncClient, logged_in_headers, active_user, json_flow): """Test downloading a project with multiple flows. This test specifically validates: 1. The download endpoint returns a valid ZIP file with multiple flows 2. The remove_api_keys function handles flows with various template structures, including components that don't have 'name' keys in their template values (e.g., Note components with only backgroundColor) 3. API keys are removed from downloaded flows 4. Non-sensitive data is preserved in the download """ # Create a project for the user (since download_file requires user ownership) project_payload = { "name": STARTER_FOLDER_NAME, "description": "Starter projects to help you get started in Langflow.", "flows_list": [], "components_list": [], } create_response = await client.post("api/v1/projects/", json=project_payload, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED starter_project = create_response.json() starter_project_id = starter_project["id"] # Create multiple flows in the project flow_data = json.loads(json_flow) # Create a flow with a Note component to test the bug fix # Note components have template values without 'name' keys flow_with_note = { "nodes": [ { "id": "note-1", "type": "genericNode", "data": { "node": { "template": { "backgroundColor": {"value": "#ffffff"}, # No 'name' key "text": {"value": "Test note"}, # No 'name' key } } }, }, # Add a node with API keys to test removal { "id": "api-node-1", "type": "genericNode", "data": { "node": { "template": { "api_key": { "name": "api_key", "value": "secret-key-123", "password": True, }, "regular_field": {"name": "regular_field", "value": "keep-this"}, } } }, }, ], "edges": [], } flows_created = [] async with session_scope() as session: # Create 3 flows: 2 from basic example + 1 with Note component for i in range(2): flow_create = FlowCreate( name=f"Starter Flow {i + 1}", description=f"Test starter flow {i + 1}", data=flow_data.get("data", {}), folder_id=starter_project_id, user_id=active_user.id, ) flow = Flow.model_validate(flow_create.model_dump(exclude={"id"})) session.add(flow) flows_created.append(flow) # Add flow with Note component flow_create_note = FlowCreate( name="Flow with Note", description="Flow with Note component and API keys", data=flow_with_note, folder_id=starter_project_id, user_id=active_user.id, ) flow_note = Flow.model_validate(flow_create_note.model_dump(exclude={"id"})) session.add(flow_note) flows_created.append(flow_note) await session.flush() # Refresh to get IDs for flow in flows_created: await session.refresh(flow) await session.commit() # Download the starter project response = await client.get( f"api/v1/projects/download/{starter_project_id}", headers=logged_in_headers, ) # Verify response assert response.status_code == status.HTTP_200_OK, response.text assert response.headers["Content-Type"] == "application/x-zip-compressed" assert "attachment" in response.headers["Content-Disposition"] assert "filename" in response.headers["Content-Disposition"] # The filename is URL-encoded in the header, so check for the project name content_disposition = response.headers["Content-Disposition"] assert ( STARTER_FOLDER_NAME.replace(" ", "%20") in content_disposition or STARTER_FOLDER_NAME.replace(" ", "_") in content_disposition ) # Verify zip file contents zip_content = response.content with zipfile.ZipFile(io.BytesIO(zip_content), "r") as zip_file: file_names = zip_file.namelist() # Should have 3 flow files assert len(file_names) == 3, f"Expected 3 files in zip, got {len(file_names)}: {file_names}" # Verify each basic flow file exists and contains valid JSON for i in range(2): expected_filename = f"Starter Flow {i + 1}.json" assert expected_filename in file_names, f"Expected {expected_filename} in zip file" # Read and verify flow content flow_content = zip_file.read(expected_filename) flow_json = json.loads(flow_content) assert flow_json["name"] == f"Starter Flow {i + 1}" assert flow_json["description"] == f"Test starter flow {i + 1}" # Verify the flow with Note component note_flow_filename = "Flow with Note.json" assert note_flow_filename in file_names, f"Expected {note_flow_filename} in zip file" # Read and verify the Note flow - this tests the bug fix note_flow_content = zip_file.read(note_flow_filename) note_flow_json = json.loads(note_flow_content) assert note_flow_json["name"] == "Flow with Note" assert note_flow_json["description"] == "Flow with Note component and API keys" # Verify the flow has the expected structure assert "data" in note_flow_json assert "nodes" in note_flow_json["data"] assert len(note_flow_json["data"]["nodes"]) == 2 # Find the API node and verify API key was removed api_node = None note_node = None for node in note_flow_json["data"]["nodes"]: if node["id"] == "api-node-1": api_node = node elif node["id"] == "note-1": note_node = node # Verify Note node exists and didn't cause errors (the bug fix) assert note_node is not None, "Note node should exist in downloaded flow" note_template = note_node["data"]["node"]["template"] assert "backgroundColor" in note_template, "Note backgroundColor should be preserved" assert "text" in note_template, "Note text should be preserved" # Verify API key was removed but regular field was kept assert api_node is not None, "API node should exist in downloaded flow" api_template = api_node["data"]["node"]["template"] assert "api_key" in api_template, "API key field should exist" assert api_template["api_key"]["value"] is None, "API key value should be removed/null" assert "regular_field" in api_template, "Regular field should be preserved" assert api_template["regular_field"]["value"] == "keep-this", "Regular field value should be kept" # Clean up: delete the project (which will cascade delete flows) delete_response = await client.delete(f"api/v1/projects/{starter_project_id}", headers=logged_in_headers) assert delete_response.status_code == status.HTTP_204_NO_CONTENT async def test_download_project_missing_returns_404(client: AsyncClient, logged_in_headers): response = await client.get(f"api/v1/projects/download/{uuid4()}", headers=logged_in_headers) assert response.status_code == status.HTTP_404_NOT_FOUND assert response.json()["detail"] == "Project not found" async def test_download_project_with_no_flows_returns_404(client: AsyncClient, logged_in_headers, basic_case): response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert response.status_code == status.HTTP_201_CREATED project_id = response.json()["id"] download_response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers) assert download_response.status_code == status.HTTP_404_NOT_FOUND assert download_response.json()["detail"] == "No flows found in project" async def test_download_project_sanitizes_windows_path_characters( client: AsyncClient, logged_in_headers, basic_case, active_user ): create_response = await client.post("api/v1/projects/", json=basic_case, headers=logged_in_headers) assert create_response.status_code == status.HTTP_201_CREATED project_id = create_response.json()["id"] async with session_scope() as session: flow_create = FlowCreate( name=r"..\evil\flow", description="Flow with unsafe filename characters", data={"nodes": [], "edges": []}, folder_id=project_id, user_id=active_user.id, ) flow = Flow.model_validate(flow_create.model_dump(exclude={"id"})) session.add(flow) await session.flush() await session.refresh(flow) await session.commit() response = await client.get(f"api/v1/projects/download/{project_id}", headers=logged_in_headers) assert response.status_code == status.HTTP_200_OK with zipfile.ZipFile(io.BytesIO(response.content), "r") as zip_file: file_names = zip_file.namelist() assert len(file_names) == 1 assert "/" not in file_names[0] assert "\\" not in file_names[0] assert ".." not in file_names[0] assert file_names[0].endswith(".json") async def _create_other_user(client: AsyncClient) -> tuple[str, dict]: from langflow.services.auth.utils import get_password_hash from langflow.services.database.models.user.model import User user_id = str(uuid4()) username = f"other_user_{user_id[:8]}" async with session_scope() as session: user = User( username=username, password=get_password_hash("testpassword"), # pragma: allowlist secret is_active=True, is_superuser=False, ) session.add(user) await session.commit() await session.refresh(user) created_id = str(user.id) response = await client.post( "api/v1/login", data={"username": username, "password": "testpassword"} ) # pragma: allowlist secret assert response.status_code == 200 token = response.json()["access_token"] return created_id, {"Authorization": f"Bearer {token}"} async def _attach_deployment_to_flow(*, user_id: UUID, flow_id: UUID, project_id: UUID) -> None: """Create provider/deployment/version/attachment rows for one flow.""" async with session_scope() as session: provider = DeploymentProviderAccount( user_id=user_id, name=f"provider-{flow_id.hex[:8]}", provider_tenant_id="tenant-1", provider_key=DeploymentProviderKey.WATSONX_ORCHESTRATE, provider_url=f"https://provider-{flow_id.hex[:8]}.example.com", api_key="encrypted-value", # pragma: allowlist secret ) session.add(provider) await session.flush() deployment = Deployment( user_id=user_id, project_id=project_id, deployment_provider_account_id=provider.id, resource_key=f"rk-{flow_id.hex[:8]}", name=f"deployment-{flow_id.hex[:8]}", deployment_type=DeploymentType.AGENT, ) session.add(deployment) await session.flush() flow_version = FlowVersion( flow_id=flow_id, user_id=user_id, version_number=1, data={"nodes": [], "edges": []}, ) session.add(flow_version) await session.flush() attachment = FlowVersionDeploymentAttachment( user_id=user_id, flow_version_id=flow_version.id, deployment_id=deployment.id, provider_snapshot_id=f"snapshot-{flow_id.hex[:8]}", ) session.add(attachment) await session.commit() async def test_create_project_does_not_reassign_other_users_flows( client: AsyncClient, logged_in_headers: dict, ): """Test that flows_list in create_project only moves flows owned by the requesting user.""" _, other_user_headers = await _create_other_user(client) flow_resp = await client.post( "api/v1/flows/", json={"name": "user-flow", "data": {}}, headers=logged_in_headers, ) assert flow_resp.status_code == status.HTTP_201_CREATED flow_id = flow_resp.json()["id"] original_folder_id = flow_resp.json()["folder_id"] proj_resp = await client.post( "api/v1/projects/", json={"name": "other-project", "flows_list": [flow_id]}, headers=other_user_headers, ) assert proj_resp.status_code == status.HTTP_201_CREATED other_project_id = proj_resp.json()["id"] flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers) assert flow_after.status_code == status.HTTP_200_OK assert flow_after.json()["folder_id"] == original_folder_id proj_detail = await client.get(f"api/v1/projects/{other_project_id}", headers=other_user_headers) assert proj_detail.status_code == status.HTTP_200_OK assert len(proj_detail.json().get("flows", [])) == 0 async def test_read_project_paginated_only_returns_current_users_flows( client: AsyncClient, logged_in_headers: dict, ): """Test that paginated GET /projects/{id} does not return flows owned by other users.""" _, other_user_headers = await _create_other_user(client) flow_resp = await client.post( "api/v1/flows/", json={"name": "user-flow-paginated", "data": {}}, headers=logged_in_headers, ) assert flow_resp.status_code == status.HTTP_201_CREATED flow_id = flow_resp.json()["id"] original_folder_id = flow_resp.json()["folder_id"] proj_resp = await client.post( "api/v1/projects/", json={"name": "other-project-paginated", "flows_list": [flow_id]}, headers=other_user_headers, ) assert proj_resp.status_code == status.HTTP_201_CREATED other_project_id = proj_resp.json()["id"] paginated = await client.get( f"api/v1/projects/{other_project_id}", params={"page": 1, "size": 50}, headers=other_user_headers, ) assert paginated.status_code == status.HTTP_200_OK items = paginated.json().get("flows", {}).get("items", []) assert all(item["id"] != flow_id for item in items) flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers) assert flow_after.json()["folder_id"] == original_folder_id async def test_create_project_with_own_flows_assigns_them_correctly( client: AsyncClient, logged_in_headers: dict, ): """Test that flows_list in create_project correctly assigns flows owned by the requesting user.""" flow_resp = await client.post( "api/v1/flows/", json={"name": "my-flow", "data": {}}, headers=logged_in_headers, ) assert flow_resp.status_code == status.HTTP_201_CREATED flow_id = flow_resp.json()["id"] proj_resp = await client.post( "api/v1/projects/", json={"name": "my-project", "flows_list": [flow_id]}, headers=logged_in_headers, ) assert proj_resp.status_code == status.HTTP_201_CREATED project_id = proj_resp.json()["id"] flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers) assert flow_after.status_code == status.HTTP_200_OK assert flow_after.json()["folder_id"] == project_id paginated = await client.get( f"api/v1/projects/{project_id}", params={"page": 1, "size": 50}, headers=logged_in_headers, ) assert paginated.status_code == status.HTTP_200_OK items = paginated.json().get("flows", {}).get("items", []) assert any(item["id"] == flow_id for item in items) async def test_create_project_with_deployed_flow_returns_409_guard( client: AsyncClient, logged_in_headers: dict, active_user, ): flow_resp = await client.post( "api/v1/flows/", json={"name": "deployed-flow", "data": {"nodes": [], "edges": []}}, headers=logged_in_headers, ) assert flow_resp.status_code == status.HTTP_201_CREATED flow_payload = flow_resp.json() flow_id = UUID(flow_payload["id"]) source_project_id = UUID(flow_payload["folder_id"]) await _attach_deployment_to_flow( user_id=active_user.id, flow_id=flow_id, project_id=source_project_id, ) create_resp = await client.post( "api/v1/projects/", json={"name": "new-target", "flows_list": [str(flow_id)], "components_list": []}, headers=logged_in_headers, ) assert create_resp.status_code == status.HTTP_409_CONFLICT assert "cannot be moved to another project" in create_resp.json()["detail"] async def test_update_project_with_deployed_component_returns_409_guard( client: AsyncClient, logged_in_headers: dict, active_user, ): project_resp = await client.post( "api/v1/projects/", json={"name": "component-project", "description": "", "flows_list": [], "components_list": []}, headers=logged_in_headers, ) assert project_resp.status_code == status.HTTP_201_CREATED project_id = project_resp.json()["id"] component_resp = await client.post( "api/v1/flows/", json={ "name": "deployed-component", "folder_id": project_id, "is_component": True, "data": {"nodes": [], "edges": []}, }, headers=logged_in_headers, ) assert component_resp.status_code == status.HTTP_201_CREATED component_id = UUID(component_resp.json()["id"]) await _attach_deployment_to_flow( user_id=active_user.id, flow_id=component_id, project_id=UUID(project_id), ) update_resp = await client.patch( f"api/v1/projects/{project_id}", json={"name": "component-project-renamed"}, headers=logged_in_headers, ) assert update_resp.status_code == status.HTTP_409_CONFLICT assert "cannot be moved to another project" in update_resp.json()["detail"] async def test_delete_project_with_deployments_returns_409_project_guard( client: AsyncClient, logged_in_headers: dict, active_user, ): project_resp = await client.post( "api/v1/projects/", json={"name": "delete-guard-project", "description": "", "flows_list": [], "components_list": []}, headers=logged_in_headers, ) assert project_resp.status_code == status.HTTP_201_CREATED project_id = project_resp.json()["id"] flow_resp = await client.post( "api/v1/flows/", json={ "name": "delete-guard-flow", "folder_id": project_id, "data": {"nodes": [], "edges": []}, }, headers=logged_in_headers, ) assert flow_resp.status_code == status.HTTP_201_CREATED flow_id = UUID(flow_resp.json()["id"]) await _attach_deployment_to_flow( user_id=active_user.id, flow_id=flow_id, project_id=UUID(project_id), ) delete_resp = await client.delete(f"api/v1/projects/{project_id}", headers=logged_in_headers) assert delete_resp.status_code == status.HTTP_409_CONFLICT detail = delete_resp.json()["detail"] assert "project cannot be deleted because it has deployments" in detail.lower() assert "flow cannot be deleted because it has deployed versions" not in detail.lower()