Compare commits

...

2 Commits

Author SHA1 Message Date
Devin AI
8f848ac5c2 Fix lint and MRO issues in Pydantic compatibility tests
- Remove unused pytest import to fix lint error
- Change inheritance order to FlowTrackable, BaseModel to fix MRO error
- Follows same pattern as LiteAgent and Crew classes in codebase

Co-Authored-By: João <joao@crewai.com>
2025-06-16 14:44:06 +00:00
Devin AI
0ca45f6bc5 Fix Pydantic compatibility issue in FlowTrackable model validator
- Remove max_depth parameter from _set_parent_flow method signature
- Convert max_depth to local variable to maintain functionality
- Ensures compatibility with Pydantic versions < 2.8
- Add comprehensive tests for FlowTrackable instantiation scenarios
- Fixes issue #3011 where Crew instantiation failed with ValidationInfo errors

The model validator signature now follows the same pattern as other
validators in the codebase (Agent, LiteAgent, BaseAgent) which only
accept 'self' parameter.

Co-Authored-By: João <joao@crewai.com>
2025-06-16 14:40:25 +00:00
2 changed files with 135 additions and 1 deletions

View File

@@ -20,7 +20,8 @@ class FlowTrackable(BaseModel):
)
@model_validator(mode="after")
def _set_parent_flow(self, max_depth: int = 5) -> "FlowTrackable":
def _set_parent_flow(self) -> "FlowTrackable":
max_depth = 5
frame = inspect.currentframe()
try:

View File

@@ -0,0 +1,133 @@
"""Tests for Pydantic version compatibility issues."""
from unittest.mock import patch, MagicMock
from pydantic import BaseModel
from crewai.flow.flow_trackable import FlowTrackable
from crewai.flow import Flow
class TestFlowTrackable(FlowTrackable, BaseModel):
"""Test class that inherits from FlowTrackable for testing."""
name: str = "test"
class MockFlow(Flow):
"""Mock Flow class for testing."""
def __init__(self):
super().__init__()
def test_flow_trackable_instantiation():
"""Test that FlowTrackable can be instantiated without ValidationInfo errors."""
trackable = TestFlowTrackable()
assert trackable.name == "test"
assert trackable.parent_flow is None
def test_flow_trackable_with_parent_flow():
"""Test that FlowTrackable correctly identifies parent flow from call stack."""
mock_flow = MockFlow()
def create_trackable_in_flow():
return TestFlowTrackable()
with patch('inspect.currentframe') as mock_frame:
mock_current_frame = MagicMock()
mock_parent_frame = MagicMock()
mock_flow_frame = MagicMock()
mock_current_frame.f_back = mock_parent_frame
mock_parent_frame.f_back = mock_flow_frame
mock_flow_frame.f_back = None
mock_parent_frame.f_locals = {}
mock_flow_frame.f_locals = {"self": mock_flow}
mock_frame.return_value = mock_current_frame
trackable = create_trackable_in_flow()
assert trackable.parent_flow == mock_flow
def test_flow_trackable_no_parent_flow():
"""Test that FlowTrackable handles case where no parent flow is found."""
with patch('inspect.currentframe') as mock_frame:
mock_current_frame = MagicMock()
mock_parent_frame = MagicMock()
mock_current_frame.f_back = mock_parent_frame
mock_parent_frame.f_back = None
mock_parent_frame.f_locals = {"self": "not_a_flow"}
mock_frame.return_value = mock_current_frame
trackable = TestFlowTrackable()
assert trackable.parent_flow is None
def test_flow_trackable_max_depth_limit():
"""Test that FlowTrackable respects max_depth limit when searching for parent flow."""
with patch('inspect.currentframe') as mock_frame:
mock_frames = []
for i in range(10):
frame = MagicMock()
frame.f_locals = {"self": f"frame_{i}"}
mock_frames.append(frame)
for i in range(len(mock_frames) - 1):
mock_frames[i].f_back = mock_frames[i + 1]
mock_frames[-1].f_back = None
mock_frame.return_value = mock_frames[0]
trackable = TestFlowTrackable()
assert trackable.parent_flow is None
def test_flow_trackable_none_frame():
"""Test that FlowTrackable handles None frame gracefully."""
with patch('inspect.currentframe', return_value=None):
trackable = TestFlowTrackable()
assert trackable.parent_flow is None
def test_pydantic_model_validator_signature():
"""Test that the model validator has the correct signature for Pydantic compatibility."""
import inspect
from crewai.flow.flow_trackable import FlowTrackable
validator_method = FlowTrackable._set_parent_flow
sig = inspect.signature(validator_method)
params = list(sig.parameters.keys())
assert params == ['self'], f"Expected ['self'], got {params}"
assert sig.return_annotation == "FlowTrackable"
def test_crew_instantiation_with_flow_trackable():
"""Test that Crew can be instantiated without ValidationInfo errors (reproduces issue #3011)."""
from crewai import Crew, Agent, Task
agent = Agent(
role="Test Agent",
goal="Test goal",
backstory="Test backstory"
)
task = Task(
description="Test task",
expected_output="Test output",
agent=agent
)
crew = Crew(
agents=[agent],
tasks=[task]
)
assert crew is not None
assert len(crew.agents) == 1
assert len(crew.tasks) == 1