From f249356e1674da786a91f8ade36f6e62b255b05b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 30 Apr 2026 17:47:18 +0000 Subject: [PATCH] feat(proxy): proactive model deprecation alerts and /model/deprecations endpoint Surfaces deprecation_date metadata that is already shipped in model_prices_and_context_window.json so operators get lead time to migrate before a provider sunsets a model. - New helper litellm.proxy.common_utils.model_deprecation classifies the router's configured models into deprecated / imminent / upcoming buckets. Resolution order: explicit model_info.deprecation_date > model_info.base_model > litellm_params.model. - New GET /model/deprecations (and /v1/model/deprecations) endpoint returns a ModelDeprecationResponse, gated by user_api_key_auth. - New AlertType.model_deprecation_warnings (in DEFAULT_ALERT_TYPES) plus SlackAlerting.send_model_deprecation_alert dispatches a Slack message for deprecated/imminent models. Severity is High when any model is already past its date, Medium when only imminent. - ProxyLogging.startup_event schedules a daily background task (_run_scheduled_deprecation_check) when the alert type is enabled. The interval is configurable via LITELLM_MODEL_DEPRECATION_CHECK_INTERVAL and the warn window via LITELLM_MODEL_DEPRECATION_WARN_DAYS. - Tests: 16 unit tests for the helper plus 4 for the Slack hook in tests/test_litellm/. Co-authored-by: Mateo Wang --- .../SlackAlerting/slack_alerting.py | 75 +++++ .../proxy/common_utils/model_deprecation.py | 247 +++++++++++++++ litellm/proxy/proxy_server.py | 51 ++++ litellm/proxy/utils.py | 14 + litellm/types/integrations/slack_alerting.py | 2 + litellm/types/proxy/model_deprecation.py | 93 ++++++ .../test_model_deprecation_alert.py | 100 +++++++ .../common_utils/test_model_deprecation.py | 280 ++++++++++++++++++ 8 files changed, 862 insertions(+) create mode 100644 litellm/proxy/common_utils/model_deprecation.py create mode 100644 litellm/types/proxy/model_deprecation.py create mode 100644 tests/test_litellm/integrations/SlackAlerting/test_model_deprecation_alert.py create mode 100644 tests/test_litellm/proxy/common_utils/test_model_deprecation.py diff --git a/litellm/integrations/SlackAlerting/slack_alerting.py b/litellm/integrations/SlackAlerting/slack_alerting.py index 771d7876fea..12b5d7525dc 100644 --- a/litellm/integrations/SlackAlerting/slack_alerting.py +++ b/litellm/integrations/SlackAlerting/slack_alerting.py @@ -1038,6 +1038,81 @@ Model Info: async def model_removed_alert(self, model_name: str): pass + async def send_model_deprecation_alert( + self, llm_router: Optional[Any] = None + ) -> bool: + """Aggregate deprecation metadata for the configured models and alert. + + Returns ``True`` when an alert payload was dispatched, ``False`` + otherwise. The ``send_alert`` helper itself is responsible for honoring + the user's webhook configuration; this method only owns producing the + message and choosing whether to send it. + """ + if ( + self.alerting is None + or AlertType.model_deprecation_warnings not in self.alert_types + ): + return False + + from litellm.proxy.common_utils.model_deprecation import ( + collect_model_deprecations, + format_deprecation_alert_message, + ) + + try: + snapshot = collect_model_deprecations(llm_router=llm_router) + except Exception as e: + verbose_proxy_logger.exception( + "Error collecting model deprecation snapshot: %s", e + ) + return False + + message = format_deprecation_alert_message(snapshot) + if message is None: + return False + + level: Literal["Low", "Medium", "High"] = ( + "High" if snapshot.deprecated else "Medium" + ) + + await self.send_alert( + message=message, + level=level, + alert_type=AlertType.model_deprecation_warnings, + alerting_metadata={ + "deprecated_count": len(snapshot.deprecated), + "imminent_count": len(snapshot.imminent), + "upcoming_count": len(snapshot.upcoming), + }, + ) + return True + + async def _run_scheduled_deprecation_check(self, llm_router: Optional[Any] = None): + """Periodic background task that emits a model deprecation alert. + + Runs immediately on startup (so operators see the current state in + Slack) and then sleeps ``DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS`` + between runs. Exits silently if the alert type is not enabled. + """ + from litellm.types.proxy.model_deprecation import ( + DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS, + ) + + if ( + self.alerting is None + or AlertType.model_deprecation_warnings not in self.alert_types + ): + return + + while True: + try: + await self.send_model_deprecation_alert(llm_router=llm_router) + except Exception as e: + verbose_proxy_logger.exception( + "Error in model deprecation alert loop: %s", e + ) + await asyncio.sleep(DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS) + async def send_webhook_alert(self, webhook_event: WebhookEvent) -> bool: """ Sends structured alert to webhook, if set. diff --git a/litellm/proxy/common_utils/model_deprecation.py b/litellm/proxy/common_utils/model_deprecation.py new file mode 100644 index 00000000000..1b11fa5abd7 --- /dev/null +++ b/litellm/proxy/common_utils/model_deprecation.py @@ -0,0 +1,247 @@ +"""Helpers for surfacing model deprecation/sunset information. + +This module reads ``deprecation_date`` metadata that is bundled in +``model_prices_and_context_window.json`` (exposed at runtime via +``litellm.model_cost``) and classifies the proxy's configured models into +``upcoming``, ``imminent`` and ``deprecated`` buckets. It is the single +source of truth used by both the ``/model/deprecations`` endpoint and the +proactive Slack alert. + +Resolution order for a deployment's deprecation date: + +1. ``model_info.deprecation_date`` – an explicit override on the deployment. +2. ``model_info.base_model`` looked up in ``litellm.model_cost``. +3. The ``litellm_params.model`` string looked up in ``litellm.model_cost``. + +Models without any deprecation metadata are skipped silently (most models +are not deprecated, and we don't want to pollute the response). +""" + +from __future__ import annotations + +from datetime import date, datetime, timezone +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple + +import litellm +from litellm._logging import verbose_logger +from litellm.types.proxy.model_deprecation import ( + DEFAULT_DEPRECATION_WARN_DAYS, + ModelDeprecationInfo, + ModelDeprecationResponse, +) + +if TYPE_CHECKING: + from litellm.router import Router as _Router + + Router = _Router +else: + Router = Any + + +def _parse_deprecation_date(raw_value: Any) -> Optional[date]: + """Parse a ``deprecation_date`` string in YYYY-MM-DD form. + + Returns ``None`` for missing, malformed, or sentinel placeholder values + (the JSON map ships a documentation sentinel of the form ``"date when..."``). + """ + if raw_value is None: + return None + if isinstance(raw_value, date): + return raw_value + if not isinstance(raw_value, str): + return None + try: + return datetime.strptime(raw_value.strip(), "%Y-%m-%d").date() + except ValueError: + return None + + +def _lookup_deprecation_date_from_cost_map( + model_key: Optional[str], +) -> Tuple[Optional[date], Optional[str]]: + """Look up a deprecation date in ``litellm.model_cost`` for ``model_key``. + + Returns a tuple of (deprecation_date, litellm_provider). + """ + if not model_key: + return None, None + entry = litellm.model_cost.get(model_key) + if not isinstance(entry, dict): + return None, None + return ( + _parse_deprecation_date(entry.get("deprecation_date")), + entry.get("litellm_provider"), + ) + + +def _resolve_deployment_deprecation( + deployment: Dict[str, Any], +) -> Tuple[Optional[date], Optional[str], Optional[str]]: + """Resolve a deployment's deprecation metadata. + + Returns a tuple of (deprecation_date, litellm_model, litellm_provider). + """ + model_info = deployment.get("model_info") or {} + explicit = _parse_deprecation_date(model_info.get("deprecation_date")) + if explicit is not None: + litellm_params = deployment.get("litellm_params") or {} + return ( + explicit, + litellm_params.get("model"), + model_info.get("litellm_provider"), + ) + + base_model = model_info.get("base_model") + dep_date, provider = _lookup_deprecation_date_from_cost_map(base_model) + if dep_date is not None: + return dep_date, base_model, provider + + litellm_params = deployment.get("litellm_params") or {} + raw_model = litellm_params.get("model") + dep_date, provider = _lookup_deprecation_date_from_cost_map(raw_model) + if dep_date is not None: + return dep_date, raw_model, provider + + if isinstance(raw_model, str) and "/" in raw_model: + # Try the un-prefixed lookup (e.g. "openai/gpt-4o" → "gpt-4o"). + bare = raw_model.split("/", 1)[1] + dep_date, provider = _lookup_deprecation_date_from_cost_map(bare) + if dep_date is not None: + return dep_date, bare, provider + + return None, raw_model, model_info.get("litellm_provider") + + +def _classify(days_until: int, warn_within_days: int) -> str: + if days_until < 0: + return "deprecated" + if days_until <= warn_within_days: + return "imminent" + return "upcoming" + + +def _model_dump_compat(deployment: Any) -> Dict[str, Any]: + """Return a plain dict for both pydantic models and dicts.""" + if isinstance(deployment, dict): + return deployment + if hasattr(deployment, "model_dump"): + return deployment.model_dump(exclude_none=True) + if hasattr(deployment, "dict"): + return deployment.dict() + return dict(deployment) + + +def collect_model_deprecations( + llm_router: Optional[Router], + warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS, + today: Optional[date] = None, +) -> ModelDeprecationResponse: + """Aggregate deprecation info for all deployments configured on the router. + + De-duplicates by ``(model_name, deprecation_date)`` so multi-deployment + model groups (load-balanced across regions) only surface once per + deprecation date. + """ + snapshot_time = datetime.now(timezone.utc) + today = today or snapshot_time.date() + + response = ModelDeprecationResponse( + warn_within_days=warn_within_days, + checked_at=snapshot_time, + ) + + if llm_router is None: + return response + + seen: set = set() + deployments = llm_router.get_model_list() or [] + for deployment in deployments: + deployment_dict = _model_dump_compat(deployment) + model_name = deployment_dict.get("model_name") + if not model_name: + continue + + dep_date, litellm_model, provider = _resolve_deployment_deprecation( + deployment_dict + ) + if dep_date is None: + continue + + dedup_key = (model_name, dep_date.isoformat()) + if dedup_key in seen: + continue + seen.add(dedup_key) + + days_until = (dep_date - today).days + status = _classify(days_until, warn_within_days) + + info = ModelDeprecationInfo( + model_name=model_name, + litellm_model=litellm_model, + deprecation_date=dep_date, + days_until_deprecation=days_until, + status=status, + litellm_provider=provider, + ) + + if status == "deprecated": + response.deprecated.append(info) + elif status == "imminent": + response.imminent.append(info) + else: + response.upcoming.append(info) + + response.deprecated.sort(key=lambda m: m.deprecation_date) + response.imminent.sort(key=lambda m: m.deprecation_date) + response.upcoming.sort(key=lambda m: m.deprecation_date) + + verbose_logger.debug( + "model_deprecation: deprecated=%d imminent=%d upcoming=%d", + len(response.deprecated), + len(response.imminent), + len(response.upcoming), + ) + + return response + + +def format_deprecation_alert_message( + snapshot: ModelDeprecationResponse, +) -> Optional[str]: + """Format a Slack-friendly alert message for the warning buckets. + + Only ``deprecated`` and ``imminent`` models are included; ``upcoming`` + models are intentionally omitted to avoid alert fatigue. Returns + ``None`` when there is nothing to alert on. + """ + if not snapshot.deprecated and not snapshot.imminent: + return None + + lines: List[str] = ["*⚠️ Model Deprecation Warning*"] + + def _format_entry(info: ModelDeprecationInfo) -> str: + suffix = ( + f"already deprecated {abs(info.days_until_deprecation)}d ago" + if info.days_until_deprecation < 0 + else f"in {info.days_until_deprecation}d" + ) + return ( + f"• `{info.model_name}` " + f"(provider: {info.litellm_provider or 'unknown'}, " + f"deprecates {info.deprecation_date.isoformat()} – {suffix})" + ) + + if snapshot.deprecated: + lines.append("\n*Already deprecated:*") + lines.extend(_format_entry(i) for i in snapshot.deprecated) + + if snapshot.imminent: + lines.append(f"\n*Deprecating within {snapshot.warn_within_days} days:*") + lines.extend(_format_entry(i) for i in snapshot.imminent) + + lines.append( + "\nPlan migrations to a supported model. See " + "https://docs.litellm.ai/docs/proxy/model_management for guidance." + ) + + return "\n".join(lines) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index bc980934f9f..3d137732075 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -319,6 +319,7 @@ from litellm.proxy.common_utils.load_config_utils import ( get_config_file_contents_from_gcs, get_file_contents_from_s3, ) +from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations from litellm.proxy.common_utils.model_listing_utils import TeamModelNameTranslator from litellm.proxy.common_utils.openai_endpoint_utils import ( remove_sensitive_info_from_deployment, @@ -624,6 +625,10 @@ from litellm.types.proxy.control_plane_endpoints import WorkerRegistryEntry from litellm.types.proxy.management_endpoints.model_management_endpoints import ( ModelGroupInfoProxy, ) +from litellm.types.proxy.model_deprecation import ( + DEFAULT_DEPRECATION_WARN_DAYS, + ModelDeprecationResponse, +) from litellm.types.proxy.management_endpoints.ui_sso import ( DefaultTeamSSOParams, LiteLLM_UpperboundKeyGenerateParams, @@ -13436,6 +13441,52 @@ async def model_info_v1( return {"data": all_models} +@router.get( + "/model/deprecations", + tags=["model management"], + dependencies=[Depends(user_api_key_auth)], + response_model=ModelDeprecationResponse, +) +@router.get( + "/v1/model/deprecations", + tags=["model management"], + dependencies=[Depends(user_api_key_auth)], + response_model=ModelDeprecationResponse, +) +async def model_deprecations( + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), + warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS, +) -> ModelDeprecationResponse: + """List models with known deprecation/sunset dates, bucketed by urgency. + + Reads `deprecation_date` metadata from `model_prices_and_context_window.json` + (and any per-deployment `model_info.deprecation_date` overrides) for the + models configured on this proxy. + + Parameters: + warn_within_days: Window (in days) used to bucket "imminent" models. + Defaults to `LITELLM_MODEL_DEPRECATION_WARN_DAYS` env var (or 30). + + Returns: + A payload with three lists of `ModelDeprecationInfo` entries: + + - `deprecated`: deprecation date is in the past — these requests may + fail at any time. + - `imminent`: deprecation date is within `warn_within_days` from today. + - `upcoming`: deprecation date is further out. + + Example: + ```shell + curl -X GET 'http://localhost:4000/model/deprecations' \\ + -H 'Authorization: Bearer sk-1234' + ``` + """ + global llm_router + return collect_model_deprecations( + llm_router=llm_router, warn_within_days=warn_within_days + ) + + def _get_model_group_info( llm_router: Router, all_models_str: list[str], model_group: str | None ) -> list[ModelGroupInfoProxy]: diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index dd0c57aa911..f98df15a346 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -442,6 +442,7 @@ class ProxyLogging: # Guard flags to prevent duplicate background tasks self.daily_report_started: bool = False self.hanging_requests_check_started: bool = False + self.deprecation_check_started: bool = False def startup_event( self, @@ -481,6 +482,19 @@ class ProxyLogging: ) # RUN HANGING REQUEST CHECK (if user wants to alert on hanging requests) self.hanging_requests_check_started = True + if ( + self.slack_alerting_instance is not None + and AlertType.model_deprecation_warnings + in self.slack_alerting_instance.alert_types + and not self.deprecation_check_started + ): + asyncio.create_task( + self.slack_alerting_instance._run_scheduled_deprecation_check( + llm_router=llm_router + ) + ) # RUN MODEL DEPRECATION ALERT LOOP (if scheduled) + self.deprecation_check_started = True + def update_values( self, alerting: list | None = None, diff --git a/litellm/types/integrations/slack_alerting.py b/litellm/types/integrations/slack_alerting.py index 56616c00aa0..768b5d35597 100644 --- a/litellm/types/integrations/slack_alerting.py +++ b/litellm/types/integrations/slack_alerting.py @@ -147,6 +147,7 @@ class AlertType(str, Enum): # Deployment alerts cooldown_deployment = "cooldown_deployment" new_model_added = "new_model_added" + model_deprecation_warnings = "model_deprecation_warnings" # Outage alerts outage_alerts = "outage_alerts" @@ -187,6 +188,7 @@ DEFAULT_ALERT_TYPES: Final[list[AlertType]] = [ # Deployment alerts AlertType.cooldown_deployment, AlertType.new_model_added, + AlertType.model_deprecation_warnings, # Outage alerts AlertType.outage_alerts, AlertType.region_outage_alerts, diff --git a/litellm/types/proxy/model_deprecation.py b/litellm/types/proxy/model_deprecation.py new file mode 100644 index 00000000000..72ccb48fbfc --- /dev/null +++ b/litellm/types/proxy/model_deprecation.py @@ -0,0 +1,93 @@ +"""Type definitions for model deprecation tracking and proactive alerts. + +The proxy reads deprecation/sunset metadata from +``litellm.model_cost`` (sourced from ``model_prices_and_context_window.json``) +and surfaces it through the ``/model/deprecations`` endpoint and Slack +alerting. These types describe the response payload and the alert payload. +""" + +from __future__ import annotations + +import os +from datetime import date, datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + + +DEFAULT_DEPRECATION_WARN_DAYS = int( + os.getenv("LITELLM_MODEL_DEPRECATION_WARN_DAYS", "30") +) +"""Number of days before the deprecation date to start raising warnings. + +Configurable via the ``LITELLM_MODEL_DEPRECATION_WARN_DAYS`` environment +variable. Defaults to 30 days, matching the typical migration window most +LLM providers offer between announcement and removal. +""" + +DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS = int( + os.getenv("LITELLM_MODEL_DEPRECATION_CHECK_INTERVAL", str(24 * 60 * 60)) +) +"""How often the periodic background check runs. Defaults to once per day.""" + + +DeprecationStatusLiteral = str +"""One of ``"upcoming"``, ``"imminent"``, ``"deprecated"``. + +* ``upcoming`` – deprecation is scheduled but more than the warn window away. +* ``imminent`` – deprecation date is within ``warn_within_days`` from today. +* ``deprecated`` – deprecation date has already passed. +""" + + +class ModelDeprecationInfo(BaseModel): + """Per-model deprecation metadata returned by ``/model/deprecations``.""" + + model_name: str = Field( + description="The public name of the model on the proxy (model_group)." + ) + litellm_model: Optional[str] = Field( + default=None, + description="The underlying litellm model string the deprecation date is sourced from.", + ) + deprecation_date: date = Field( + description="The date (UTC) when the model becomes deprecated." + ) + days_until_deprecation: int = Field( + description=( + "Days remaining until the deprecation date. Negative if the model " + "is already deprecated." + ), + ) + status: DeprecationStatusLiteral = Field( + description="One of 'upcoming', 'imminent', or 'deprecated'.", + ) + litellm_provider: Optional[str] = Field( + default=None, description="The provider this model belongs to." + ) + + +class ModelDeprecationResponse(BaseModel): + """Response payload for ``GET /model/deprecations``.""" + + deprecated: List[ModelDeprecationInfo] = Field( + default_factory=list, + description="Models whose deprecation date has already passed.", + ) + imminent: List[ModelDeprecationInfo] = Field( + default_factory=list, + description=( + "Models whose deprecation date is within ``warn_within_days`` from " + "today and require immediate migration planning." + ), + ) + upcoming: List[ModelDeprecationInfo] = Field( + default_factory=list, + description="Models with a future deprecation date outside the warn window.", + ) + warn_within_days: int = Field( + description="The window (in days) used to bucket 'imminent' models." + ) + checked_at: datetime = Field( + description="UTC timestamp when the deprecation snapshot was generated." + ) diff --git a/tests/test_litellm/integrations/SlackAlerting/test_model_deprecation_alert.py b/tests/test_litellm/integrations/SlackAlerting/test_model_deprecation_alert.py new file mode 100644 index 00000000000..1cf6bbe0354 --- /dev/null +++ b/tests/test_litellm/integrations/SlackAlerting/test_model_deprecation_alert.py @@ -0,0 +1,100 @@ +"""Tests for the Slack alerting model deprecation hook.""" + +import os +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.abspath("../../../..")) + +import litellm +from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting +from litellm.proxy._types import AlertType + + +def _make_router(deployments): + router = MagicMock() + router.get_model_list.return_value = deployments + return router + + +@pytest.mark.asyncio +async def test_should_skip_when_alert_type_disabled(): + alerting = SlackAlerting( + alerting=["slack"], + alert_types=[AlertType.llm_exceptions], + ) + sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock()) + assert sent is False + + +@pytest.mark.asyncio +async def test_should_skip_when_no_alerting_configured(): + alerting = SlackAlerting( + alerting=None, + alert_types=[AlertType.model_deprecation_warnings], + ) + sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock()) + assert sent is False + + +@pytest.mark.asyncio +async def test_should_skip_when_no_deprecations_found(monkeypatch): + monkeypatch.setattr(litellm, "model_cost", {}) + alerting = SlackAlerting( + alerting=["slack"], + alert_types=[AlertType.model_deprecation_warnings], + ) + router = _make_router( + [ + { + "model_name": "fresh", + "litellm_params": {"model": "openai/gpt-4o"}, + "model_info": {"id": "x"}, + } + ] + ) + sent = await alerting.send_model_deprecation_alert(llm_router=router) + assert sent is False + + +@pytest.mark.asyncio +async def test_should_dispatch_high_severity_when_deprecated(monkeypatch): + monkeypatch.setattr( + litellm, + "model_cost", + { + "dead-model": { + "deprecation_date": "2020-01-01", + "litellm_provider": "openai", + } + }, + ) + alerting = SlackAlerting( + alerting=["slack"], + alert_types=[AlertType.model_deprecation_warnings], + ) + router = _make_router( + [ + { + "model_name": "dead-alias", + "litellm_params": {"model": "dead-model"}, + "model_info": {"id": "1"}, + } + ] + ) + + with patch.object( + alerting, "send_alert", new_callable=AsyncMock + ) as mock_send_alert: + sent = await alerting.send_model_deprecation_alert(llm_router=router) + + assert sent is True + mock_send_alert.assert_awaited_once() + call_kwargs = mock_send_alert.await_args.kwargs + assert call_kwargs["alert_type"] == AlertType.model_deprecation_warnings + assert call_kwargs["level"] == "High" + assert call_kwargs["alerting_metadata"]["deprecated_count"] == 1 + assert call_kwargs["alerting_metadata"]["imminent_count"] == 0 + assert "dead-alias" in call_kwargs["message"] diff --git a/tests/test_litellm/proxy/common_utils/test_model_deprecation.py b/tests/test_litellm/proxy/common_utils/test_model_deprecation.py new file mode 100644 index 00000000000..c873e2494f9 --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_model_deprecation.py @@ -0,0 +1,280 @@ +"""Tests for the model deprecation helper module. + +These tests focus on the helper itself — not on the proxy endpoint or +Slack integration — so they can run without the full proxy stack. +""" + +import os +import sys +from datetime import date +from unittest.mock import MagicMock + + +sys.path.insert(0, os.path.abspath("../../../..")) + +import litellm +from litellm.proxy.common_utils.model_deprecation import ( + _classify, + _parse_deprecation_date, + collect_model_deprecations, + format_deprecation_alert_message, +) + + +def _make_router(deployments): + router = MagicMock() + router.get_model_list.return_value = deployments + return router + + +class TestParseDeprecationDate: + def test_should_parse_iso_string(self): + assert _parse_deprecation_date("2026-12-31") == date(2026, 12, 31) + + def test_should_pass_through_date_object(self): + d = date(2026, 1, 1) + assert _parse_deprecation_date(d) == d + + def test_should_return_none_for_documentation_sentinel(self): + # The JSON map ships a sentinel string under the "sample_spec" key. + assert ( + _parse_deprecation_date( + "date when the model becomes deprecated in the format YYYY-MM-DD" + ) + is None + ) + + def test_should_return_none_for_none(self): + assert _parse_deprecation_date(None) is None + + def test_should_return_none_for_unsupported_type(self): + assert _parse_deprecation_date(12345) is None + + +class TestClassify: + def test_should_classify_past_dates_as_deprecated(self): + assert _classify(-1, warn_within_days=30) == "deprecated" + assert _classify(-365, warn_within_days=30) == "deprecated" + + def test_should_classify_inside_window_as_imminent(self): + assert _classify(0, warn_within_days=30) == "imminent" + assert _classify(15, warn_within_days=30) == "imminent" + assert _classify(30, warn_within_days=30) == "imminent" + + def test_should_classify_outside_window_as_upcoming(self): + assert _classify(31, warn_within_days=30) == "upcoming" + assert _classify(365, warn_within_days=30) == "upcoming" + + +class TestCollectModelDeprecations: + def test_should_return_empty_response_when_router_is_none(self): + snapshot = collect_model_deprecations(llm_router=None) + assert snapshot.deprecated == [] + assert snapshot.imminent == [] + assert snapshot.upcoming == [] + + def test_should_skip_models_without_deprecation_metadata(self, monkeypatch): + monkeypatch.setattr(litellm, "model_cost", {}) + router = _make_router( + [ + { + "model_name": "gpt-4o", + "litellm_params": {"model": "openai/gpt-4o"}, + "model_info": {"id": "abc"}, + } + ] + ) + snapshot = collect_model_deprecations(llm_router=router) + assert snapshot.deprecated == [] + assert snapshot.imminent == [] + assert snapshot.upcoming == [] + + def test_should_classify_into_three_buckets(self, monkeypatch): + today = date(2026, 6, 1) + monkeypatch.setattr( + litellm, + "model_cost", + { + "deprecated-model": { + "deprecation_date": "2026-01-01", + "litellm_provider": "openai", + }, + "imminent-model": { + "deprecation_date": "2026-06-15", + "litellm_provider": "openai", + }, + "upcoming-model": { + "deprecation_date": "2027-01-01", + "litellm_provider": "openai", + }, + }, + ) + router = _make_router( + [ + { + "model_name": "deprecated-alias", + "litellm_params": {"model": "openai/deprecated-model"}, + "model_info": {"id": "1"}, + }, + { + "model_name": "imminent-alias", + "litellm_params": {"model": "imminent-model"}, + "model_info": {"id": "2"}, + }, + { + "model_name": "upcoming-alias", + "litellm_params": {"model": "openai/upcoming-model"}, + "model_info": {"id": "3"}, + }, + ] + ) + + snapshot = collect_model_deprecations( + llm_router=router, warn_within_days=30, today=today + ) + + assert [m.model_name for m in snapshot.deprecated] == ["deprecated-alias"] + assert [m.model_name for m in snapshot.imminent] == ["imminent-alias"] + assert [m.model_name for m in snapshot.upcoming] == ["upcoming-alias"] + + assert snapshot.deprecated[0].days_until_deprecation < 0 + assert snapshot.imminent[0].days_until_deprecation == 14 + assert snapshot.upcoming[0].days_until_deprecation > 30 + + def test_should_prefer_explicit_deployment_override(self, monkeypatch): + today = date(2026, 6, 1) + monkeypatch.setattr( + litellm, + "model_cost", + {"some-model": {"deprecation_date": "2030-01-01"}}, + ) + router = _make_router( + [ + { + "model_name": "my-alias", + "litellm_params": {"model": "some-model"}, + "model_info": { + "id": "x", + "deprecation_date": "2026-06-10", + }, + } + ] + ) + + snapshot = collect_model_deprecations( + llm_router=router, warn_within_days=30, today=today + ) + + assert len(snapshot.imminent) == 1 + assert snapshot.imminent[0].deprecation_date == date(2026, 6, 10) + + def test_should_dedupe_duplicate_deployments_in_same_group(self, monkeypatch): + today = date(2026, 6, 1) + monkeypatch.setattr( + litellm, + "model_cost", + {"shared-model": {"deprecation_date": "2026-06-10"}}, + ) + router = _make_router( + [ + { + "model_name": "alias", + "litellm_params": {"model": "shared-model"}, + "model_info": {"id": "1"}, + }, + { + "model_name": "alias", + "litellm_params": {"model": "shared-model"}, + "model_info": {"id": "2"}, + }, + ] + ) + + snapshot = collect_model_deprecations( + llm_router=router, warn_within_days=30, today=today + ) + + assert len(snapshot.imminent) == 1 + + def test_should_resolve_via_base_model(self, monkeypatch): + today = date(2026, 6, 1) + monkeypatch.setattr( + litellm, + "model_cost", + {"base-thing": {"deprecation_date": "2026-06-10"}}, + ) + router = _make_router( + [ + { + "model_name": "alias", + "litellm_params": {"model": "azure/some-deployment-name"}, + "model_info": {"id": "1", "base_model": "base-thing"}, + } + ] + ) + + snapshot = collect_model_deprecations( + llm_router=router, warn_within_days=30, today=today + ) + + assert len(snapshot.imminent) == 1 + assert snapshot.imminent[0].litellm_model == "base-thing" + + +class TestFormatDeprecationAlertMessage: + def test_should_return_none_when_nothing_to_alert(self): + snapshot = collect_model_deprecations(llm_router=None) + assert format_deprecation_alert_message(snapshot) is None + + def test_should_render_imminent_and_deprecated_sections(self, monkeypatch): + today = date(2026, 6, 1) + monkeypatch.setattr( + litellm, + "model_cost", + { + "dead-model": { + "deprecation_date": "2026-01-01", + "litellm_provider": "openai", + }, + "soon-model": { + "deprecation_date": "2026-06-15", + "litellm_provider": "anthropic", + }, + "later-model": { + "deprecation_date": "2027-01-01", + "litellm_provider": "anthropic", + }, + }, + ) + router = _make_router( + [ + { + "model_name": "dead", + "litellm_params": {"model": "dead-model"}, + "model_info": {"id": "1"}, + }, + { + "model_name": "soon", + "litellm_params": {"model": "soon-model"}, + "model_info": {"id": "2"}, + }, + { + "model_name": "later", + "litellm_params": {"model": "later-model"}, + "model_info": {"id": "3"}, + }, + ] + ) + + snapshot = collect_model_deprecations( + llm_router=router, warn_within_days=30, today=today + ) + message = format_deprecation_alert_message(snapshot) + + assert message is not None + assert "Already deprecated" in message + assert "Deprecating within 30 days" in message + assert "`dead`" in message + assert "`soon`" in message + # Upcoming models must NOT be in the alert (avoid alert fatigue). + assert "`later`" not in message