fix: Add pytest.ini and improve test initialization

Co-Authored-By: Joe Moura <joao@crewai.com>

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    agentops: Tests for AgentOps integration

src/crewai/crew.py

@@ -53,10 +53,15 @@ from crewai.utilities.planning_handler import CrewPlanner
 from crewai.utilities.task_output_storage_handler import TaskOutputStorageHandler
 from crewai.utilities.training_handler import CrewTrainingHandler
 
+from typing import Optional
+
 try:
     import agentops  # type: ignore
+    from agentops.exceptions import AgentOpsError, AuthenticationError  # type: ignore
 except ImportError:
     agentops = None
+    AgentOpsError = None
+    AuthenticationError = None
 
 
 warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
@@ -91,6 +96,8 @@ class Crew(BaseModel):
     __hash__ = object.__hash__  # type: ignore
     _execution_span: Any = PrivateAttr()
     _rpm_controller: RPMController = PrivateAttr()
+    _agentops: Optional['agentops.AgentOps'] = PrivateAttr(default=None)
+    _telemetry: Optional[Telemetry] = PrivateAttr(default=None)
     _logger: Logger = PrivateAttr()
     _file_handler: FileHandler = PrivateAttr()
     _cache_handler: InstanceOf[CacheHandler] = PrivateAttr(default=CacheHandler())
@@ -241,38 +248,72 @@ class Crew(BaseModel):
         # TODO: Improve typing
         return json.loads(v) if isinstance(v, Json) else v  # type: ignore
 
-    @model_validator(mode="after")
+    def _validate_api_key(self, api_key: Optional[str]) -> bool:
+        """Validate the AgentOps API key.
+        
+        Args:
+            api_key: The API key to validate
+            
+        Returns:
+            bool: True if the API key is valid, False otherwise
+        """
+        if not api_key:
+            return False
+        stripped_key = api_key.strip()
+        return bool(stripped_key and len(stripped_key) > 10)
+
     def set_private_attrs(self) -> "Crew":
-        """Set private attributes."""
+        """Initialize private attributes including AgentOps integration.
+        
+        This method sets up:
+        - Logger and file handler for output logging
+        - RPM controller for rate limiting
+        - AgentOps integration for monitoring (if available and configured)
+        """
         self._cache_handler = CacheHandler()
         self._logger = Logger(verbose=self.verbose)
         if self.output_log_file:
             self._file_handler = FileHandler(self.output_log_file)
         self._rpm_controller = RPMController(max_rpm=self.max_rpm, logger=self._logger)
+        self._telemetry = Telemetry()
+        self._telemetry.set_tracer()
 
         # Initialize agentops if available and API key is present
         if agentops:
             api_key = os.getenv("AGENTOPS_API_KEY")
-            if api_key and api_key.strip():  # Validate API key
+            if self._validate_api_key(api_key):
                 try:
                     agentops.init(api_key)
+                    self._agentops = agentops
                     self._logger.log(
                         "info",
                         "Successfully initialized agentops",
                         color="green"
                     )
-                except (ConnectionError, ValueError) as e:
+                except (ConnectionError, AuthenticationError) as e:
                     self._logger.log(
                         "warning",
-                        f"Failed to initialize agentops: {e}",
+                        f"Failed to connect to agentops: {e}",
                         color="yellow"
                     )
+                    self._agentops = None
+                except (ValueError, AgentOpsError) as e:
+                    self._logger.log(
+                        "warning",
+                        f"Invalid agentops configuration: {e}",
+                        color="yellow"
+                    )
+                    self._agentops = None
+            else:
+                self._logger.log(
+                    "warning",
+                    "Invalid AGENTOPS_API_KEY provided",
+                    color="yellow"
+                )
+                self._agentops = None
 
         if self.function_calling_llm and not isinstance(self.function_calling_llm, LLM):
             self.function_calling_llm = create_llm(self.function_calling_llm)
-
-        self._telemetry = Telemetry()
-        self._telemetry.set_tracer()
         return self
 
     @model_validator(mode="after")

tests/crew_test.py

@@ -30,87 +30,98 @@ from crewai.utilities.rpm_controller import RPMController
 from crewai.utilities.task_output_storage_handler import TaskOutputStorageHandler
 
 
