diff --git a/lib/crewai/src/crewai/cli/version.py b/lib/crewai/src/crewai/cli/version.py index a7c1087a7..69170e16c 100644 --- a/lib/crewai/src/crewai/cli/version.py +++ b/lib/crewai/src/crewai/cli/version.py @@ -1,6 +1,107 @@ +"""Version utilities for CrewAI CLI.""" + +from collections.abc import Mapping +from datetime import datetime, timedelta +from functools import lru_cache import importlib.metadata +import json +from pathlib import Path +from typing import Any, cast +from urllib import request +from urllib.error import URLError + +import appdirs +from packaging.version import InvalidVersion, parse + + +@lru_cache(maxsize=1) +def _get_cache_file() -> Path: + """Get the path to the version cache file. + + Cached to avoid repeated filesystem operations. + """ + cache_dir = Path(appdirs.user_cache_dir("crewai")) + cache_dir.mkdir(parents=True, exist_ok=True) + return cache_dir / "version_cache.json" def get_crewai_version() -> str: - """Get the version number of CrewAI running the CLI""" + """Get the version number of CrewAI running the CLI.""" return importlib.metadata.version("crewai") + + +def _is_cache_valid(cache_data: Mapping[str, Any]) -> bool: + """Check if the cache is still valid, less than 24 hours old.""" + if "timestamp" not in cache_data: + return False + + try: + cache_time = datetime.fromisoformat(str(cache_data["timestamp"])) + return datetime.now() - cache_time < timedelta(hours=24) + except (ValueError, TypeError): + return False + + +def get_latest_version_from_pypi(timeout: int = 2) -> str | None: + """Get the latest version of CrewAI from PyPI. + + Args: + timeout: Request timeout in seconds. + + Returns: + Latest version string or None if unable to fetch. + """ + cache_file = _get_cache_file() + if cache_file.exists(): + try: + cache_data = json.loads(cache_file.read_text()) + if _is_cache_valid(cache_data): + return cast(str | None, cache_data.get("version")) + except (json.JSONDecodeError, OSError): + pass + + try: + with request.urlopen( + "https://pypi.org/pypi/crewai/json", timeout=timeout + ) as response: + data = json.loads(response.read()) + latest_version = cast(str, data["info"]["version"]) + + cache_data = { + "version": latest_version, + "timestamp": datetime.now().isoformat(), + } + cache_file.write_text(json.dumps(cache_data)) + + return latest_version + except (URLError, json.JSONDecodeError, KeyError, OSError): + return None + + +def check_version() -> tuple[str, str | None]: + """Check current and latest versions. + + Returns: + Tuple of (current_version, latest_version). + latest_version is None if unable to fetch from PyPI. + """ + current = get_crewai_version() + latest = get_latest_version_from_pypi() + return current, latest + + +def is_newer_version_available() -> tuple[bool, str, str | None]: + """Check if a newer version is available. + + Returns: + Tuple of (is_newer, current_version, latest_version). + """ + current, latest = check_version() + + if latest is None: + return False, current, None + + try: + return parse(latest) > parse(current), current, latest + except (InvalidVersion, TypeError): + return False, current, latest diff --git a/lib/crewai/src/crewai/events/utils/console_formatter.py b/lib/crewai/src/crewai/events/utils/console_formatter.py index 4aaec2cca..ac6caabcf 100644 --- a/lib/crewai/src/crewai/events/utils/console_formatter.py +++ b/lib/crewai/src/crewai/events/utils/console_formatter.py @@ -1,11 +1,14 @@ +import os import threading -from typing import Any, ClassVar +from typing import Any, ClassVar, cast from rich.console import Console from rich.live import Live from rich.panel import Panel from rich.text import Text +from crewai.cli.version import is_newer_version_available + class ConsoleFormatter: tool_usage_counts: ClassVar[dict[str, int]] = {} @@ -35,6 +38,39 @@ class ConsoleFormatter: padding=(1, 2), ) + def _show_version_update_message_if_needed(self) -> None: + """Show version update message if a newer version is available. + + Only displays when verbose mode is enabled and not running in CI/CD. + """ + if not self.verbose: + return + + if os.getenv("CI", "").lower() in ("true", "1"): + return + + try: + is_newer, current, latest = is_newer_version_available() + if is_newer and latest: + message = f"""A new version of CrewAI is available! + +Current version: {current} +Latest version: {latest} + +To update, run: uv sync --upgrade-package crewai""" + + panel = Panel( + message, + title="✨ Update Available ✨", + border_style="yellow", + padding=(1, 2), + ) + self.console.print(panel) + self.console.print() + except Exception: # noqa: S110 + # Silently ignore errors in version check - it's non-critical + pass + def _show_tracing_disabled_message_if_needed(self) -> None: """Show tracing disabled message if tracing is not enabled.""" from crewai.events.listeners.tracing.utils import ( @@ -176,9 +212,10 @@ To enable tracing, do any one of these: if not self.verbose: return - # Reset the crew completion event for this new crew execution ConsoleFormatter.crew_completion_printed.clear() + self._show_version_update_message_if_needed() + content = self.create_status_content( "Crew Execution Started", crew_name, @@ -237,6 +274,8 @@ To enable tracing, do any one of these: def handle_flow_started(self, flow_name: str, flow_id: str) -> None: """Show flow started panel.""" + self._show_version_update_message_if_needed() + content = Text() content.append("Flow Started\n", style="blue bold") content.append("Name: ", style="white") @@ -885,7 +924,7 @@ To enable tracing, do any one of these: is_a2a_delegation = False try: - output_data = json.loads(formatted_answer.output) + output_data = json.loads(cast(str, formatted_answer.output)) if isinstance(output_data, dict): if output_data.get("is_a2a") is True: is_a2a_delegation = True diff --git a/lib/crewai/tests/cli/test_version.py b/lib/crewai/tests/cli/test_version.py index 9706a282d..260064096 100644 --- a/lib/crewai/tests/cli/test_version.py +++ b/lib/crewai/tests/cli/test_version.py @@ -1,10 +1,20 @@ """Test for version management.""" +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import MagicMock, patch + from crewai import __version__ -from crewai.cli.version import get_crewai_version +from crewai.cli.version import ( + _get_cache_file, + _is_cache_valid, + get_crewai_version, + get_latest_version_from_pypi, + is_newer_version_available, +) -def test_dynamic_versioning_consistency(): +def test_dynamic_versioning_consistency() -> None: """Test that dynamic versioning provides consistent version across all access methods.""" cli_version = get_crewai_version() package_version = __version__ @@ -15,3 +25,186 @@ def test_dynamic_versioning_consistency(): # Version should not be empty assert package_version is not None assert len(package_version.strip()) > 0 + + +class TestVersionChecking: + """Test version checking utilities.""" + + def test_get_crewai_version(self) -> None: + """Test getting current crewai version.""" + version = get_crewai_version() + assert isinstance(version, str) + assert len(version) > 0 + + def test_get_cache_file(self) -> None: + """Test cache file path generation.""" + cache_file = _get_cache_file() + assert isinstance(cache_file, Path) + assert cache_file.name == "version_cache.json" + + def test_is_cache_valid_with_fresh_cache(self) -> None: + """Test cache validation with fresh cache.""" + cache_data = {"timestamp": datetime.now().isoformat(), "version": "1.0.0"} + assert _is_cache_valid(cache_data) is True + + def test_is_cache_valid_with_stale_cache(self) -> None: + """Test cache validation with stale cache.""" + old_time = datetime.now() - timedelta(hours=25) + cache_data = {"timestamp": old_time.isoformat(), "version": "1.0.0"} + assert _is_cache_valid(cache_data) is False + + def test_is_cache_valid_with_missing_timestamp(self) -> None: + """Test cache validation with missing timestamp.""" + cache_data = {"version": "1.0.0"} + assert _is_cache_valid(cache_data) is False + + @patch("crewai.cli.version.Path.exists") + @patch("crewai.cli.version.request.urlopen") + def test_get_latest_version_from_pypi_success( + self, mock_urlopen: MagicMock, mock_exists: MagicMock + ) -> None: + """Test successful PyPI version fetch.""" + # Mock cache not existing to force fetch from PyPI + mock_exists.return_value = False + + mock_response = MagicMock() + mock_response.read.return_value = b'{"info": {"version": "2.0.0"}}' + mock_urlopen.return_value.__enter__.return_value = mock_response + + version = get_latest_version_from_pypi() + assert version == "2.0.0" + + @patch("crewai.cli.version.Path.exists") + @patch("crewai.cli.version.request.urlopen") + def test_get_latest_version_from_pypi_failure( + self, mock_urlopen: MagicMock, mock_exists: MagicMock + ) -> None: + """Test PyPI version fetch failure.""" + from urllib.error import URLError + + # Mock cache not existing to force fetch from PyPI + mock_exists.return_value = False + + mock_urlopen.side_effect = URLError("Network error") + + version = get_latest_version_from_pypi() + assert version is None + + @patch("crewai.cli.version.get_crewai_version") + @patch("crewai.cli.version.get_latest_version_from_pypi") + def test_is_newer_version_available_true( + self, mock_latest: MagicMock, mock_current: MagicMock + ) -> None: + """Test when newer version is available.""" + mock_current.return_value = "1.0.0" + mock_latest.return_value = "2.0.0" + + is_newer, current, latest = is_newer_version_available() + assert is_newer is True + assert current == "1.0.0" + assert latest == "2.0.0" + + @patch("crewai.cli.version.get_crewai_version") + @patch("crewai.cli.version.get_latest_version_from_pypi") + def test_is_newer_version_available_false( + self, mock_latest: MagicMock, mock_current: MagicMock + ) -> None: + """Test when no newer version is available.""" + mock_current.return_value = "2.0.0" + mock_latest.return_value = "2.0.0" + + is_newer, current, latest = is_newer_version_available() + assert is_newer is False + assert current == "2.0.0" + assert latest == "2.0.0" + + @patch("crewai.cli.version.get_crewai_version") + @patch("crewai.cli.version.get_latest_version_from_pypi") + def test_is_newer_version_available_with_none_latest( + self, mock_latest: MagicMock, mock_current: MagicMock + ) -> None: + """Test when PyPI fetch fails.""" + mock_current.return_value = "1.0.0" + mock_latest.return_value = None + + is_newer, current, latest = is_newer_version_available() + assert is_newer is False + assert current == "1.0.0" + assert latest is None + + +class TestConsoleFormatterVersionCheck: + """Test version check display in ConsoleFormatter.""" + + @patch("crewai.events.utils.console_formatter.is_newer_version_available") + @patch.dict("os.environ", {"CI": ""}) + def test_version_message_shows_when_update_available_and_verbose( + self, mock_check: MagicMock + ) -> None: + """Test version message shows when update available and verbose enabled.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + mock_check.return_value = (True, "1.0.0", "2.0.0") + + formatter = ConsoleFormatter(verbose=True) + with patch.object(formatter.console, "print") as mock_print: + formatter._show_version_update_message_if_needed() + assert mock_print.call_count == 2 + + @patch("crewai.events.utils.console_formatter.is_newer_version_available") + def test_version_message_hides_when_verbose_false( + self, mock_check: MagicMock + ) -> None: + """Test version message hidden when verbose disabled.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + mock_check.return_value = (True, "1.0.0", "2.0.0") + + formatter = ConsoleFormatter(verbose=False) + with patch.object(formatter.console, "print") as mock_print: + formatter._show_version_update_message_if_needed() + mock_print.assert_not_called() + + @patch("crewai.events.utils.console_formatter.is_newer_version_available") + def test_version_message_hides_when_no_update_available( + self, mock_check: MagicMock + ) -> None: + """Test version message hidden when no update available.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + mock_check.return_value = (False, "2.0.0", "2.0.0") + + formatter = ConsoleFormatter(verbose=True) + with patch.object(formatter.console, "print") as mock_print: + formatter._show_version_update_message_if_needed() + mock_print.assert_not_called() + + @patch("crewai.events.utils.console_formatter.is_newer_version_available") + @patch.dict("os.environ", {"CI": "true"}) + def test_version_message_hides_in_ci_environment( + self, mock_check: MagicMock + ) -> None: + """Test version message hidden when running in CI/CD.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + mock_check.return_value = (True, "1.0.0", "2.0.0") + + formatter = ConsoleFormatter(verbose=True) + with patch.object(formatter.console, "print") as mock_print: + formatter._show_version_update_message_if_needed() + mock_print.assert_not_called() + + @patch("crewai.events.utils.console_formatter.is_newer_version_available") + @patch.dict("os.environ", {"CI": "1"}) + def test_version_message_hides_in_ci_environment_with_numeric_value( + self, mock_check: MagicMock + ) -> None: + """Test version message hidden when CI=1.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + mock_check.return_value = (True, "1.0.0", "2.0.0") + + formatter = ConsoleFormatter(verbose=True) + with patch.object(formatter.console, "print") as mock_print: + formatter._show_version_update_message_if_needed() + mock_print.assert_not_called()