Add tests for human_in_the_loop samples and fix wait_condition graph

- Fix approval_wait_condition graph to use temporal_node_metadata()
- Import workflow inside function for sandboxing
- Add tests for approval_graph_interrupt sample
- Add tests for approval_wait_condition sample
- Remove old approval_workflow_test.py

Author: mfateev

langgraph_samples/activity_from_node/graph.py

@@ -10,7 +10,6 @@
 from langgraph.graph import END, START, StateGraph
 from typing_extensions import TypedDict
 
-
 # =============================================================================
 # State Definition
 # =============================================================================

langgraph_samples/human_in_the_loop/approval_graph_interrupt/run_respond.py

@@ -9,7 +9,9 @@
 from temporalio.client import Client
 from temporalio.envconfig import ClientConfig
 
-from langgraph_samples.human_in_the_loop.approval_graph_interrupt.workflow import ApprovalWorkflow
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.workflow import (
+    ApprovalWorkflow,
+)
 
 
 async def main() -> None:

langgraph_samples/human_in_the_loop/approval_graph_interrupt/run_worker.py

@@ -11,9 +11,15 @@
 from temporalio.envconfig import ClientConfig
 from temporalio.worker import Worker
 
-from langgraph_samples.human_in_the_loop.approval_graph_interrupt.activities import notify_approver
-from langgraph_samples.human_in_the_loop.approval_graph_interrupt.graph import build_approval_graph
-from langgraph_samples.human_in_the_loop.approval_graph_interrupt.workflow import ApprovalWorkflow
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.activities import (
+    notify_approver,
+)
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.graph import (
+    build_approval_graph,
+)
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.workflow import (
+    ApprovalWorkflow,
+)
 
 TASK_QUEUE = "langgraph-approval-interrupt"
 

langgraph_samples/human_in_the_loop/approval_wait_condition/graph.py

@@ -13,7 +13,7 @@
 from typing import Any
 
 from langgraph.graph import END, START, StateGraph
-from temporalio import activity, workflow
+from temporalio import activity
 from typing_extensions import TypedDict
 
 
@@ -112,6 +112,10 @@ async def request_approval(state: ApprovalState) -> ApprovalState:
 
     It waits for an approval signal using workflow.wait_condition().
     """
+    from datetime import timedelta
+
+    from temporalio import workflow
+
     # Get access to the workflow instance
     wf = workflow.instance()
 
@@ -128,7 +132,9 @@ async def request_approval(state: ApprovalState) -> ApprovalState:
     # Store approval request for queries
     wf._pending_approval = approval_request
 
-    workflow.logger.info("Workflow paused for approval: %s", approval_request["message"])
+    workflow.logger.info(
+        "Workflow paused for approval: %s", approval_request["message"]
+    )
 
     # Notify the approver via activity
     await workflow.execute_activity(
@@ -180,12 +186,18 @@ def build_approval_graph() -> Any:
     The request_approval node uses run_in_workflow=True to access
     Temporal operations directly (signals, activities, etc.).
     """
+    from temporalio.contrib.langgraph import temporal_node_metadata
+
     graph = StateGraph(ApprovalState)
 
     # Add nodes
     graph.add_node("process_request", process_request)
     # Mark request_approval as run_in_workflow - it can access Temporal operations
-    graph.add_node("request_approval", request_approval, metadata={"run_in_workflow": True})
+    graph.add_node(
+        "request_approval",
+        request_approval,
+        metadata=temporal_node_metadata(run_in_workflow=True),
+    )
     graph.add_node("execute_action", execute_action)
 
     # Define edges

langgraph_samples/human_in_the_loop/approval_wait_condition/run_respond.py

@@ -9,7 +9,9 @@
 from temporalio.client import Client
 from temporalio.envconfig import ClientConfig
 
