Testing OpenAI Agents SDK Examples (Basic Examples)

README.md

@@ -90,3 +90,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 To run the tests:
 
     uv run poe test
+
+By default, this won't run the OpenAI SDK sample tests. To run those, set your `OPENAI_API_KEY` in your environment, and run:
+
+    uv run poe test tests/openai_agents

pyproject.toml

@@ -26,6 +26,7 @@ dev = [
     "types-pyyaml>=6.0.12.20241230,<7",
     "pytest-pretty>=1.3.0",
     "poethepoet>=0.36.0",
+    "pytest-mock>=3.15.1",
 ]
 bedrock = ["boto3>=1.34.92,<2"]
 dsl = [
@@ -122,7 +123,7 @@ build-backend = "hatchling.build"
 format = [{cmd = "uv run black ."}, {cmd = "uv run isort ."}]
 lint = [{cmd = "uv run black --check ."}, {cmd = "uv run isort --check-only ."}, {ref = "lint-types" }]
 lint-types = "uv run --all-groups mypy --check-untyped-defs --namespace-packages ."
-test = "uv run --all-groups pytest"
+test = "uv run --all-groups pytest --ignore=tests/openai_agents"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

tests/conftest.py

@@ -37,8 +37,15 @@ def event_loop():
     loop.close()
 
 
-@pytest_asyncio.fixture(scope="session")
-async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]:
+@pytest.fixture
+def plugins():
+    # By default, no plugins.
+    # Other tests can override this fixture, such as in tests/openai_agents/conftest.py
+    return []
+
+
+@pytest_asyncio.fixture
+async def env(request, plugins) -> AsyncGenerator[WorkflowEnvironment, None]:
     env_type = request.config.getoption("--workflow-environment")
     if env_type == "local":
         env = await WorkflowEnvironment.start_local(
@@ -47,12 +54,15 @@ async def env(request) -> AsyncGenerator[WorkflowEnvironment, None]:
                 "frontend.enableExecuteMultiOperation=true",
                 "--dynamic-config-value",
                 "system.enableEagerWorkflowStart=true",
-            ]
+            ],
+            plugins=plugins,
         )
     elif env_type == "time-skipping":
         env = await WorkflowEnvironment.start_time_skipping()
     else:
-        env = WorkflowEnvironment.from_client(await Client.connect(env_type))
+        env = WorkflowEnvironment.from_client(
+            await Client.connect(env_type, plugins=plugins)
+        )
     yield env
     await env.shutdown()
 

tests/openai_agents/basic/test_agent_lifecycle_workflow.py

