Enhance: Implement code quality improvements from review feedback

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

src/crewai/cli/install_crew.py

@@ -1,19 +1,37 @@
+import logging
 import subprocess
+from typing import List
 
 import click
 
+UV_COMMAND = "uv"
+SYNC_COMMAND = "sync"
+ACTIVE_FLAG = "--active"
 
-def install_crew(proxy_options: list[str]) -> None:
+logger = logging.getLogger(__name__)
+
+def install_crew(proxy_options: List[str]) -> None:
     """
     Install the crew by running the UV command to lock and install.
+    
+    Args:
+        proxy_options (List[str]): List of proxy-related command options.
+    
+    Note:
+        Uses --active flag to ensure proper virtual environment detection.
     """
+    if not isinstance(proxy_options, list):
+        raise ValueError("proxy_options must be a list")
+
     try:
-        command = ["uv", "sync"] + proxy_options
+        command = [UV_COMMAND, SYNC_COMMAND, ACTIVE_FLAG] + proxy_options
+        logger.debug(f"Executing command: {' '.join(command)}")
         subprocess.run(command, check=True, capture_output=False, text=True)
 
     except subprocess.CalledProcessError as e:
         click.echo(f"An error occurred while running the crew: {e}", err=True)
-        click.echo(e.output, err=True)
+        if e.output is not None:
+            click.echo(e.output, err=True)
 
     except Exception as e:
         click.echo(f"An unexpected error occurred: {e}", err=True)

src/crewai/project/crew_base.py

@@ -1,6 +1,6 @@
 import inspect
 from pathlib import Path
-from typing import Any, Callable, Dict, List, TypeVar, Union, cast
+from typing import Any, Callable, Dict, TypeVar, cast
 
 import yaml
 from dotenv import load_dotenv
@@ -116,33 +116,13 @@ def CrewBase(cls: T) -> T:
         def _map_agent_variables(
             self,
             agent_name: str,
-            agent_info: Union[Dict[str, Any], List[Dict[str, Any]]],
+            agent_info: Dict[str, Any],
             agents: Dict[str, Callable],
             llms: Dict[str, Callable],
             tool_functions: Dict[str, Callable],
             cache_handler_functions: Dict[str, Callable],
             callbacks: Dict[str, Callable],
         ) -> None:
-            """Maps agent variables from configuration to internal state.
-            
-            Args:
-                agent_name: Name of the agent.
-                agent_info: Configuration as a dictionary or list of configurations.
-                agents: Dictionary of agent functions.
-                llms: Dictionary of LLM functions.
-                tool_functions: Dictionary of tool functions.
-                cache_handler_functions: Dictionary of cache handler functions.
-                callbacks: Dictionary of callback functions.
-            
-            Raises:
-                ValueError: When an empty list is provided as agent_info.
-            """
-            # If agent_info is a list, use the first item as the configuration
-            if isinstance(agent_info, list):
-                if not agent_info:
-                    raise ValueError(f"Empty agent configuration list for agent {agent_name}")
-                agent_info = agent_info[0]
-                
             if llm := agent_info.get("llm"):
                 try:
                     self.agents_config[agent_name]["llm"] = llms[llm]()

tests/cli/install_crew_test.py

@@ -0,0 +1,79 @@
+from unittest import mock
+from typing import List, Any
+import pytest
+import subprocess
+
+from crewai.cli.install_crew import install_crew, UV_COMMAND, SYNC_COMMAND, ACTIVE_FLAG
+
+
+@pytest.mark.parametrize(
+    "proxy_options,expected_command",
+    [
+        ([], [UV_COMMAND, SYNC_COMMAND, ACTIVE_FLAG]),
+        (
+            ["--index-url", "https://custom-pypi.org/simple"],
+            [UV_COMMAND, SYNC_COMMAND, ACTIVE_FLAG, "--index-url", "https://custom-pypi.org/simple"],
+        ),
+    ],
+)
+@mock.patch("subprocess.run")
+def test_install_crew_with_options(
+    mock_subprocess: mock.MagicMock, proxy_options: List[str], expected_command: List[str]
+) -> None:
+    """Test that install_crew correctly passes options to the command."""
+    install_crew(proxy_options)
+    mock_subprocess.assert_called_once_with(
+        expected_command, check=True, capture_output=False, text=True
+    )
+
+
+@mock.patch("subprocess.run")
+@mock.patch("click.echo")
+def test_install_crew_with_subprocess_error(
+    mock_echo: mock.MagicMock, mock_subprocess: mock.MagicMock
+) -> None:
+    """Test that install_crew handles subprocess errors correctly."""
+    error = subprocess.CalledProcessError(1, f"{UV_COMMAND} {SYNC_COMMAND} {ACTIVE_FLAG}")
+    error.output = "Error output"
+    mock_subprocess.side_effect = error
+    
+    install_crew([])
+    
+    assert mock_echo.call_count == 2
+    mock_echo.assert_any_call(f"An error occurred while running the crew: {error}", err=True)
+    mock_echo.assert_any_call("Error output", err=True)
+
+
+@mock.patch("subprocess.run")
+@mock.patch("click.echo")
+def test_install_crew_with_subprocess_error_empty_output(
+    mock_echo: mock.MagicMock, mock_subprocess: mock.MagicMock
+) -> None:
+    """Test that install_crew handles subprocess errors with empty output correctly."""
+    error = subprocess.CalledProcessError(1, f"{UV_COMMAND} {SYNC_COMMAND} {ACTIVE_FLAG}")
+    error.output = None
+    mock_subprocess.side_effect = error
+    
+    install_crew([])
+    
+    mock_echo.assert_called_once_with(f"An error occurred while running the crew: {error}", err=True)
+
+
+@mock.patch("subprocess.run")
+@mock.patch("click.echo")
+def test_install_crew_with_generic_exception(
+    mock_echo: mock.MagicMock, mock_subprocess: mock.MagicMock
+) -> None:
+    """Test that install_crew handles generic exceptions correctly."""
+    error = Exception("Generic error")
+    mock_subprocess.side_effect = error
+    
+    install_crew([])
+    
+    mock_echo.assert_called_once_with(f"An unexpected error occurred: {error}", err=True)
+
+
+def test_install_crew_with_invalid_proxy_options() -> None:
+    """Test that install_crew validates the proxy_options parameter."""
+    with pytest.raises(ValueError, match="proxy_options must be a list"):
+        install_crew("not a list")  # type: ignore

