wip: test the sentry interceptor

gregbrowndev · gregbrowndev · commit f89fc54a57bc · 2025-07-20T14:54:31.000+01:00
diff --git a/sentry/activity.py b/sentry/activity.py
@@ -4,12 +4,22 @@
 
 
 @dataclass
-class ComposeGreetingInput:
-    greeting: str
-    name: str
+class WorkingActivityInput:
+    message: str
 
 
 @activity.defn
-async def compose_greeting(input: ComposeGreetingInput) -> str:
+async def working_activity(input: WorkingActivityInput) -> str:
+    activity.logger.info("Running activity with parameter %s" % input)
+    return "Success"
+
+
+@dataclass
+class BrokenActivityInput:
+    message: str
+
+
+@activity.defn
+async def broken_activity(input: BrokenActivityInput) -> str:
     activity.logger.info("Running activity with parameter %s" % input)
     raise Exception("Activity failed!")
diff --git a/sentry/starter.py b/sentry/starter.py
@@ -2,7 +2,7 @@
 
 from temporalio.client import Client
 
-from sentry.workflow import GreetingWorkflow
+from sentry.workflow import SentryExampleWorkflow, SentryExampleWorkflowInput
 
 
 async def main():
@@ -12,8 +12,8 @@ async def main():
     # Run workflow
     try:
         result = await client.execute_workflow(
-            GreetingWorkflow.run,
-            "World",
+            SentryExampleWorkflow.run,
+            SentryExampleWorkflowInput(option="broken"),
             id="sentry-workflow-id",
             task_queue="sentry-task-queue",
         )
diff --git a/sentry/worker.py b/sentry/worker.py
@@ -1,5 +1,4 @@
 import asyncio
-import logging
 import os
 
 import sentry_sdk
@@ -12,11 +11,9 @@
     SandboxRestrictions,
 )
 
-from sentry.activity import compose_greeting
+from sentry.activity import broken_activity, working_activity
 from sentry.interceptor import SentryInterceptor
-from sentry.workflow import GreetingWorkflow
-
-logger = logging.getLogger(__name__)
+from sentry.workflow import SentryExampleWorkflow
 
 interrupt_event = asyncio.Event()
 
@@ -29,27 +26,30 @@ def before_send(event: Event, hint: Hint) -> Event | None:
     return event
 
 
