More tests!

Author: donald-pinckney

tests/openai_agents/basic/test_agent_lifecycle_workflow.py

@@ -0,0 +1,35 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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),
+        # No external activities needed - workflow uses function tools
+    ):
+        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 hasattr(result, "number")
+        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,55 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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),
+        # No external activities needed
+    ):
+        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),
+        # No external activities needed
+    ):
+        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_lifecycle_workflow.py

@@ -0,0 +1,33 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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),
+        # No external activities needed - workflow uses function tools
+    ):
+        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 hasattr(result, "number")
+        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,54 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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,46 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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)
+
+        # Should have either strict_result or strict_error
+        assert "strict_result" in result or "strict_error" in result
+
+        # Should have either non_strict_result or non_strict_error
+        assert "non_strict_result" in result or "non_strict_error" 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
+
+        # If there's a non_strict_result, verify it's valid
+        if "non_strict_result" in result:
+            assert result["non_strict_result"] is not None

tests/openai_agents/basic/test_previous_response_id_workflow.py

@@ -0,0 +1,44 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+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),
+        # No external activities needed
+    ):
+        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

tests/openai_agents/basic/test_remote_image_workflow.py

@@ -0,0 +1,60 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.workflows.remote_image_workflow import RemoteImageWorkflow
+
+# TODO(@donald-pinckney): debug this test
+# 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=[RemoteImageWorkflow],
+#         activity_executor=ThreadPoolExecutor(5),
+#         # No external activities needed - uses remote URL directly
+#     ):
+#         # Using a reliable test image URL
+#         test_image_url = "https://httpbin.org/image/jpeg"
+
+#         result = await client.execute_workflow(
+#             RemoteImageWorkflow.run,
+#             test_image_url,
+#             id=str(uuid.uuid4()),
+#             task_queue=task_queue_name,
+#         )
+
+#         # Verify the result is a string response
+#         assert isinstance(result, str)
+#         assert len(result) > 0
+
+
+# TODO(@donald-pinckney): debug this test
+# 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=[RemoteImageWorkflow],
+#         activity_executor=ThreadPoolExecutor(5),
+#         # No external activities needed - uses remote URL directly
+#     ):
+#         # Using a reliable test image URL
+#         test_image_url = "https://httpbin.org/image/png"
+#         custom_question = "What do you see in this image?"
+
+#         result = await client.execute_workflow(
+#             RemoteImageWorkflow.run,
+#             args=[test_image_url, 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_tools_workflow.py

@@ -0,0 +1,31 @@
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from openai_agents.basic.activities.get_weather_activity import get_weather
+from openai_agents.basic.workflows.tools_workflow import ToolsWorkflow
+
+
+async def test_execute_workflow(client: Client):
+    task_queue_name = str(uuid.uuid4())
+
+    async with Worker(
+        client,
+        task_queue=task_queue_name,
+        workflows=[ToolsWorkflow],
+        activity_executor=ThreadPoolExecutor(5),
+        activities=[get_weather],
+    ):
+        result = await client.execute_workflow(
+            ToolsWorkflow.run,
+            "What's the weather like in New York?",
+            id=str(uuid.uuid4()),
+            task_queue=task_queue_name,
+        )
+
+        # Verify the result is a string response
+        assert isinstance(result, str)
+        assert len(result) > 0