WIP: failure test

Author: dandavison

tests/test_serialization_context.py

@@ -8,7 +8,7 @@
 from datetime import timedelta
 from itertools import zip_longest
 from pprint import pformat, pprint
-from typing import Any, List, Literal, Optional, Sequence, Type
+from typing import Any, List, Literal, Never, Optional, Sequence, Type
 from warnings import warn
 
 import pytest
@@ -17,7 +17,7 @@
 from temporalio import activity, workflow
 from temporalio.api.common.v1 import Payload
 from temporalio.api.failure.v1 import Failure
-from temporalio.client import Client, WorkflowUpdateFailedError
+from temporalio.client import Client, WorkflowFailureError, WorkflowUpdateFailedError
 from temporalio.common import RetryPolicy
 from temporalio.contrib.pydantic import PydanticJSONPlainPayloadConverter
 from temporalio.converter import (
@@ -42,7 +42,12 @@
 @dataclass
 class TraceItem:
     context_type: Literal["workflow", "activity"]
-    method: Literal["to_payload", "from_payload"]
+    method: Literal[
+        "to_payload",
+        "from_payload",
+        "to_failure",
+        "from_failure",
+    ]
     context: dict[str, Any]
     in_workflow: bool
     caller_location: list[str] = field(default_factory=list)
@@ -943,24 +948,23 @@ async def test_external_workflow_signal_and_cancel_payload_conversion(
         # The cancel context would only be used for failure deserialization
 
 
+# Failure conversion
+
+
 @activity.defn
-async def failing_activity() -> TraceData:
-    raise ApplicationError("test error", TraceData())
+async def failing_activity() -> Never:
+    raise ApplicationError("test error", dataclasses.asdict(TraceData()))
 
 
 @workflow.defn
 class FailureContextWorkflow:
     @workflow.run
-    async def run(self) -> TraceData:
-        try:
-            await workflow.execute_activity(
-                failing_activity,
-                start_to_close_timeout=timedelta(seconds=10),
-                retry_policy=RetryPolicy(maximum_attempts=1),
-            )
-        except ActivityError as e:
-            assert isinstance(e.cause, ApplicationError) and e.cause.details
-            return e.cause.details[0]
+    async def run(self) -> Never:
+        await workflow.execute_activity(
+            failing_activity,
+            start_to_close_timeout=timedelta(seconds=10),
+            retry_policy=RetryPolicy(maximum_attempts=1),
+        )
         raise Exception("Unreachable")
 
 
@@ -982,36 +986,80 @@ def to_failure(
         payload_converter: PayloadConverter,
         failure: Failure,
     ) -> None:
-        super().to_failure(exception, payload_converter, failure)
+        print("🌈 to_failure")
         if isinstance(exception, ApplicationError) and exception.details:
-            for detail in exception.details:
-                if isinstance(detail, TraceData) and self.context:
-                    if isinstance(self.context, ActivitySerializationContext):
-                        detail.items.append(
-                            TraceItem(
-                                context_type="activity",
-                                in_workflow=False,
-                                method="to_payload",
-                                context=dataclasses.asdict(self.context),
-                            )
+            if isinstance(
+                self.context,
+                (WorkflowSerializationContext, ActivitySerializationContext),
+            ):
+                context_type = (
+                    "workflow"
+                    if isinstance(self.context, WorkflowSerializationContext)
+                    else "activity"
+                )
+                print(
+                    f"    🌈 to_failure appending {context_type}: {exception.details}"
+                )
+                exception.details[0]["items"].append(
+                    dataclasses.asdict(
+                        TraceItem(
+                            context_type=context_type,
+                            in_workflow=workflow.in_workflow(),
+                            method="to_failure",
+                            context=dataclasses.asdict(self.context),
                         )
+                    )
+                )
+            else:
+                raise TypeError(f"self.context is {type(self.context)}")
+
+        super().to_failure(exception, payload_converter, failure)
 
     def from_failure(
         self, failure: Failure, payload_converter: PayloadConverter
     ) -> BaseException:
+        print("🌈 from_failure")
         # Let the base class create the exception
         exception = super().from_failure(failure, payload_converter)
-        # The context tracing is already in the payloads that will be decoded with context
+        print(f"    🌈 {exception.__class__}")
+        if isinstance(exception, ApplicationError) and exception.details:
+            if isinstance(
+                self.context,
+                (WorkflowSerializationContext, ActivitySerializationContext),
+            ):
+                context_type = (
+                    "workflow"
+                    if isinstance(self.context, WorkflowSerializationContext)
+                    else "activity"
+                )
+                print(
+                    f"    🌈 from_failure appending {context_type}: {exception.details}"
+                )
+                exception.details[0]["items"].append(
+                    dataclasses.asdict(
+                        TraceItem(
+                            context_type=context_type,
+                            in_workflow=workflow.in_workflow(),
+                            method="from_failure",
+                            context=dataclasses.asdict(self.context),
+                        )
+                    )
+                )
+            else:
+                raise TypeError(f"self.context is {type(self.context)}")
         return exception
 
 
 async def test_failure_conversion_with_context(client: Client):
+    print()
+    workflow_id = str(uuid.uuid4())
     task_queue = str(uuid.uuid4())
+
     test_client = Client(
         client.service_client,
         namespace=client.namespace,
-        data_converter=DataConverter(
-            payload_converter_class=SerializationContextTestPayloadConverter,
+        data_converter=dataclasses.replace(
+            DataConverter.default,
             failure_converter_class=ContextFailureConverter,
         ),
     )
@@ -1022,12 +1070,73 @@ async def test_failure_conversion_with_context(client: Client):
         activities=[failing_activity],
         workflow_runner=UnsandboxedWorkflowRunner(),
     ):
-        result = await test_client.execute_workflow(
-            FailureContextWorkflow.run,
-            id=str(uuid.uuid4()),
-            task_queue=task_queue,
-        )
-        pprint(result.items)
+        try:
+            await test_client.execute_workflow(
+                FailureContextWorkflow.run,
+                id=workflow_id,
+                task_queue=task_queue,
+            )
+        except Exception as err:
+            assert isinstance(err, WorkflowFailureError)
+            assert isinstance(err.cause, ActivityError)
+            assert isinstance(err.cause.cause, ApplicationError)
+            pprint(err.cause.cause.details)
+
+            workflow_context = dataclasses.asdict(
+                WorkflowSerializationContext(
+                    namespace="default",
+                    workflow_id=workflow_id,
+                )
+            )
+            activity_context = dataclasses.asdict(
+                ActivitySerializationContext(
+                    namespace="default",
+                    workflow_id=workflow_id,
+                    workflow_type="FailureContextWorkflow",
+                    activity_type="failing_activity",
+                    activity_task_queue=task_queue,
+                    is_local=False,
+                )
+            )
+            # 1. Exception raised in activity
+            # 2. outbound activity result to_failure(act, activity_ctx) appends and serializes
+            # 3. -> server -> WFT -> WF
+            # 4. inbound activity result from_failure(wf, activity_ctx, in_wf=False) deserializes and appends
+            # 5. outbound wf result to_failure(wf, in_wf=True) appends and serializes
+            # 6. inbound wf result from_failure(client, wf_context, in_wf=False)
+            if False:
+                assert_trace(
+                    err.cause.cause.details,
+                    [
+                        dataclasses.asdict(d)
+                        for d in [
+                            TraceItem(
+                                context_type="activity",
+                                context=activity_context,
+                                in_workflow=False,
+                                method="to_failure",  # outbound activity result
+                            ),
+                            TraceItem(
+                                context_type="activity",
+                                context=activity_context,
+                                in_workflow=False,
+                                method="from_failure",  # inbound activity result
+                            ),
+                            TraceItem(
+                                context_type="workflow",
+                                context=workflow_context,
+                                in_workflow=True,
+                                method="to_failure",  # outbound workflow result
+                            ),
+                            TraceItem(
+                                context_type="workflow",
+                                context=workflow_context,
+                                in_workflow=False,
+                                method="from_failure",  # inbound workflow result
+                            ),
+                        ]
+                    ],
+                )
 
 
 class ContextCodec(PayloadCodec, WithSerializationContext):
@@ -1193,12 +1302,16 @@ def assert_trace(trace: list[TraceItem], expected: list[TraceItem]):
     history: list[str] = []
     for item, expected_item in zip_longest(trace, expected):
         if item is None:
-            raise AssertionError("Fewer items in trace than expected")
+            raise AssertionError(
+                f"Fewer items in trace than expected.\n\n History:\n{'\n'.join(history)}"
+            )
         if expected_item is None:
-            raise AssertionError("More items in trace than expected")
+            raise AssertionError(
+                f"More items in trace than expected.\n\n History:\n{'\n'.join(history)}"
+            )
         if item != expected_item:
             raise AssertionError(
-                f"Item:\n{pformat(item)}\n\ndoes not match expected:\n\n {pformat(expected_item)}.\n\n History:\n{chr(10).join(history)}"
+                f"Item:\n{pformat(item)}\n\ndoes not match expected:\n\n {pformat(expected_item)}.\n\n History:\n{'\n'.join(history)}"
             )
         history.append(f"{item.context_type} {item.method}")