-def test_agentops_initialization_with_api_key(monkeypatch):
-    """Test that agentops is properly initialized when API key is present."""
-    import agentops
-    
-    # Mock agentops.init
-    mock_init = MagicMock()
-    monkeypatch.setattr(agentops, "init", mock_init)
-    
-    # Set API key
-    monkeypatch.setenv("AGENTOPS_API_KEY", "test-key")
-    
-    # Create crew
-    task = Task(
-        description="Test task",
-        expected_output="Test output",
-        agent=researcher,
+@pytest.fixture
+def researcher():
+    """Fixture to create a researcher agent."""
+    return Agent(
+        role="Researcher",
+        goal="Make the best research and analysis on content about AI and AI agents",
+        backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups.",
+        allow_delegation=False,
     )
-    crew = Crew(agents=[researcher], tasks=[task])
-    
-    # Verify agentops was initialized
-    mock_init.assert_called_once_with("test-key")
 
-def test_agentops_initialization_without_api_key(monkeypatch):
-    """Test that agentops is not initialized when API key is not present."""
-    import agentops
-    
-    # Mock agentops.init
-    mock_init = MagicMock()
-    monkeypatch.setattr(agentops, "init", mock_init)
-    
-    # Create crew without setting API key
-    task = Task(
-        description="Test task",
-        expected_output="Test output",
-        agent=researcher,
-    )
-    crew = Crew(agents=[researcher], tasks=[task])
-    
-    # Verify agentops.init was not called
-    mock_init.assert_not_called()
-
-def test_gemini_llm_with_agentops(monkeypatch):
-    """Test that Gemini LLM works correctly with agentops."""
-    # Mock agentops
+@pytest.fixture
+def mock_agentops():
+    """Fixture to mock agentops for testing."""
     mock_agentops = MagicMock()
-    monkeypatch.setattr("crewai.crew.agentops", mock_agentops)
-    
-    # Set API keys
-    monkeypatch.setenv("AGENTOPS_API_KEY", "test-key")
-    monkeypatch.setenv("GOOGLE_API_KEY", "test-key")
-    
-    # Create crew with Gemini LLM
-    llm = LLM(model="gemini-pro")
-    agent = Agent(
-        role="test",
-        goal="test",
-        backstory="test",
-        llm=llm
-    )
-    task = Task(
-        description="test task",
-        expected_output="test output",
-        agent=agent
-    )
-    crew = Crew(agents=[agent], tasks=[task])
-    
-    # Mock the agent execution to avoid actual API calls
-    with patch.object(Task, 'execute_sync', return_value=TaskOutput(
-        description="test",
-        raw="test output",
-        agent=agent.role
-    )):
-        # Run crew
-        crew.kickoff()
-    
-    # Verify agentops.end_session was called correctly
-    mock_agentops.end_session.assert_called_once_with(
-        end_state="Success",
-        end_state_reason="Finished Execution",
-        is_auto_end=True
-    )
+    mock_agentops.init = MagicMock()
+    return mock_agentops
+
+@pytest.mark.agentops
+class TestAgentOpsIntegration:
+    """Tests for AgentOps integration."""
+
+    def test_initialization_with_api_key(self, mock_agentops, monkeypatch):
+        """Test that agentops is properly initialized when API key is present."""
+        monkeypatch.setattr("crewai.crew.agentops", mock_agentops)
+        monkeypatch.setenv("AGENTOPS_API_KEY", "test-key-12345")
+        crew = Crew(agents=[researcher], tasks=[Task(
+            description="Test task",
+            expected_output="Test output",
+            agent=researcher,
+        )])
+        crew.set_private_attrs()
+        mock_agentops.init.assert_called_once_with("test-key-12345")
+
+    def test_initialization_without_api_key(self, mock_agentops):
+        """Test that agentops is not initialized when API key is not present."""
+        crew = Crew(agents=[researcher], tasks=[Task(
+            description="Test task",
+            expected_output="Test output",
+            agent=researcher,
+        )])
+        mock_agentops.assert_not_called()
+
+    def test_initialization_with_invalid_api_key(self, mock_agentops, monkeypatch):
+        """Test that agentops is not initialized when API key is invalid."""
+        monkeypatch.setenv("AGENTOPS_API_KEY", "   ")
+        crew = Crew(agents=[researcher], tasks=[Task(
+            description="Test task",
+            expected_output="Test output",
+            agent=researcher,
+        )])
+        mock_agentops.assert_not_called()
+
+    def test_gemini_llm_integration(self, mock_agentops, monkeypatch):
+        """Test that Gemini LLM works correctly with agentops."""
+        # Mock agentops
+        monkeypatch.setattr("crewai.crew.agentops", mock_agentops)
+        
+        # Set API keys
+        monkeypatch.setenv("AGENTOPS_API_KEY", "test-key-12345")
+        monkeypatch.setenv("GOOGLE_API_KEY", "test-key")
+        
+        # Create crew with Gemini LLM
+        llm = LLM(model="gemini-pro")
+        agent = Agent(
+            role="test",
+            goal="test",
+            backstory="test",
+            llm=llm
+        )
+        task = Task(
+            description="test task",
+            expected_output="test output",
+            agent=agent
+        )
+        crew = Crew(agents=[agent], tasks=[task])
+        crew.set_private_attrs()
+        
+        # Mock the agent execution to avoid actual API calls
+        with patch.object(Task, 'execute_sync', return_value=TaskOutput(
+            description="test",
+            raw="test output",
+            agent=agent.role
+        )):
+            # Run crew
+            crew.kickoff()
+        
+        # Verify agentops.end_session was called correctly
+        mock_agentops.end_session.assert_called_once_with(
+            end_state="Success",
+            end_state_reason="Finished Execution",
+            is_auto_end=True
+        )
 
 ceo = Agent(
     role="CEO",