-async def main():
-    # Configure logging
-    logging.basicConfig(level=logging.INFO)
-
-    # Initialize the Sentry SDK
-    if sentry_dsn := os.environ.get("SENTRY_DSN"):
-        environment = os.environ.get("ENVIRONMENT")
-        sentry_sdk.init(
-            dsn=sentry_dsn,
-            environment=environment,
-            integrations=[
-                AsyncioIntegration(),
-            ],
-            attach_stacktrace=True,
-            before_send=before_send,
-        )
-        logger.info(f"Sentry SDK initialized for environment: {environment!r}")
-    else:
-        logger.warning(
+def initialise_sentry() -> None:
+    sentry_dsn = os.environ.get("SENTRY_DSN")
+    if not sentry_dsn:
+        print(
             "SENTRY_DSN environment variable is not set. Sentry will not be initialized."
         )
+        return
+
+    environment = os.environ.get("ENVIRONMENT")
+    sentry_sdk.init(
+        dsn=sentry_dsn,
+        environment=environment,
+        integrations=[
+            AsyncioIntegration(),
+        ],
+        attach_stacktrace=True,
+        before_send=before_send,
+    )
+    print(f"Sentry SDK initialized for environment: {environment!r}")
+
+
+async def main():
+    # Initialize the Sentry SDK
+    initialise_sentry()
 
     # Start client
     client = await Client.connect("localhost:7233")
@@ -58,8 +58,8 @@ async def main():
     async with Worker(
         client,
         task_queue="sentry-task-queue",
-        workflows=[GreetingWorkflow],
-        activities=[compose_greeting],
+        workflows=[SentryExampleWorkflow],
+        activities=[broken_activity, working_activity],
         interceptors=[SentryInterceptor()],  # Use SentryInterceptor for error reporting
         workflow_runner=SandboxedWorkflowRunner(
             restrictions=SandboxRestrictions.default.with_passthrough_modules(
diff --git a/sentry/workflow.py b/sentry/workflow.py
@@ -1,20 +1,38 @@
+import typing
+from dataclasses import dataclass
 from datetime import timedelta
 
 from temporalio import workflow
 from temporalio.common import RetryPolicy
 
+from sentry.activity import WorkingActivityInput, working_activity
+
 with workflow.unsafe.imports_passed_through():
-    from sentry.activity import ComposeGreetingInput, compose_greeting
+    from sentry.activity import BrokenActivityInput, broken_activity
+
+
+@dataclass
+class SentryExampleWorkflowInput:
+    option: typing.Literal["working", "broken"]
 
 
 @workflow.defn
-class GreetingWorkflow:
+class SentryExampleWorkflow:
     @workflow.run
-    async def run(self, name: str) -> str:
-        workflow.logger.info("Running workflow with parameter %s" % name)
+    async def run(self, input: SentryExampleWorkflowInput) -> str:
+        workflow.logger.info("Running workflow with parameter %r" % input)
+
+        if input.option == "working":
+            return await workflow.execute_activity(
+                working_activity,
+                WorkingActivityInput(message="Hello, Temporal!"),
+                start_to_close_timeout=timedelta(seconds=10),
+                retry_policy=RetryPolicy(maximum_attempts=1),
+            )
+
         return await workflow.execute_activity(
-            compose_greeting,
-            ComposeGreetingInput("Hello", name),
+            broken_activity,
+            BrokenActivityInput(message="Hello, Temporal!"),
             start_to_close_timeout=timedelta(seconds=10),
             retry_policy=RetryPolicy(maximum_attempts=1),
         )
diff --git a/tests/sentry/fake_sentry_transport.py b/tests/sentry/fake_sentry_transport.py
@@ -0,0 +1,17 @@
+import sentry_sdk
+import sentry_sdk.types
+
+
+class FakeSentryTransport:
+    """A fake transport that captures Sentry events in memory"""
+
+    # Note: we could extend from sentry_sdk.transport.Transport
+    # but `sentry_sdk.init` also takes a simple callable that takes
+    # an Event rather than a serialised Envelope object, so testing
+    # is easier.
+
+    def __init__(self):
+        self.events: list[sentry_sdk.types.Event] = []
+
+    def __callable__(self, event: sentry_sdk.types.Event) -> None:
+        self.events.append(event)
diff --git a/tests/sentry/test_interceptor.py b/tests/sentry/test_interceptor.py
@@ -0,0 +1,158 @@
+import unittest.mock
+from collections import abc
+
+import pytest
+import sentry_sdk
+import temporalio.activity
+import temporalio.workflow
+from sentry_sdk.integrations.asyncio import AsyncioIntegration
+from temporalio.client import Client
+from temporalio.worker import Worker
+from temporalio.worker.workflow_sandbox import (
+    SandboxedWorkflowRunner,
+    SandboxRestrictions,
+)
+
+from sentry.activity import broken_activity, working_activity
+from sentry.interceptor import SentryInterceptor
+from sentry.workflow import SentryExampleWorkflow, SentryExampleWorkflowInput
+from tests.sentry.fake_sentry_transport import FakeSentryTransport
+
+
+@pytest.fixture
+def transport() -> FakeSentryTransport:
+    """Fixture to provide a fake transport for Sentry SDK."""
+    return FakeSentryTransport()
+
+
+@pytest.fixture(autouse=True)
+def sentry_init(transport: FakeSentryTransport) -> None:
+    """Initialize Sentry for testing."""
+    sentry_sdk.init(
+        # Pass __callable__ explicitly so SDK treats it as Callable[[Event], None]
+        # it confuses it otherwise
+        transport=transport.__callable__,
+        integrations=[
+            AsyncioIntegration(),
+        ],
+    )
+
+
+@pytest.fixture
+async def worker(client: Client) -> abc.AsyncIterator[Worker]:
+    """Fixture to provide a worker for testing."""
+    async with Worker(
+        client,
+        task_queue="sentry-task-queue",
+        workflows=[SentryExampleWorkflow],
+        activities=[broken_activity, working_activity],
+        interceptors=[SentryInterceptor()],
+        workflow_runner=SandboxedWorkflowRunner(
+            restrictions=SandboxRestrictions.default.with_passthrough_modules(
+                "sentry_sdk"
+            )
+        ),
+    ) as worker:
+        yield worker
+
+
+async def test_sentry_interceptor_reports_no_errors_when_workflow_succeeds(
+    client: Client, worker: Worker, transport: FakeSentryTransport
+) -> None:
+    """Test that Sentry interceptor reports no errors when workflow succeeds."""
+    # WHEN
+    try:
+        await client.execute_workflow(
+            SentryExampleWorkflow.run,
+            SentryExampleWorkflowInput(option="working"),
+            id="sentry-workflow-id",
+            task_queue=worker.task_queue,
+        )
+    except Exception:
+        pytest.fail("Workflow should not raise an exception")
+
+    # THEN
+    assert len(transport.events) == 0, "No events should be captured"
+
+
+async def test_sentry_interceptor_captures_errors(
+    client: Client, worker: Worker, transport: FakeSentryTransport
+) -> None:
+    """Test that errors are captured with correct Sentry metadata."""
+    # WHEN
+    try:
+        await client.execute_workflow(
+            SentryExampleWorkflow.run,
+            SentryExampleWorkflowInput(option="broken"),
+            id="sentry-workflow-id",
+            task_queue=worker.task_queue,
+        )
+        pytest.fail("Workflow should raise an exception")
+    except Exception:
+        pass
+
+    # THEN
+    # there should be two events: one for the failed activity and one for the failed workflow
+    assert len(transport.events) == 2, "Two events should be captured"
+
+    # Check the first event - should be the activity exception
+    # --------------------------------------------------------
+    event = transport.events[0]
+
+    # Check exception was captured
+    assert event["exception"]["values"][0]["type"] == "Exception"
+    assert event["exception"]["values"][0]["value"] == "Activity failed!"
+
+    # Check useful metadata were were captured as tags
+    assert event["tags"] == {
+        "temporal.execution_type": "activity",
+        "module": "sentry.activity.broken_activity",
+        "temporal.workflow.type": "SentryExampleWorkflow",
+        "temporal.workflow.id": "sentry-workflow-id",
+        "temporal.activity.id": "1",
+        "temporal.activity.type": "broken_activity",
+        "temporal.activity.task_queue": "sentry-task-queue",
+        "temporal.workflow.namespace": "default",
+        "temporal.workflow.run_id": unittest.mock.ANY,
+    }
+
+    # Check activity input was captured as context
+    assert event["contexts"]["temporal.activity.input"] == {
+        "message": "Hello, Temporal!",
+    }
+
+    # Check activity info was captured as context
+    activity_info = temporalio.activity.Info(
+        **event["contexts"]["temporal.activity.info"]  # type: ignore
+    )
+    assert activity_info.activity_type == "broken_activity"
+
+    # Check the second event - should be the workflow exception
+    # ---------------------------------------------------------
+    event = transport.events[1]
+
+    # Check exception was captured
+    assert event["exception"]["values"][0]["type"] == "ApplicationError"
+    assert event["exception"]["values"][0]["value"] == "Activity failed!"
+
+    # Check useful metadata were were captured as tags
+    assert event["tags"] == {
+        "temporal.execution_type": "workflow",
+        "module": "sentry.workflow.SentryExampleWorkflow.run",
+        "temporal.workflow.type": "SentryExampleWorkflow",
+        "temporal.workflow.id": "sentry-workflow-id",
+        "temporal.workflow.task_queue": "sentry-task-queue",
+        "temporal.workflow.namespace": "default",
+        "temporal.workflow.run_id": unittest.mock.ANY,
+    }
+
+    # Check workflow input was captured as context
+    assert event["contexts"]["temporal.workflow.input"] == {
+        "option": "broken",
+    }
+
+    # Check workflow info was captured as context
+    workflow_info = temporalio.workflow.Info(
+        **event["contexts"]["temporal.workflow.info"]  # type: ignore
+    )
+    assert workflow_info.workflow_type == "SentryExampleWorkflow"