-from langgraph_samples.human_in_the_loop.approval_wait_condition.workflow import ApprovalWorkflow
+from langgraph_samples.human_in_the_loop.approval_wait_condition.workflow import (
+    ApprovalWorkflow,
+)
 
 
 async def main() -> None:

langgraph_samples/human_in_the_loop/approval_wait_condition/run_worker.py

@@ -15,7 +15,9 @@
     build_approval_graph,
     notify_approver,
 )
-from langgraph_samples.human_in_the_loop.approval_wait_condition.workflow import ApprovalWorkflow
+from langgraph_samples.human_in_the_loop.approval_wait_condition.workflow import (
+    ApprovalWorkflow,
+)
 
 TASK_QUEUE = "langgraph-approval-condition"
 

…ggraph_samples/approval_workflow_test.py …samples/approval_graph_interrupt_test.pytests/langgraph_samples/approval_workflow_test.py renamed to tests/langgraph_samples/approval_graph_interrupt_test.py tests/langgraph_samples/approval_workflow_test.py renamed to tests/langgraph_samples/approval_graph_interrupt_test.py

@@ -1,20 +1,24 @@
-"""Tests for the approval_workflow LangGraph sample."""
+"""Tests for the approval_graph_interrupt LangGraph sample."""
 
 import uuid
 
 from temporalio.client import Client, WorkflowHandle
 from temporalio.contrib.langgraph import LangGraphPlugin
 from temporalio.worker import Worker
 
