litellm/tests/litellm_utils_tests/base_token_counter_test.py
devin-ai-integration[bot] 04d97abffb
test: make 77 legacy live tests offline in litellm_utils, router_unit and responses dirs (#45298)
* test: make 77 legacy live tests offline in litellm_utils, router_unit and responses dirs

* test: restore lost coverage in litellm_utils offline replacements

Bound live health-check tasks to the concurrency limit so eager task
creation behind a semaphore fails, route the Langfuse trace-id check
through completion -> Logging.get_trace_id with all four metadata
combinations, run the callback dedup checks through acompletion, pass
the dynamic key via litellm_params, cover the env-inferred default
model list, and add the per-provider audio transcription config lookup

* test: restore router coverage lost in the offline move

Add the missing non-stream LIT-3058 header/count-once test, pin the UTC minute on every
usage-counter test, assert router-level client reuse for transcription, and replace the
private selector-attribute checks with routing outcomes per strategy. Assert outbound bodies
for speech, rerank, image, assistants and moderation, add timeouts to event waits, and drop
the router_unit_tests husk files that no longer hold tests

* test: restore sync stream, router sync stream, error event and field-type coverage for migrated responses tests

Add the sync streaming logging and sync Router.responses streaming cases the
offline replacements dropped, port the legacy per-event and response field-type
validation, assert raw headers, the float created_at conversion, MCP tool headers,
the search_context_size input, and add an offline replacement for the in-stream
context-window error event. Remove helpers left dead in the legacy file.

* test: run numpydoc-backed unit tests in GHA and drop a test that pinned a crash

* test: restore sequence number, item id and content part checks in the responses stream validator

* test: match router logging events to the test's own deployment so late events from other tests are ignored

* test: restore the s3 cold storage history test to its original assertions

---------

Co-authored-by: yuneng <yuneng@berri.ai>
2026-10-08 02:37:55 -07:00

72 lines
2.3 KiB
Python

"""
Base Token Counter Test Suite.
This module provides an abstract base test class that enforces common tests
across all token counter implementations. Similar to base_llm_unit_tests.py
for LLM chat tests.
Usage:
Create a test class that inherits from BaseTokenCounterTest and implement
the abstract methods to provide provider-specific configuration.
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
import pytest
from litellm.llms.base_llm.base_utils import BaseTokenCounter
class BaseTokenCounterTest(ABC):
"""
Abstract base test class for token counter implementations.
Subclasses must implement:
- get_token_counter(): Returns the token counter instance
- get_test_model(): Returns the model name to use for testing
- get_test_messages(): Returns test messages for token counting
- get_deployment_config(): Returns deployment configuration with credentials
- get_custom_llm_provider(): Returns the provider name for should_use_token_counting_api
"""
@abstractmethod
def get_token_counter(self) -> BaseTokenCounter:
"""Must return the token counter instance to test."""
pass
@abstractmethod
def get_test_model(self) -> str:
"""Must return the model name to use for testing."""
pass
@abstractmethod
def get_test_messages(self) -> List[Dict[str, Any]]:
"""Must return test messages for token counting."""
pass
@abstractmethod
def get_deployment_config(self) -> Dict[str, Any]:
"""Must return deployment configuration with credentials."""
pass
@abstractmethod
def get_custom_llm_provider(self) -> str:
"""Must return the provider name for should_use_token_counting_api check."""
pass
@pytest.fixture(autouse=True)
def _handle_missing_credentials(self):
"""Fixture to skip tests when credentials are missing."""
try:
yield
except Exception as e:
error_str = str(e).lower()
if (
"api key" in error_str
or "api_key" in error_str
or "unauthorized" in error_str
):
pytest.skip(f"Missing or invalid credentials: {e}")
raise