tests/project/test_yaml_config.py

@@ -1,115 +0,0 @@
-import os
-import sys
-import tempfile
-from pathlib import Path
-
-import pytest
-import yaml
-
-
-class TestYamlConfig:
-    """Tests for YAML configuration handling."""
-    
-    def test_list_format_in_yaml(self):
-        """Test that list format in YAML is handled correctly."""
-        # Create a test YAML content with list format
-        yaml_content = """
-        test_agent:
-          - name: test_agent
-            role: Test Agent
-            goal: Test Goal
-        """
-        
-        # Parse the YAML content
-        data = yaml.safe_load(yaml_content)
-        
-        # Get the agent_info which should be a list
-        agent_name = "test_agent"
-        agent_info = data[agent_name]
-        
-        # Verify it's a list
-        assert isinstance(agent_info, list)
-        
-        # Create a function that simulates the behavior of _map_agent_variables
-        # with our fix applied
-        def map_agent_variables(agent_name, agent_info):
-            # This is the fix we implemented
-            if isinstance(agent_info, list):
-                if not agent_info:
-                    raise ValueError(f"Empty agent configuration list for agent {agent_name}")
-                agent_info = agent_info[0]
-                
-            # Try to access a dictionary method on agent_info
-            # This would fail with AttributeError if agent_info is still a list
-            value = agent_info.get("name")
-            return value
-        
-        # Call the function - this would raise AttributeError before the fix
-        result = map_agent_variables(agent_name, agent_info)
-        
-    def test_empty_list_in_yaml(self):
-        """Test that empty list in YAML raises appropriate error."""
-        # Create a test YAML content with empty list
-        yaml_content = """
-        test_agent: []
-        """
-        
-        # Parse the YAML content
-        data = yaml.safe_load(yaml_content)
-        
-        # Get the agent_info which should be an empty list
-        agent_name = "test_agent"
-        agent_info = data[agent_name]
-        
-        # Verify it's a list
-        assert isinstance(agent_info, list)
-        assert len(agent_info) == 0
-        
-        # Create a function that simulates the behavior of _map_agent_variables
-        def map_agent_variables(agent_name, agent_info):
-            if isinstance(agent_info, list):
-                if not agent_info:
-                    raise ValueError(f"Empty agent configuration list for agent {agent_name}")
-                agent_info = agent_info[0]
-            return agent_info
-        
-        # Call the function - should raise ValueError
-        with pytest.raises(ValueError, match=f"Empty agent configuration list for agent {agent_name}"):
-            map_agent_variables(agent_name, agent_info)
-    def test_multiple_items_in_list(self):
-        """Test that when multiple items are in the list, the first one is used."""
-        # Create a test YAML content with multiple items in the list
-        yaml_content = """
-        test_agent:
-          - name: first_agent
-            role: First Agent
-            goal: First Goal
-          - name: second_agent
-            role: Second Agent
-            goal: Second Goal
-        """
-        
-        # Parse the YAML content
-        data = yaml.safe_load(yaml_content)
-        
-        # Get the agent_info which should be a list
-        agent_name = "test_agent"
-        agent_info = data[agent_name]
-        
-        # Verify it's a list with multiple items
-        assert isinstance(agent_info, list)
-        assert len(agent_info) > 1
-        
-        # Create a function that simulates the behavior of _map_agent_variables
-        def map_agent_variables(agent_name, agent_info):
-            if isinstance(agent_info, list):
-                if not agent_info:
-                    raise ValueError(f"Empty agent configuration list for agent {agent_name}")
-                agent_info = agent_info[0]
-            return agent_info.get("name")
-        
-        # Call the function - should return name from the first item
-        result = map_agent_variables(agent_name, agent_info)
-        
-        # Verify only the first item was used
-        assert result == "first_agent"