crewAI/lib/cli/tests/test_version.py

"""Test for version management."""

import json
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch

from crewai_cli.version import get_crewai_version as _get_ver
from crewai_cli.version import (
    _find_latest_non_yanked_version,
    _get_cache_file,
    _is_cache_valid,
    _is_version_yanked,
    get_crewai_version,
    get_latest_version_from_pypi,
    is_current_version_yanked,
    is_newer_version_available,
)


def test_dynamic_versioning_consistency() -> None:
    """Test that dynamic versioning provides consistent version across all access methods."""
    cli_version = get_crewai_version()
    package_version = _get_ver()

    assert cli_version == package_version

    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 uses releases data."""
        mock_exists.return_value = False

        releases = {
            "1.0.0": [{"yanked": False}],
            "2.0.0": [{"yanked": False}],
            "2.1.0": [{"yanked": True, "yanked_reason": "bad release"}],
        }
        mock_response = MagicMock()
        mock_response.read.return_value = json.dumps(
            {"info": {"version": "2.1.0"}, "releases": releases}
        ).encode()
        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_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 TestFindLatestNonYankedVersion:
    """Test _find_latest_non_yanked_version helper."""

    def test_skips_yanked_versions(self) -> None:
        """Test that yanked versions are skipped."""
        releases = {
            "1.0.0": [{"yanked": False}],
            "2.0.0": [{"yanked": True}],
        }
        assert _find_latest_non_yanked_version(releases) == "1.0.0"

    def test_returns_highest_non_yanked(self) -> None:
        """Test that the highest non-yanked version is returned."""
        releases = {
            "1.0.0": [{"yanked": False}],
            "1.5.0": [{"yanked": False}],
            "2.0.0": [{"yanked": True}],
        }
        assert _find_latest_non_yanked_version(releases) == "1.5.0"

    def test_returns_none_when_all_yanked(self) -> None:
        """Test that None is returned when all versions are yanked."""
        releases = {
            "1.0.0": [{"yanked": True}],
            "2.0.0": [{"yanked": True}],
        }
        assert _find_latest_non_yanked_version(releases) is None

    def test_skips_prerelease_versions(self) -> None:
        """Test that pre-release versions are skipped."""
        releases = {
            "1.0.0": [{"yanked": False}],
            "2.0.0a1": [{"yanked": False}],
            "2.0.0rc1": [{"yanked": False}],
        }
        assert _find_latest_non_yanked_version(releases) == "1.0.0"

    def test_skips_versions_with_empty_files(self) -> None:
        """Test that versions with no files are skipped."""
        releases: dict[str, list[dict[str, bool]]] = {
            "1.0.0": [{"yanked": False}],
            "2.0.0": [],
        }
        assert _find_latest_non_yanked_version(releases) == "1.0.0"

    def test_handles_invalid_version_strings(self) -> None:
        """Test that invalid version strings are skipped."""
        releases = {
            "1.0.0": [{"yanked": False}],
            "not-a-version": [{"yanked": False}],
        }
        assert _find_latest_non_yanked_version(releases) == "1.0.0"

    def test_partially_yanked_files_not_considered_yanked(self) -> None:
        """Test that a version with some non-yanked files is not yanked."""
        releases = {
            "1.0.0": [{"yanked": False}],
            "2.0.0": [{"yanked": True}, {"yanked": False}],
        }
        assert _find_latest_non_yanked_version(releases) == "2.0.0"


class TestIsVersionYanked:
    """Test _is_version_yanked helper."""

    def test_non_yanked_version(self) -> None:
        """Test a non-yanked version returns False."""
        releases = {"1.0.0": [{"yanked": False}]}
        is_yanked, reason = _is_version_yanked("1.0.0", releases)
        assert is_yanked is False
        assert reason == ""

    def test_yanked_version_with_reason(self) -> None:
        """Test a yanked version returns True with reason."""
        releases = {
            "1.0.0": [{"yanked": True, "yanked_reason": "critical bug"}],
        }
        is_yanked, reason = _is_version_yanked("1.0.0", releases)
        assert is_yanked is True
        assert reason == "critical bug"

    def test_yanked_version_without_reason(self) -> None:
        """Test a yanked version returns True with empty reason."""
        releases = {"1.0.0": [{"yanked": True}]}
        is_yanked, reason = _is_version_yanked("1.0.0", releases)
        assert is_yanked is True
        assert reason == ""

    def test_unknown_version(self) -> None:
        """Test an unknown version returns False."""
        releases = {"1.0.0": [{"yanked": False}]}
        is_yanked, reason = _is_version_yanked("9.9.9", releases)
        assert is_yanked is False
        assert reason == ""

    def test_partially_yanked_files(self) -> None:
        """Test a version with mixed yanked/non-yanked files is not yanked."""
        releases = {
            "1.0.0": [{"yanked": True}, {"yanked": False}],
        }
        is_yanked, reason = _is_version_yanked("1.0.0", releases)
        assert is_yanked is False
        assert reason == ""

    def test_multiple_yanked_files_picks_first_reason(self) -> None:
        """Test that the first available reason is returned."""
        releases = {
            "1.0.0": [
                {"yanked": True, "yanked_reason": ""},
                {"yanked": True, "yanked_reason": "second reason"},
            ],
        }
        is_yanked, reason = _is_version_yanked("1.0.0", releases)
        assert is_yanked is True
        assert reason == "second reason"


class TestIsCurrentVersionYanked:
    """Test is_current_version_yanked public function."""

    @patch("crewai_cli.version.get_crewai_version")
    @patch("crewai_cli.version._get_cache_file")
    def test_reads_from_valid_cache(
        self, mock_cache_file: MagicMock, mock_version: MagicMock, tmp_path: Path
    ) -> None:
        """Test reading yanked status from a valid cache."""
        mock_version.return_value = "1.0.0"
        cache_file = tmp_path / "version_cache.json"
        cache_data = {
            "version": "2.0.0",
            "timestamp": datetime.now().isoformat(),
            "current_version": "1.0.0",
            "current_version_yanked": True,
            "current_version_yanked_reason": "bad release",
        }
        cache_file.write_text(json.dumps(cache_data))
        mock_cache_file.return_value = cache_file

        is_yanked, reason = is_current_version_yanked()
        assert is_yanked is True
        assert reason == "bad release"

    @patch("crewai_cli.version.get_crewai_version")
    @patch("crewai_cli.version._get_cache_file")
    def test_not_yanked_from_cache(
        self, mock_cache_file: MagicMock, mock_version: MagicMock, tmp_path: Path
    ) -> None:
        """Test non-yanked status from a valid cache."""
        mock_version.return_value = "2.0.0"
        cache_file = tmp_path / "version_cache.json"
        cache_data = {
            "version": "2.0.0",
            "timestamp": datetime.now().isoformat(),
            "current_version": "2.0.0",
            "current_version_yanked": False,
            "current_version_yanked_reason": "",
        }
        cache_file.write_text(json.dumps(cache_data))
        mock_cache_file.return_value = cache_file

        is_yanked, reason = is_current_version_yanked()
        assert is_yanked is False
        assert reason == ""

    @patch("crewai_cli.version.get_latest_version_from_pypi")
    @patch("crewai_cli.version.get_crewai_version")
    @patch("crewai_cli.version._get_cache_file")
    def test_triggers_fetch_on_stale_cache(
        self,
        mock_cache_file: MagicMock,
        mock_version: MagicMock,
        mock_fetch: MagicMock,
        tmp_path: Path,
    ) -> None:
        """Test that a stale cache triggers a re-fetch."""
        mock_version.return_value = "1.0.0"
        cache_file = tmp_path / "version_cache.json"
        old_time = datetime.now() - timedelta(hours=25)
        cache_data = {
            "version": "2.0.0",
            "timestamp": old_time.isoformat(),
            "current_version": "1.0.0",
            "current_version_yanked": True,
            "current_version_yanked_reason": "old reason",
        }
        cache_file.write_text(json.dumps(cache_data))
        mock_cache_file.return_value = cache_file

        fresh_cache = {
            "version": "2.0.0",
            "timestamp": datetime.now().isoformat(),
            "current_version": "1.0.0",
            "current_version_yanked": False,
            "current_version_yanked_reason": "",
        }

        def write_fresh_cache() -> str:
            cache_file.write_text(json.dumps(fresh_cache))
            return "2.0.0"

        mock_fetch.side_effect = lambda: write_fresh_cache()

        is_yanked, reason = is_current_version_yanked()
        assert is_yanked is False
        mock_fetch.assert_called_once()

    @patch("crewai_cli.version.get_latest_version_from_pypi")
    @patch("crewai_cli.version.get_crewai_version")
    @patch("crewai_cli.version._get_cache_file")
    def test_returns_false_on_fetch_failure(
        self,
        mock_cache_file: MagicMock,
        mock_version: MagicMock,
        mock_fetch: MagicMock,
        tmp_path: Path,
    ) -> None:
        """Test that fetch failure returns not yanked."""
        mock_version.return_value = "1.0.0"
        cache_file = tmp_path / "version_cache.json"
        mock_cache_file.return_value = cache_file
        mock_fetch.return_value = None

        is_yanked, reason = is_current_version_yanked()
        assert is_yanked is False
        assert reason == ""



# TestConsoleFormatterVersionCheck tests remain in lib/crewai/tests/cli/test_version.py
# as they depend on crewai.events.utils.console_formatter (core package).