-from langgraph_samples.approval_workflow.activities import notify_approver
-from langgraph_samples.approval_workflow.graph import build_approval_graph
-from langgraph_samples.approval_workflow.workflow import (
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.activities import (
+    notify_approver,
+)
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.graph import (
+    build_approval_graph,
+)
+from langgraph_samples.human_in_the_loop.approval_graph_interrupt.workflow import (
     ApprovalRequest,
     ApprovalWorkflow,
 )
 
 
-async def test_approval_workflow_approved(client: Client) -> None:
+async def test_approval_graph_interrupt_approved(client: Client) -> None:
     """Test approval workflow when request is approved."""
     task_queue = f"approval-test-{uuid.uuid4()}"
 
@@ -67,7 +71,7 @@ async def test_approval_workflow_approved(client: Client) -> None:
         assert "manager" in result["result"]
 
 
-async def test_approval_workflow_rejected(client: Client) -> None:
+async def test_approval_graph_interrupt_rejected(client: Client) -> None:
     """Test approval workflow when request is rejected."""
     task_queue = f"approval-test-{uuid.uuid4()}"
 
@@ -113,47 +117,3 @@ async def test_approval_workflow_rejected(client: Client) -> None:
         assert result["executed"] is False
         assert "rejected" in result["result"]
         assert "cfo" in result["result"]
-
-
-async def test_approval_workflow_low_risk(client: Client) -> None:
-    """Test approval workflow with low risk amount."""
-    task_queue = f"approval-test-{uuid.uuid4()}"
-
-    plugin = LangGraphPlugin(graphs={"approval_workflow": build_approval_graph})
-
-    async with Worker(
-        client,
-        task_queue=task_queue,
-        workflows=[ApprovalWorkflow],
-        activities=[notify_approver],
-        plugins=[plugin],
-    ):
-        handle: WorkflowHandle[ApprovalWorkflow, dict] = await client.start_workflow(
-            ApprovalWorkflow.run,
-            ApprovalRequest(request_type="supplies", amount=25.0),
-            id=f"approval-{uuid.uuid4()}",
-            task_queue=task_queue,
-        )
-
-        # Wait for approval state
-        import asyncio
-
-        for _ in range(20):
-            status = await handle.query(ApprovalWorkflow.get_status)
-            if status == "waiting_for_approval":
-                break
-            await asyncio.sleep(0.1)
-
-        # Verify low risk level
-        pending = await handle.query(ApprovalWorkflow.get_pending_approval)
-        assert pending is not None
-        assert pending["risk_level"] == "low"
-
-        # Approve
-        await handle.signal(
-            ApprovalWorkflow.provide_approval,
-            {"approved": True, "reason": "Auto-approved", "approver": "system"},
-        )
-
-        result = await handle.result()
-        assert result["approved"] is True

tests/langgraph_samples/approval_wait_condition_test.py

@@ -0,0 +1,117 @@
+"""Tests for the approval_wait_condition LangGraph sample."""
+
+import uuid
+
+from temporalio.client import Client, WorkflowHandle
+from temporalio.contrib.langgraph import LangGraphPlugin
+from temporalio.worker import Worker
+
+from langgraph_samples.human_in_the_loop.approval_wait_condition.graph import (
+    build_approval_graph,
+    notify_approver,
+)
+from langgraph_samples.human_in_the_loop.approval_wait_condition.workflow import (
+    ApprovalRequest,
+    ApprovalWorkflow,
+)
+
+
+async def test_approval_wait_condition_approved(client: Client) -> None:
+    """Test approval workflow when request is approved."""
+    task_queue = f"approval-test-{uuid.uuid4()}"
+
+    plugin = LangGraphPlugin(graphs={"approval_workflow": build_approval_graph})
+
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[ApprovalWorkflow],
+        activities=[notify_approver],
+        plugins=[plugin],
+    ):
+        # Start the workflow
+        handle: WorkflowHandle[ApprovalWorkflow, dict] = await client.start_workflow(
+            ApprovalWorkflow.run,
+            ApprovalRequest(request_type="expense", amount=500.0),
+            id=f"approval-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+
+        # Wait for the workflow to reach the approval point
+        import asyncio
+
+        for _ in range(20):
+            status = await handle.query(ApprovalWorkflow.get_status)
+            if status == "waiting_for_approval":
+                break
+            await asyncio.sleep(0.1)
+
+        assert status == "waiting_for_approval"
+
+        # Query the pending approval
+        pending = await handle.query(ApprovalWorkflow.get_pending_approval)
+        assert pending is not None
+        assert pending["amount"] == 500.0
+        assert pending["risk_level"] == "medium"
+
+        # Send approval signal
+        await handle.signal(
+            ApprovalWorkflow.provide_approval,
+            {"approved": True, "reason": "Looks good", "approver": "manager"},
+        )
+
+        # Wait for result
+        result = await handle.result()
+
+        assert result["approved"] is True
+        assert result["executed"] is True
+        assert "Successfully processed" in result["result"]
+        assert "manager" in result["result"]
+
+
+async def test_approval_wait_condition_rejected(client: Client) -> None:
+    """Test approval workflow when request is rejected."""
+    task_queue = f"approval-test-{uuid.uuid4()}"
+
+    plugin = LangGraphPlugin(graphs={"approval_workflow": build_approval_graph})
+
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[ApprovalWorkflow],
+        activities=[notify_approver],
+        plugins=[plugin],
+    ):
+        handle: WorkflowHandle[ApprovalWorkflow, dict] = await client.start_workflow(
+            ApprovalWorkflow.run,
+            ApprovalRequest(request_type="purchase", amount=5000.0),
+            id=f"approval-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+
+        # Wait for approval state
+        import asyncio
+
+        for _ in range(20):
+            status = await handle.query(ApprovalWorkflow.get_status)
+            if status == "waiting_for_approval":
+                break
+            await asyncio.sleep(0.1)
+
+        # Verify high risk level for large amount
+        pending = await handle.query(ApprovalWorkflow.get_pending_approval)
+        assert pending is not None
+        assert pending["risk_level"] == "high"
+
+        # Reject the request
+        await handle.signal(
+            ApprovalWorkflow.provide_approval,
+            {"approved": False, "reason": "Budget exceeded", "approver": "cfo"},
+        )
+
+        result = await handle.result()
+
+        assert result["approved"] is False
+        assert result["executed"] is False
+        assert "rejected" in result["result"]
+        assert "cfo" in result["result"]