"""Comprehensive unit tests for V2 Workflow API endpoints. This test module provides extensive coverage of the workflow execution endpoints, including authentication, authorization, error handling, and execution modes. Test Coverage: - Developer API protection (enabled/disabled scenarios) - API key authentication requirements - Flow validation and error handling - Database error handling - Execution timeout protection - Synchronous execution with various component types - Error response structure validation - Multiple execution modes (sync, stream, background) Test Organization: - TestWorkflowDeveloperAPIProtection: Tests developer API feature flag - TestWorkflowErrorHandling: Tests comprehensive error scenarios - TestWorkflowSyncExecution: Tests successful execution flows Test Strategy: - Uses real database with proper cleanup - Mocks external dependencies (LLM APIs, file operations) - Tests both success and failure paths - Validates response structure and status codes """ import asyncio from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID, uuid4 import pytest from httpx import AsyncClient from langflow.exceptions.api import WorkflowValidationError from langflow.services.database.models.flow.model import Flow from langflow.services.database.models.jobs.model import Job, JobType from lfx.schema.workflow import JobStatus from lfx.services.deps import session_scope from sqlalchemy.exc import OperationalError class TestWorkflowDeveloperAPIProtection: """Test developer API protection for workflow endpoints.""" @pytest.fixture def mock_settings_dev_api_disabled(self): """Mock settings with developer API disabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = False mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_execute_workflow_blocked_when_dev_api_disabled( self, client: AsyncClient, created_api_key, mock_settings_dev_api_disabled, # noqa: ARG002 ): """Test workflow execution is blocked when developer API is disabled.""" request_data = { "flow_id": "550e8400-e29b-41d4-a716-446655440000", "background": False, "stream": False, "inputs": None, } headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows", json=request_data, headers=headers, ) assert response.status_code == 403 result = response.json() assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" assert "Developer API" in result["detail"]["message"] async def test_stop_workflow_blocked_when_dev_api_disabled( self, client: AsyncClient, created_api_key, mock_settings_dev_api_disabled, # noqa: ARG002 ): """Test POST workflows/stop endpoint is blocked when developer API is disabled.""" request_data = {"job_id": "550e8400-e29b-41d4-a716-446655440001"} headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json=request_data, headers=headers, ) assert response.status_code == 403 result = response.json() assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_execute_workflow_allowed_when_dev_api_enabled_flow_not_found( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST workflow execution is allowed when developer API is enabled - flow not found.""" request_data = { "flow_id": "550e8400-e29b-41d4-a716-446655440000", # Non-existent flow ID "background": False, "stream": False, "inputs": None, } headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows", json=request_data, headers=headers, ) # Should return 404 because flow doesn't exist, NOT because endpoint is disabled assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "FLOW_NOT_FOUND" assert "550e8400-e29b-41d4-a716-446655440000" in result["detail"]["flow_id"] async def test_get_workflow_allowed_when_dev_api_enabled_job_not_found( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET workflow endpoint is allowed when developer API is enabled - job not found.""" headers = {"x-api-key": created_api_key.api_key} response = await client.get( "api/v2/workflows?job_id=550e8400-e29b-41d4-a716-446655440001", # Non-existent job ID headers=headers, ) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert "550e8400-e29b-41d4-a716-446655440001" in result["detail"]["job_id"] assert "This endpoint is not available" not in response.text async def test_stop_workflow_allowed_when_dev_api_enabled_job_not_found( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST workflow/stop endpoint is allowed when developer API is enabled - job not found.""" request_data = { "job_id": "550e8400-e29b-41d4-a716-446655440001" # Non-existent job ID } headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json=request_data, headers=headers, ) # Should return 404 because job doesn't exist, NOT because endpoint is disabled assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert "550e8400-e29b-41d4-a716-446655440001" in result["detail"]["job_id"] assert "This endpoint is not available" not in response.text async def test_get_workflow_blocked_when_dev_api_disabled( self, client: AsyncClient, created_api_key, mock_settings_dev_api_disabled, # noqa: ARG002 ): """Test GET workflow endpoint is blocked when developer API is disabled.""" headers = {"x-api-key": created_api_key.api_key} response = await client.get( "api/v2/workflows?job_id=550e8400-e29b-41d4-a716-446655440001", headers=headers, ) assert response.status_code == 403 result = response.json() assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" async def test_execute_workflow_allowed_when_dev_api_enabled_flow_exists( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST /workflow allowed when dev API enabled - flow exists and executes.""" flow_id = uuid4() # Create a flow in the database using the established pattern async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Test flow for API testing", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow.id), "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows", json=request_data, headers=headers, ) # Should return 200 because flow is valid (empty nodes/edges is valid) # The execution will complete successfully with no outputs assert response.status_code == 200 result = response.json() # Verify response contains expected fields with proper structure assert "outputs" in result or "errors" in result if "outputs" in result: assert isinstance(result["outputs"], dict) if "errors" in result: assert isinstance(result["errors"], list) finally: # Clean up the flow following established pattern async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_get_workflow_allowed_when_dev_api_enabled_job_exists( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow allowed when dev API enabled - job exists (501 not implemented).""" # Since job management isn't implemented, we'll test with any job_id # The endpoint should return 501 regardless of whether the job exists headers = {"x-api-key": created_api_key.api_key} response = await client.get( "api/v2/workflows?job_id=550e8400-e29b-41d4-a716-446655440002", headers=headers, ) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert "This endpoint is not available" not in response.text async def test_stop_workflow_allowed_when_dev_api_enabled_job_exists( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST /workflow/stop allowed when dev API enabled - job exists (501 not implemented).""" # Since job management isn't implemented, we'll test with any job_id # The endpoint should return 501 regardless of whether the job exists request_data = {"job_id": "550e8400-e29b-41d4-a716-446655440002"} headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json=request_data, headers=headers, ) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert "This endpoint is not available" not in response.text async def test_all_endpoints_require_api_key_authentication( self, client: AsyncClient, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that all workflow endpoints require API key authentication.""" # Test POST /workflow without API key request_data = { "flow_id": "550e8400-e29b-41d4-a716-446655440000", "background": False, "stream": False, "inputs": None, } response = await client.post( "api/v2/workflows", json=request_data, ) # The API returns 403 Forbidden for missing API keys (not 401 Unauthorized) # This is the correct behavior according to the api_key_security implementation assert response.status_code == 403 assert "API key must be passed" in response.json()["detail"] class TestWorkflowErrorHandling: """Test comprehensive error handling for workflow endpoints.""" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_flow_not_found_returns_404_with_error_code( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that non-existent flow returns 404 with FLOW_NOT_FOUND error code.""" flow_id = str(uuid4()) request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "FLOW_NOT_FOUND" assert result["detail"]["flow_id"] == flow_id assert "Verify the flow_id" in result["detail"]["message"] async def test_database_error_returns_503( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that database errors return 503 with DATABASE_ERROR code.""" flow_id = str(uuid4()) request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} # Mock get_flow_by_id_or_endpoint_name to raise OperationalError with patch("langflow.api.v2.workflow.get_flow_by_id_or_endpoint_name") as mock_get_flow: mock_get_flow.side_effect = OperationalError("statement", "params", "orig") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 503 result = response.json() assert result["detail"]["code"] == "DATABASE_ERROR" assert "Failed to fetch flow" in result["detail"]["message"] assert result["detail"]["flow_id"] == flow_id async def test_flow_with_no_data_returns_500( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that flow with no data returns 500 with INVALID_FLOW_DATA code.""" flow_id = uuid4() # Create a flow with no data async with session_scope() as session: flow = Flow( id=flow_id, name="Flow with no data", description="Test flow with no data", data=None, # No data user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 400 result = response.json() assert result["detail"]["code"] == "INVALID_FLOW_DATA" assert "has no data" in result["detail"]["message"] assert result["detail"]["flow_id"] == str(flow_id) finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_graph_build_failure_returns_500( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that graph build failure returns 500 with INVALID_FLOW_DATA code.""" flow_id = uuid4() # Create a flow with invalid data that will fail validation/graph building async with session_scope() as session: flow = Flow( id=flow_id, name="Flow with invalid data", description="Test flow with invalid graph data", data={"invalid": "data"}, # Invalid graph data (missing 'nodes' field) user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 400 result = response.json() assert result["detail"]["code"] == "INVALID_FLOW_DATA" # The error message should indicate invalid flow data structure error_msg = result["detail"]["message"].lower() assert "invalid data structure" in error_msg or "must have nodes" in error_msg assert result["detail"]["flow_id"] == str(flow_id) finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_execution_timeout_with_real_delay( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that execution timeout works with real async delay.""" flow_id = uuid4() # Create a valid flow async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Test flow for timeout", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} # Mock execute_sync_workflow to sleep longer than timeout async def slow_execution(*args, **kwargs): # noqa: ARG001 await asyncio.sleep(2) # Sleep for 2 seconds return MagicMock() # Temporarily reduce timeout for testing with ( patch("langflow.api.v2.workflow.execute_sync_workflow", side_effect=slow_execution), patch("langflow.api.v2.workflow.EXECUTION_TIMEOUT", 0.5), # 0.5 second timeout ): headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 408 result = response.json() assert result["detail"]["code"] == "EXECUTION_TIMEOUT" assert "exceeded" in result["detail"]["message"] assert result["detail"]["flow_id"] == str(flow_id) assert "job_id" in result["detail"] finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_background_mode_returns_501( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that background mode returns 501 with NOT_IMPLEMENTED code.""" flow_id = uuid4() # Create a valid flow async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Test flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": True, # Background mode "stream": False, "inputs": None, } headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) # Now background mode is partially implemented and should NOT return 501 # It should return a WorkflowJobResponse (wrapped in WorkflowExecutionResponse or similar) assert response.status_code == 200 result = response.json() assert result["object"] == "job" assert result["status"] == "queued" assert result["flow_id"] == str(flow_id) assert "links" in result assert "status" in result["links"] finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_streaming_mode_returns_501( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that streaming mode returns 501 with NOT_IMPLEMENTED code.""" flow_id = uuid4() # Create a valid flow async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Test flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": False, "stream": True, # Streaming mode "inputs": None, } headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 501 result = response.json() assert result["detail"]["code"] == "NOT_IMPLEMENTED" assert "Streaming execution not yet implemented" in result["detail"]["message"] finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_error_response_structure( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that all error responses have consistent structure.""" flow_id = str(uuid4()) request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 404 result = response.json() # Verify error structure assert "detail" in result assert "error" in result["detail"] assert "code" in result["detail"] assert "message" in result["detail"] assert "flow_id" in result["detail"] # Verify types assert isinstance(result["detail"]["error"], str) assert isinstance(result["detail"]["code"], str) assert isinstance(result["detail"]["message"], str) assert isinstance(result["detail"]["flow_id"], str) async def test_workflow_validation_error_propagation( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that WorkflowValidationError is properly caught and converted to 500.""" flow_id = uuid4() # Create a flow async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Test flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} # Mock execute_sync_workflow to raise WorkflowValidationError with patch("langflow.api.v2.workflow.execute_sync_workflow") as mock_execute: mock_execute.side_effect = WorkflowValidationError("Test validation error") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 400 result = response.json() assert result["detail"]["code"] == "INVALID_FLOW_DATA" assert "Test validation error" in result["detail"]["message"] finally: # Clean up async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) # Test GET /workflow without API key response = await client.get("api/v2/workflows?job_id=550e8400-e29b-41d4-a716-446655440001") assert response.status_code == 403 assert "API key must be passed" in response.json()["detail"] class TestWorkflowSyncExecution: """Test synchronous workflow execution with realistic component mocking.""" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_sync_execution_with_empty_flow_returns_200( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test sync execution with empty flow returns 200 with empty outputs.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Empty Flow", description="Flow with no nodes", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() # Verify response structure assert "flow_id" in result assert result["flow_id"] == str(flow_id) assert "job_id" in result # Verify outputs or errors are present with actual content assert "outputs" in result or "errors" in result if "outputs" in result: assert isinstance(result["outputs"], dict) if "errors" in result: assert isinstance(result["errors"], list) # session_id is only present if provided in inputs finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_component_error_returns_200_with_error_in_body( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that component execution errors return 200 with error in response body.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Flow for testing component errors", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} # Mock run_graph_internal to raise a component execution error with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: mock_run.side_effect = Exception("Component execution failed: LLM API key not configured") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) # Component errors should return 200 with error in body assert response.status_code == 200 result = response.json() # Verify error is in response body (via create_error_response) assert "errors" in result assert len(result["errors"]) > 0 assert "Component execution failed" in str(result["errors"][0]) assert result["status"] == "failed" assert result["flow_id"] == str(flow_id) assert "job_id" in result finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_with_chat_input_output( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test sync execution with ChatInput and ChatOutput components.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Chat Flow", description="Flow with chat input/output", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: # Input format: component_id.param = value request_data = { "flow_id": str(flow_id), "background": False, "stream": False, "inputs": { "ChatInput-abc123.input_value": "Hello, how are you?", "ChatInput-abc123.session_id": "session-456", }, } # Mock successful execution with ChatOutput mock_result_data = MagicMock() mock_result_data.component_id = "ChatOutput-xyz789" mock_result_data.outputs = {"message": {"message": "I'm doing well, thank you for asking!", "type": "text"}} mock_result_data.metadata = {} # Wrap ResultData in RunOutputs mock_run_output = MagicMock() mock_run_output.outputs = [mock_result_data] with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: # run_graph_internal returns tuple[list[RunOutputs], str] mock_run.return_value = ([mock_run_output], "session-456") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() # Verify response structure assert result["flow_id"] == str(flow_id) assert "job_id" in result assert "outputs" in result # Note: Detailed content validation requires proper graph/vertex mocking # which is beyond the scope of unit tests. Integration tests should validate content. # Verify inputs were echoed back assert "inputs" in result assert result["inputs"] == request_data["inputs"] # Verify session_id is present when provided in inputs if "session_id" in result: assert result["session_id"] == "session-456" finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_with_llm_output( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test sync execution with LLM component output including model metadata.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="LLM Flow", description="Flow with LLM component", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": False, "stream": False, "inputs": { "ChatInput-abc.input_value": "Explain quantum computing", "OpenAIModel-def.temperature": 0.7, }, } # Mock LLM execution with model metadata mock_result_data = MagicMock() mock_result_data.component_id = "OpenAIModel-def" mock_result_data.outputs = { "model_output": { "message": { "message": "Quantum computing uses quantum mechanics...", "model_name": "gpt-4", "type": "text", } } } mock_result_data.metadata = {"tokens_used": 150} # Wrap ResultData in RunOutputs mock_run_output = MagicMock() mock_run_output.outputs = [mock_result_data] with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: # run_graph_internal returns tuple[list[RunOutputs], str] mock_run.return_value = ([mock_run_output], "session-789") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() assert result["flow_id"] == str(flow_id) assert "job_id" in result assert "outputs" in result # Note: Detailed content validation requires proper graph/vertex mocking finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_with_file_save_output( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test sync execution with SaveToFile component.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="File Save Flow", description="Flow with file save component", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": False, "stream": False, "inputs": { "TextInput-abc.text": "Content to save", "SaveToFile-xyz.file_path": "/tmp/output.txt", # noqa: S108 }, } # Mock SaveToFile execution mock_result_data = MagicMock() mock_result_data.component_id = "SaveToFile-xyz" mock_result_data.outputs = { "message": {"message": "File saved successfully to /tmp/output.txt", "type": "text"} } mock_result_data.metadata = {"bytes_written": 1024} # Wrap ResultData in RunOutputs mock_run_output = MagicMock() mock_run_output.outputs = [mock_result_data] with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: # run_graph_internal returns tuple[list[RunOutputs], str] mock_run.return_value = ([mock_run_output], "session-101") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() assert result["flow_id"] == str(flow_id) assert "job_id" in result assert "outputs" in result # Note: Detailed content validation requires proper graph/vertex mocking finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_with_multiple_terminal_nodes( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test sync execution with multiple terminal nodes (outputs).""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Multi-Output Flow", description="Flow with multiple terminal nodes", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": False, "stream": False, "inputs": {"ChatInput-abc.input_value": "Process this"}, } # Mock execution with multiple outputs mock_chat_output = MagicMock() mock_chat_output.component_id = "ChatOutput-aaa" mock_chat_output.outputs = {"message": {"message": "Chat response", "type": "text"}} mock_file_output = MagicMock() mock_file_output.component_id = "SaveToFile-bbb" mock_file_output.outputs = {"message": {"message": "File saved successfully", "type": "text"}} with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: # run_graph_internal returns tuple[list[RunOutputs], str] mock_run.return_value = ([mock_chat_output, mock_file_output], "session-202") headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() assert result["flow_id"] == str(flow_id) assert "job_id" in result assert "outputs" in result # Note: Detailed content validation requires proper graph/vertex mocking finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_response_structure_validation( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test that sync execution response has correct WorkflowExecutionResponse structure.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Test Flow", description="Flow for response validation", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() # Verify WorkflowExecutionResponse structure assert "flow_id" in result assert isinstance(result["flow_id"], str) assert result["flow_id"] == str(flow_id) assert "job_id" in result assert isinstance(result["job_id"], str) # session_id is optional - only present if provided in inputs if "session_id" in result: assert isinstance(result["session_id"], str) assert "object" in result assert result["object"] == "response" assert "created_timestamp" in result assert isinstance(result["created_timestamp"], str) assert "status" in result assert result["status"] in ["completed", "failed", "running", "queued"] assert "errors" in result assert isinstance(result["errors"], list) assert "inputs" in result assert isinstance(result["inputs"], dict) assert "outputs" in result assert isinstance(result["outputs"], dict) finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) # Test POST /workflow/stop without API key response = await client.post( "api/v2/workflows/stop", json={"job_id": "550e8400-e29b-41d4-a716-446655440001"}, ) assert response.status_code == 403 assert "API key must be passed" in response.json()["detail"] class TestWorkflowBackgroundQueueing: """Test background workflow execution and queueing behavior.""" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_background_execution_flow( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test the full background job submission flow.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Background Flow", description="Flow for background testing", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() await session.refresh(flow) try: request_data = { "flow_id": str(flow_id), "background": True, "inputs": {"test.input": "data"}, } headers = {"x-api-key": created_api_key.api_key} # Mock uuid4 to return a predictable job_id mock_job_id = "550e8400-e29b-41d4-a716-446655440001" with ( patch("langflow.api.v2.workflow.get_task_service") as mock_get_task_service, patch("langflow.api.v2.workflow.uuid4", return_value=UUID(mock_job_id)), ): mock_task_service = MagicMock() # fire_and_forget_task is now awaited but its return value is not used for the job_id in response # as it uses graph.run_id. However, we still need it to be an awaitable if it's awaited. mock_task_service.fire_and_forget_task.return_value = asyncio.Future() mock_task_service.fire_and_forget_task.return_value.set_result(mock_job_id) mock_get_task_service.return_value = mock_task_service response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() assert result["job_id"] == mock_job_id assert result["flow_id"] == str(flow_id) assert result["object"] == "job" assert result["status"] == "queued" assert "links" in result assert "status" in result["links"] assert mock_job_id in result["links"]["status"] finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_background_execution_invalid_flow( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test background execution with a non-existent flow ID.""" request_data = { "flow_id": str(uuid4()), "background": True, } headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 404 detail = response.json()["detail"] message = detail["message"] if isinstance(detail, dict) else detail assert "does not exist" in message.lower() async def test_background_execution_queue_exception( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test handling of exceptions during task queueing.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Fail Flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() try: request_data = {"flow_id": str(flow_id), "background": True} headers = {"x-api-key": created_api_key.api_key} with patch("langflow.api.v2.workflow.get_task_service") as mock_get_task_service: mock_task_service = MagicMock() mock_task_service.fire_and_forget_task.side_effect = Exception("Queueing failed") mock_get_task_service.return_value = mock_task_service response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 500 detail = response.json()["detail"] message = detail["message"] if isinstance(detail, dict) else detail assert "Queueing failed" in message finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) async def test_sync_execution_error_handling( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test error handling during synchronous execution.""" flow_id = uuid4() async with session_scope() as session: flow = Flow( id=flow_id, name="Error Flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() try: request_data = {"flow_id": str(flow_id), "background": False} headers = {"x-api-key": created_api_key.api_key} with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: mock_run.side_effect = Exception("Internal execution engine failure") response = await client.post("api/v2/workflows", json=request_data, headers=headers) assert response.status_code == 200 result = response.json() assert "errors" in result assert "Internal execution engine failure" in str(result["errors"]) finally: async with session_scope() as session: flow = await session.get(Flow, flow_id) if flow: await session.delete(flow) class TestWorkflowStatus: """Test workflow status retrieval endpoints.""" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_get_status_queued( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow returns 200 for a queued job.""" job_id = uuid4() flow_id = uuid4() mock_job = MagicMock() mock_job.job_id = job_id mock_job.flow_id = flow_id mock_job.status = JobStatus.QUEUED mock_job.type = JobType.WORKFLOW mock_job.user_id = None mock_job.created_timestamp = datetime.now(timezone.utc) with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 200 result = response.json() assert result["job_id"] == str(job_id) assert result["status"] == "queued" assert result["flow_id"] == str(flow_id) async def test_get_status_not_found( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow returns 404 for a non-existent job.""" job_id = uuid4() with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=None) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" async def test_get_status_failed( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow returns 500 for a failed job.""" job_id = uuid4() mock_job = MagicMock() mock_job.job_id = job_id mock_job.status = JobStatus.FAILED mock_job.type = JobType.WORKFLOW mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 500 result = response.json() assert result["detail"]["code"] == "JOB_FAILED" assert result["detail"]["job_id"] == str(job_id) async def test_get_status_completed_reconstruction( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow returns reconstructed response for a completed job.""" job_id = uuid4() flow_id = uuid4() mock_job = MagicMock() mock_job.job_id = job_id mock_job.flow_id = flow_id mock_job.status = JobStatus.COMPLETED mock_job.type = JobType.WORKFLOW mock_job.user_id = None with ( patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service, patch("langflow.api.v2.workflow.get_flow_by_id_or_endpoint_name") as mock_get_flow, patch("langflow.api.v2.workflow.reconstruct_workflow_response_from_job_id") as mock_reconstruct, ): mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_get_job_service.return_value = mock_service mock_flow = MagicMock() mock_flow.id = flow_id mock_get_flow.return_value = mock_flow mock_reconstruct.return_value = {"flow_id": str(flow_id), "status": "completed", "outputs": {}} headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 200 result = response.json() assert result["status"] == "completed" mock_reconstruct.assert_called_once() async def test_get_status_timed_out( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test GET /workflow returns 408 for a timed out job.""" job_id = uuid4() flow_id = uuid4() mock_job = MagicMock() mock_job.job_id = job_id mock_job.flow_id = flow_id mock_job.status = JobStatus.TIMED_OUT mock_job.type = JobType.WORKFLOW mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} # Add timeout to client.get to avoid hanging if something goes wrong response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 408 result = response.json() assert result["detail"]["code"] == "EXECUTION_TIMEOUT" assert result["detail"]["job_id"] == str(job_id) assert result["detail"]["flow_id"] == str(flow_id) class TestWorkflowStop: """Test workflow stop endpoints.""" @pytest.fixture def mock_settings_dev_api_enabled(self): """Mock settings with developer API enabled.""" with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings_service.return_value = mock_service yield mock_settings async def test_stop_workflow_success( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST /workflow/stop cancels a running job.""" job_id = str(uuid4()) mock_job = MagicMock() mock_job.job_id = job_id mock_job.status = JobStatus.IN_PROGRESS mock_job.type = JobType.WORKFLOW mock_job.user_id = None with ( patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service, patch("langflow.api.v2.workflow.get_task_service") as mock_get_task_service, ): mock_job_service = MagicMock() mock_job_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_job_service.update_job_status = AsyncMock() mock_get_job_service.return_value = mock_job_service mock_task_service = MagicMock() mock_task_service.revoke_task = AsyncMock(return_value=True) mock_get_task_service.return_value = mock_task_service headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows/stop", json={"job_id": job_id}, headers=headers) assert response.status_code == 200 result = response.json() assert result["job_id"] == job_id assert "cancelled successfully" in result["message"] mock_task_service.revoke_task.assert_called_once() mock_job_service.update_job_status.assert_called_once_with(UUID(job_id), JobStatus.CANCELLED) async def test_stop_workflow_not_found( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST /workflow/stop returns 404 for non-existent job.""" job_id = str(uuid4()) with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=None) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows/stop", json={"job_id": job_id}, headers=headers) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" async def test_stop_workflow_already_cancelled( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """Test POST /workflow/stop handles already cancelled jobs.""" job_id = str(uuid4()) mock_job = MagicMock() mock_job.job_id = job_id mock_job.status = JobStatus.CANCELLED mock_job.type = JobType.WORKFLOW mock_job.user_id = None with patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service: mock_service = MagicMock() mock_service.get_job_by_job_id = AsyncMock(return_value=mock_job) mock_get_job_service.return_value = mock_service headers = {"x-api-key": created_api_key.api_key} response = await client.post("api/v2/workflows/stop", json={"job_id": job_id}, headers=headers) assert response.status_code == 200 result = response.json() assert result["job_id"] == job_id assert "already cancelled" in result["message"] @pytest.mark.security class TestWorkflowIDORProtection: """Security regression tests for IDOR on workflow job endpoints. Covers GHSA-qfw4-cjhf-3g3q: background workflow jobs are queryable and cancellable cross-user unless an ownership check is enforced. """ @pytest.fixture def mock_settings_dev_api_enabled(self): with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings: mock_service = MagicMock() mock_settings = MagicMock() mock_settings.developer_api_enabled = True mock_service.settings = mock_settings mock_get_settings.return_value = mock_service yield mock_settings @pytest.mark.security async def test_get_workflow_status_forbidden_for_other_user_job( self, client: AsyncClient, created_api_key, created_user_two_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """GET /api/v2/workflows returns 404 when the job belongs to a different user. GHSA-qfw4-cjhf-3g3q: job status must not be visible cross-user. Ownership is enforced at the SQL level — unauthorized access returns 404. """ job_id = uuid4() other_user_id = created_user_two_api_key.user_id async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.QUEUED, type=JobType.WORKFLOW, user_id=other_user_id, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert str(job_id) in result["detail"]["job_id"] finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_get_workflow_status_allowed_for_own_job( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """GET /api/v2/workflows returns 200 when the job belongs to the requesting user.""" job_id = uuid4() owner_user_id = created_api_key.user_id async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.QUEUED, type=JobType.WORKFLOW, user_id=owner_user_id, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 200 result = response.json() assert str(job_id) in result["job_id"] finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_stop_workflow_forbidden_for_other_user_job( self, client: AsyncClient, created_api_key, created_user_two_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """POST /api/v2/workflows/stop returns 404 when the job belongs to a different user. GHSA-qfw4-cjhf-3g3q: job cancellation must not be allowed cross-user. Ownership is enforced at the SQL level — unauthorized access returns 404. """ job_id = uuid4() other_user_id = created_user_two_api_key.user_id async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.IN_PROGRESS, type=JobType.WORKFLOW, user_id=other_user_id, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json={"job_id": str(job_id)}, headers=headers, ) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" assert str(job_id) in result["detail"]["job_id"] finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_stop_workflow_allowed_for_own_job( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """POST /api/v2/workflows/stop succeeds when the job belongs to the requesting user.""" job_id = uuid4() owner_user_id = created_api_key.user_id async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.CANCELLED, type=JobType.WORKFLOW, user_id=owner_user_id, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json={"job_id": str(job_id)}, headers=headers, ) assert response.status_code == 200 finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_get_workflow_status_allowed_for_legacy_job_with_no_user_id( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """GET /api/v2/workflows does NOT block legacy jobs where user_id is NULL. Jobs created before the fix have user_id=None and must not be broken. """ job_id = uuid4() async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.QUEUED, type=JobType.WORKFLOW, user_id=None, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 200 result = response.json() assert result["job_id"] == str(job_id) finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_stop_workflow_allowed_for_legacy_job_with_no_user_id( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """POST /api/v2/workflows/stop does NOT block legacy jobs where user_id is NULL. Jobs created before the ownership fix have user_id=None and must not be broken by the ownership check (parity with the equivalent GET test). """ job_id = uuid4() async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.IN_PROGRESS, type=JobType.WORKFLOW, user_id=None, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json={"job_id": str(job_id)}, headers=headers, ) assert response.status_code == 200, ( "Legacy jobs with user_id=None must not be blocked by the ownership check" ) finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_stop_workflow_returns_404_for_non_workflow_job_type( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """POST /api/v2/workflows/stop returns 404 for non-WORKFLOW job types. Prevents stop endpoint from cancelling ingestion or evaluation jobs. """ job_id = uuid4() async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.IN_PROGRESS, type=JobType.INGESTION, user_id=None, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows/stop", json={"job_id": str(job_id)}, headers=headers, ) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_get_workflow_status_returns_404_for_non_workflow_job_type( self, client: AsyncClient, created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """GET /api/v2/workflows returns 404 for non-WORKFLOW job types. Prevents status endpoint from exposing ingestion or evaluation jobs. """ job_id = uuid4() async with session_scope() as session: job = Job( job_id=job_id, flow_id=uuid4(), status=JobStatus.IN_PROGRESS, type=JobType.INGESTION, user_id=None, ) session.add(job) await session.flush() try: headers = {"x-api-key": created_api_key.api_key} response = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) assert response.status_code == 404 result = response.json() assert result["detail"]["code"] == "JOB_NOT_FOUND" finally: async with session_scope() as session: db_job = await session.get(Job, job_id) if db_job: await session.delete(db_job) @pytest.mark.security async def test_background_job_stores_user_id_and_blocks_cross_user_access( self, client: AsyncClient, created_api_key, created_user_two_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): """End-to-end: POST /workflows (background=true) stores user_id and blocks cross-user GET. This test exercises the real create_job code path to catch regressions where user_id is not passed to create_job (causing user_id=NULL and bypassing the ownership check — the exact bug found in Postman). Steps: 1. Alice creates a background job via the real endpoint (no mock on create_job) 2. Bob queries the job_id returned — must get 404 (ownership enforced at SQL level) """ flow_id = uuid4() job_id_str = None async with session_scope() as session: flow = Flow( id=flow_id, name="Alice End-to-End Flow", data={"nodes": [], "edges": []}, user_id=created_api_key.user_id, ) session.add(flow) await session.flush() try: # Mock only the task service to prevent real background execution with patch("langflow.api.v2.workflow.get_task_service") as mock_task_svc: mock_task_service = MagicMock() mock_task_service.fire_and_forget_task = AsyncMock() mock_task_svc.return_value = mock_task_service alice_headers = {"x-api-key": created_api_key.api_key} response = await client.post( "api/v2/workflows", json={ "flow_id": str(flow_id), "background": True, "stream": False, "inputs": {}, }, headers=alice_headers, ) assert response.status_code in (200, 202), ( f"Expected 200/202 from Alice's background job creation, got {response.status_code}: {response.text}" ) job_id_str = response.json()["job_id"] # Bob queries Alice's job — must be 404 (ownership enforced at SQL level) bob_headers = {"x-api-key": created_user_two_api_key.api_key} response = await client.get( f"api/v2/workflows?job_id={job_id_str}", headers=bob_headers, ) assert response.status_code == 404, ( f"Expected 404 Not Found but got {response.status_code}. " "user_id is not being persisted in create_job — Bob can access Alice's job." ) assert response.json()["detail"]["code"] == "JOB_NOT_FOUND" finally: async with session_scope() as session: if job_id_str: db_job = await session.get(Job, UUID(job_id_str)) if db_job: await session.delete(db_job) db_flow = await session.get(Flow, flow_id) if db_flow: await session.delete(db_flow)