@@ -0,0 +1,32 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.agent_lifecycle_workflow import (
+    AgentLifecycleWorkflow,
+)
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[AgentLifecycleWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        result = await client.execute_workflow(
+            AgentLifecycleWorkflow.run,
+            10,  # max_number parameter
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result has the expected structure
+        assert isinstance(result.number, int)
+        assert (
+            0 <= result.number <= 20
+        )  # Should be between 0 and max*2 due to multiply operation

tests/openai_agents/basic/test_dynamic_system_prompt_workflow.py

@@ -0,0 +1,52 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.dynamic_system_prompt_workflow import (
+    DynamicSystemPromptWorkflow,
+)
+
+
+async def test_execute_workflow_with_random_style(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[DynamicSystemPromptWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        result = await client.execute_workflow(
+            DynamicSystemPromptWorkflow.run,
+            "Tell me about the weather today.",
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result has the expected format
+        assert "Style:" in result
+        assert "Response:" in result
+        assert any(style in result for style in ["haiku", "pirate", "robot"])
+
+
+async def test_execute_workflow_with_specific_style(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[DynamicSystemPromptWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        result = await client.execute_workflow(
+            DynamicSystemPromptWorkflow.run,
+            args=["Tell me about the weather today.", "haiku"],
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result has the expected format and style
+        assert "Style: haiku" in result
+        assert "Response:" in result

tests/openai_agents/basic/test_hello_world_workflow.py

@@ -0,0 +1,57 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from agents import ModelResponse, Usage
+from openai.types.responses import ResponseOutputMessage, ResponseOutputText
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.hello_world_workflow import HelloWorldAgent
+
+
+@pytest.fixture
+def mocked_model(mocker):
+    mock = mocker.AsyncMock()
+    mock.get_response.side_effect = [
+        ModelResponse(
+            output=[
+                ResponseOutputMessage(
+                    id="1",
+                    content=[
+                        ResponseOutputText(
+                            annotations=[],
+                            text="This is a haiku (not really)",
+                            type="output_text",
+                        )
+                    ],
+                    role="assistant",
+                    status="completed",
+                    type="message",
+                )
+            ],
+            usage=Usage(requests=1, input_tokens=1, output_tokens=1, total_tokens=1),
+            response_id="1",
+        )
+    ]
+
+    return mock
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[HelloWorldAgent],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        result = await client.execute_workflow(
+            HelloWorldAgent.run,
+            "Write a recursive haiku about recursive haikus.",
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+        assert isinstance(result, str)
+        assert len(result) > 0

tests/openai_agents/basic/test_lifecycle_workflow.py

@@ -0,0 +1,30 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.lifecycle_workflow import LifecycleWorkflow
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[LifecycleWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        result = await client.execute_workflow(
+            LifecycleWorkflow.run,
+            10,  # max_number parameter
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result has the expected structure
+        assert isinstance(result.number, int)
+        assert (
+            0 <= result.number <= 20
+        )  # Should be between 0 and max*2 due to multiply operation

tests/openai_agents/basic/test_local_image_workflow.py

@@ -0,0 +1,53 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.activities.image_activities import read_image_as_base64
+from openai_agents.basic.workflows.local_image_workflow import LocalImageWorkflow
+
+
+async def test_execute_workflow_default_question(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[LocalImageWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+        activities=[read_image_as_base64],
+    ):
+        result = await client.execute_workflow(
+            LocalImageWorkflow.run,
+            "openai_agents/basic/media/image_bison.jpg",  # Path to test image
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result is a string response
+        assert isinstance(result, str)
+        assert len(result) > 0
+
+
+async def test_execute_workflow_custom_question(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[LocalImageWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+        activities=[read_image_as_base64],
+    ):
+        custom_question = "What animals do you see in this image?"
+        result = await client.execute_workflow(
+            LocalImageWorkflow.run,
+            args=["openai_agents/basic/media/image_bison.jpg", custom_question],
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result is a string response
+        assert isinstance(result, str)
+        assert len(result) > 0

tests/openai_agents/basic/test_non_strict_output_workflow.py

@@ -0,0 +1,42 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.non_strict_output_workflow import (
+    NonStrictOutputWorkflow,
+)
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[NonStrictOutputWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+        # No external activities needed
+    ):
+        result = await client.execute_workflow(
+            NonStrictOutputWorkflow.run,
+            "Tell me 3 funny jokes about programming.",
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result has the expected structure
+        assert isinstance(result, dict)
+
+        assert "strict_error" in result
+        assert "non_strict_result" in result
+
+        # If there's a strict_error, it should be a string
+        if "strict_error" in result:
+            assert isinstance(result["strict_error"], str)
+            assert len(result["strict_error"]) > 0
+
+        jokes = result["non_strict_result"]["jokes"]
+        assert isinstance(jokes, dict)
+        assert isinstance(jokes[list(jokes.keys())[0]], str)

tests/openai_agents/basic/test_previous_response_id_workflow.py

@@ -0,0 +1,42 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.previous_response_id_workflow import (
+    PreviousResponseIdWorkflow,
+)
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[PreviousResponseIdWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+    ):
+        first_question = "What is the capital of France?"
+        follow_up_question = "What is the population of that city?"
+
+        result = await client.execute_workflow(
+            PreviousResponseIdWorkflow.run,
+            args=[first_question, follow_up_question],
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result is a tuple with two string responses
+        assert isinstance(result, tuple)
+        assert len(result) == 2
+
+        first_response, second_response = result
+        assert isinstance(first_response, str)
+        assert isinstance(second_response, str)
+        assert len(first_response) > 0
+        assert len(second_response) > 0
+
+        # The responses should be different (not identical)
+        assert first_response != second_response