|
| 1 | +"""Tests for partial WorkflowMetaDisplay override serialization.""" |
| 2 | + |
| 3 | +from vellum.workflows import BaseWorkflow |
| 4 | +from vellum.workflows.inputs.base import BaseInputs |
| 5 | +from vellum.workflows.nodes.bases.base import BaseNode |
| 6 | +from vellum.workflows.state.base import BaseState |
| 7 | +from vellum_ee.workflows.display.base import WorkflowDisplayData, WorkflowDisplayDataViewport, WorkflowMetaDisplay |
| 8 | +from vellum_ee.workflows.display.workflows.base_workflow_display import BaseWorkflowDisplay |
| 9 | +from vellum_ee.workflows.display.workflows.get_vellum_workflow_display_class import get_workflow_display |
| 10 | + |
| 11 | + |
| 12 | +def test_triggerless_workflow_with_partial_display_override_creates_entrypoint(): |
| 13 | + """ |
| 14 | + Tests that a triggerless workflow with partial WorkflowMetaDisplay overrides |
| 15 | + (only display_data set) still creates an ENTRYPOINT node with backfilled IDs. |
| 16 | + """ |
| 17 | + |
| 18 | + # GIVEN a simple triggerless workflow |
| 19 | + class ProcessNode(BaseNode): |
| 20 | + pass |
| 21 | + |
| 22 | + class TestWorkflow(BaseWorkflow[BaseInputs, BaseState]): |
| 23 | + graph = ProcessNode |
| 24 | + |
| 25 | + # AND a display class that only overrides display_data (viewport), not entrypoint IDs |
| 26 | + class TestWorkflowDisplay(BaseWorkflowDisplay[TestWorkflow]): |
| 27 | + workflow_display = WorkflowMetaDisplay( |
| 28 | + display_data=WorkflowDisplayData(viewport=WorkflowDisplayDataViewport(x=100.0, y=200.0, zoom=1.5)) |
| 29 | + ) |
| 30 | + |
| 31 | + # WHEN we serialize the workflow |
| 32 | + result = get_workflow_display(workflow_class=TestWorkflow).serialize() |
| 33 | + |
| 34 | + # THEN the workflow should have an ENTRYPOINT node |
| 35 | + workflow_raw_data = result["workflow_raw_data"] |
| 36 | + assert isinstance(workflow_raw_data, dict) |
| 37 | + nodes = workflow_raw_data["nodes"] |
| 38 | + assert isinstance(nodes, list) |
| 39 | + entrypoint_nodes = [n for n in nodes if isinstance(n, dict) and n.get("type") == "ENTRYPOINT"] |
| 40 | + assert len(entrypoint_nodes) == 1, "Triggerless workflow with partial overrides should have an ENTRYPOINT node" |
| 41 | + |
| 42 | + # AND the ENTRYPOINT node should have valid IDs (not None) |
| 43 | + entrypoint_node = entrypoint_nodes[0] |
| 44 | + assert isinstance(entrypoint_node, dict) |
| 45 | + assert entrypoint_node["id"] is not None, "ENTRYPOINT node should have a non-None id" |
| 46 | + source_handle_id = entrypoint_node["data"]["source_handle_id"] |
| 47 | + assert source_handle_id is not None, "ENTRYPOINT node should have a non-None source_handle_id" |
| 48 | + |
| 49 | + # AND there should be an edge from ENTRYPOINT to the process node |
| 50 | + edges = workflow_raw_data["edges"] |
| 51 | + assert isinstance(edges, list) |
| 52 | + entrypoint_edges = [e for e in edges if isinstance(e, dict) and e.get("source_node_id") == entrypoint_node["id"]] |
| 53 | + assert len(entrypoint_edges) > 0, "Should have edges from ENTRYPOINT node" |
0 commit comments