You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
153 lines
4.3 KiB
153 lines
4.3 KiB
# Import necessary modules and functions for testing
|
|
import functools
|
|
import subprocess
|
|
import sys
|
|
import traceback
|
|
|
|
import pytest
|
|
|
|
# Try importing phoenix and handle exceptions
|
|
try:
|
|
import phoenix as px
|
|
except Exception as error:
|
|
print(f"Error importing phoenix: {error}")
|
|
print("Please install phoenix: pip install phoenix")
|
|
subprocess.run(
|
|
[sys.executable, "-m", "pip", "install", "arize-mlflow"]
|
|
)
|
|
|
|
# Import the code to be tested
|
|
from swarms.utils.phoenix_handler import phoenix_trace_decorator
|
|
|
|
|
|
# Define a fixture for Phoenix session
|
|
@pytest.fixture(scope="function")
|
|
def phoenix_session():
|
|
session = px.active_session() or px.launch_app()
|
|
yield session
|
|
session.stop()
|
|
|
|
|
|
# Define test cases for the phoenix_trace_decorator function
|
|
def test_phoenix_trace_decorator_documentation():
|
|
"""Test if phoenix_trace_decorator has a docstring."""
|
|
assert phoenix_trace_decorator.__doc__ is not None
|
|
|
|
|
|
def test_phoenix_trace_decorator_functionality(
|
|
capsys, phoenix_session
|
|
):
|
|
"""Test the functionality of phoenix_trace_decorator."""
|
|
|
|
# Define a function to be decorated
|
|
@phoenix_trace_decorator("This is a test function.")
|
|
def test_function():
|
|
print("Hello, Phoenix!")
|
|
|
|
# Execute the decorated function
|
|
test_function()
|
|
|
|
# Capture the printed output
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "Hello, Phoenix!\n"
|
|
|
|
|
|
def test_phoenix_trace_decorator_exception_handling(phoenix_session):
|
|
"""Test if phoenix_trace_decorator handles exceptions correctly."""
|
|
|
|
# Define a function that raises an exception
|
|
@phoenix_trace_decorator("This function raises an exception.")
|
|
def exception_function():
|
|
raise ValueError("An error occurred.")
|
|
|
|
# Execute the decorated function
|
|
with pytest.raises(ValueError):
|
|
exception_function()
|
|
|
|
# Check if the exception was traced by Phoenix
|
|
traces = phoenix_session.get_traces()
|
|
assert len(traces) == 1
|
|
assert traces[0].get("error") is not None
|
|
assert traces[0].get("error_info") is not None
|
|
|
|
|
|
# Define test cases for phoenix_trace_decorator
|
|
def test_phoenix_trace_decorator_docstring():
|
|
"""Test if phoenix_trace_decorator's inner function has a docstring."""
|
|
|
|
@phoenix_trace_decorator("This is a test function.")
|
|
def test_function():
|
|
"""Test function docstring."""
|
|
pass
|
|
|
|
assert test_function.__doc__ is not None
|
|
|
|
|
|
def test_phoenix_trace_decorator_functionality_with_params(
|
|
capsys, phoenix_session
|
|
):
|
|
"""Test the functionality of phoenix_trace_decorator with parameters."""
|
|
|
|
# Define a function with parameters to be decorated
|
|
@phoenix_trace_decorator("This function takes parameters.")
|
|
def param_function(a, b):
|
|
result = a + b
|
|
print(f"Result: {result}")
|
|
|
|
# Execute the decorated function with parameters
|
|
param_function(2, 3)
|
|
|
|
# Capture the printed output
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "Result: 5\n"
|
|
|
|
|
|
def test_phoenix_trace_decorator_nested_calls(
|
|
capsys, phoenix_session
|
|
):
|
|
"""Test nested calls of phoenix_trace_decorator."""
|
|
|
|
# Define a nested function with decorators
|
|
@phoenix_trace_decorator("Outer function")
|
|
def outer_function():
|
|
print("Outer function")
|
|
|
|
@phoenix_trace_decorator("Inner function")
|
|
def inner_function():
|
|
print("Inner function")
|
|
|
|
inner_function()
|
|
|
|
# Execute the decorated functions
|
|
outer_function()
|
|
|
|
# Capture the printed output
|
|
captured = capsys.readouterr()
|
|
assert "Outer function" in captured.out
|
|
assert "Inner function" in captured.out
|
|
|
|
|
|
def test_phoenix_trace_decorator_nested_exception_handling(
|
|
phoenix_session,
|
|
):
|
|
"""Test exception handling with nested phoenix_trace_decorators."""
|
|
|
|
# Define a function with nested decorators and an exception
|
|
@phoenix_trace_decorator("Outer function")
|
|
def outer_function():
|
|
@phoenix_trace_decorator("Inner function")
|
|
def inner_function():
|
|
raise ValueError("Inner error")
|
|
|
|
inner_function()
|
|
|
|
# Execute the decorated functions
|
|
with pytest.raises(ValueError):
|
|
outer_function()
|
|
|
|
# Check if both exceptions were traced by Phoenix
|
|
traces = phoenix_session.get_traces()
|
|
assert len(traces) == 2
|
|
assert "Outer function" in traces[0].get("error_info")
|
|
assert "Inner function" in traces[1].get("error_info")
|