Compare commits

..

1 Commits

Author SHA1 Message Date
Devin AI
1579c44373 Fix: Add conditional import for Self type to support Python 3.10
Co-Authored-By: Joe Moura <joao@crewai.com>
2025-04-09 05:49:26 +00:00
5 changed files with 21 additions and 141 deletions

View File

@@ -30,6 +30,7 @@ dependencies = [
"pdfplumber>=0.11.4",
"openpyxl>=3.1.5",
"blinker>=1.9.0",
"typing-extensions>=4.5.0",
]
[project.urls]

View File

@@ -1,34 +1,12 @@
import subprocess
from pathlib import Path
from typing import List
import click
def _check_pyproject_exists() -> bool:
"""
Check if pyproject.toml exists in the current directory.
Returns:
bool: True if pyproject.toml exists, False otherwise.
"""
if not Path("pyproject.toml").exists():
click.echo("Error: No pyproject.toml found in current directory.", err=True)
click.echo("This command must be run from the root of a crew project.", err=True)
return False
return True
def install_crew(proxy_options: List[str]) -> None:
def install_crew(proxy_options: list[str]) -> None:
"""
Install the crew by running the UV command to lock and install.
Args:
proxy_options: List of proxy options to pass to UV.
"""
if not _check_pyproject_exists():
return
try:
command = ["uv", "sync"] + proxy_options
subprocess.run(command, check=True, capture_output=False, text=True)

View File

@@ -1,5 +1,10 @@
from typing import Any, Dict, List, Optional
try:
from typing import Self
except ImportError:
from typing_extensions import Self
from crewai.memory.storage.rag_storage import RAGStorage

View File

@@ -1,118 +0,0 @@
import subprocess
from pathlib import Path
from typing import List
from unittest import mock
import pytest
from click.testing import CliRunner
from crewai.cli.install_crew import _check_pyproject_exists, install_crew
TEST_CONSTANTS = {
"ERROR_NO_PYPROJECT": "Error: No pyproject.toml found in current directory.",
"ERROR_MUST_RUN_FROM_ROOT": "This command must be run from the root of a crew project.",
"ERROR_RUNNING_CREW": "An error occurred while running the crew",
"ERROR_UNEXPECTED": "An unexpected error occurred: Test exception",
}
@pytest.fixture
def mock_subprocess_run() -> mock.MagicMock:
"""Mock subprocess.run for testing."""
with mock.patch("subprocess.run") as mock_run:
yield mock_run
@pytest.fixture
def mock_path_exists() -> mock.MagicMock:
"""Mock Path.exists for testing."""
with mock.patch("pathlib.Path.exists") as mock_exists:
yield mock_exists
@pytest.mark.parametrize(
"proxy_options,expected_command",
[
([], ["uv", "sync"]),
(["--proxy", "http://proxy.com"], ["uv", "sync", "--proxy", "http://proxy.com"]),
],
)
def test_install_crew_pyproject_exists(
mock_subprocess_run: mock.MagicMock,
mock_path_exists: mock.MagicMock,
proxy_options: List[str],
expected_command: List[str],
) -> None:
"""Test install_crew when pyproject.toml exists."""
mock_path_exists.return_value = True
install_crew(proxy_options)
mock_subprocess_run.assert_called_once_with(
expected_command, check=True, capture_output=False, text=True
)
def test_install_crew_pyproject_not_exists(
mock_subprocess_run: mock.MagicMock,
mock_path_exists: mock.MagicMock,
capsys: pytest.CaptureFixture,
) -> None:
"""Test install_crew when pyproject.toml does not exist."""
mock_path_exists.return_value = False
proxy_options: List[str] = []
install_crew(proxy_options)
mock_subprocess_run.assert_not_called()
captured = capsys.readouterr()
assert TEST_CONSTANTS["ERROR_NO_PYPROJECT"] in captured.err
assert TEST_CONSTANTS["ERROR_MUST_RUN_FROM_ROOT"] in captured.err
def test_install_crew_subprocess_error(
mock_subprocess_run: mock.MagicMock,
mock_path_exists: mock.MagicMock,
capsys: pytest.CaptureFixture,
) -> None:
"""Test install_crew when subprocess raises CalledProcessError."""
mock_path_exists.return_value = True
mock_subprocess_run.side_effect = subprocess.CalledProcessError(
cmd=["uv", "sync"], returncode=1
)
proxy_options: List[str] = []
install_crew(proxy_options)
captured = capsys.readouterr()
assert TEST_CONSTANTS["ERROR_RUNNING_CREW"] in captured.err
def test_install_crew_general_exception(
mock_subprocess_run: mock.MagicMock,
mock_path_exists: mock.MagicMock,
capsys: pytest.CaptureFixture,
) -> None:
"""Test install_crew when a general exception occurs."""
mock_path_exists.return_value = True
mock_subprocess_run.side_effect = Exception("Test exception")
proxy_options: List[str] = []
install_crew(proxy_options)
captured = capsys.readouterr()
assert TEST_CONSTANTS["ERROR_UNEXPECTED"] in captured.err
def test_check_pyproject_exists(
mock_path_exists: mock.MagicMock, capsys: pytest.CaptureFixture
) -> None:
"""Test _check_pyproject_exists function."""
mock_path_exists.return_value = True
assert _check_pyproject_exists() is True
mock_path_exists.return_value = False
assert _check_pyproject_exists() is False
captured = capsys.readouterr()
assert TEST_CONSTANTS["ERROR_NO_PYPROJECT"] in captured.err
assert TEST_CONSTANTS["ERROR_MUST_RUN_FROM_ROOT"] in captured.err

View File

@@ -0,0 +1,14 @@
def test_self_import_compatibility():
"""
Test that the Self type is properly imported and used
in a way that supports Python 3.10+.
This test will pass as long as the module imports successfully.
The actual failure would happen during import if the compatibility
fix is not working.
"""
try:
from crewai.memory.memory import Self
assert True, "Self type imported successfully"
except ImportError:
assert False, "Failed to import Self type from crewai.memory.memory"